Spaces:
Running
Running
File size: 5,871 Bytes
b196778 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 | """Browser transport only; session_service remains the registered-session authority."""
from __future__ import annotations
import hashlib
import hmac
import os
import time
from urllib.parse import urlsplit
from starlette.requests import Request
from starlette.responses import JSONResponse, Response
from sinetti_services import session_token
from sinetti_api.runtime_security import local_loopback_request
_SAFE = frozenset({"GET", "HEAD", "OPTIONS"})
_PREAUTH = frozenset({
"/api/auth/login", "/api/auth/email-code/verify", "/api/auth/email-code/resend",
"/api/auth/oidc/complete", "/api/auth/password-reset/request",
"/api/auth/password-reset/confirm", "/api/auth/browser-session",
})
def secure_cookies(request: Request) -> bool:
return request.url.scheme != "http" or not local_loopback_request(request)
def cookie_names(request: Request) -> tuple[str, str]:
prefix = "__Host-" if secure_cookies(request) else ""
return prefix + "sinetti_session", prefix + "sinetti_csrf"
def request_token(request: Request, authorization: str | None = None) -> str:
# Header presence, including an empty/malformed header, has absolute precedence.
if "authorization" in request.headers or authorization is not None:
value = request.headers.get("authorization", "") if authorization is None else authorization
return value[7:].strip() if value.lower().startswith("bearer ") else ""
return request.cookies.get(cookie_names(request)[0], "")
def _csrf(token: str) -> str:
secret = os.environ.get("SINETTI_SESSION_SECRET", "")
if not secret:
raise RuntimeError("Sinetti session secret is not configured")
return hmac.new(secret.encode(), b"browser-csrf-v1\0" + token.encode(), hashlib.sha256).hexdigest()
def set_session_cookies(response: Response, request: Request, token: str, *, now: float | None = None) -> None:
checked_at = time.time() if now is None else now
claims = session_token.verify(token, now=checked_at)
if claims is None:
raise RuntimeError("Cannot install an invalid session")
remaining = max(0, int(claims["exp"] - checked_at))
session_name, csrf_name = cookie_names(request)
options = dict(max_age=remaining, path="/", secure=secure_cookies(request), samesite="lax")
response.set_cookie(session_name, token, httponly=True, **options)
response.set_cookie(csrf_name, _csrf(token), httponly=False, **options)
response.headers["Cache-Control"] = "no-store"
def clear_session_cookies(response: Response, request: Request) -> None:
for name in cookie_names(request):
response.delete_cookie(name, path="/", secure=secure_cookies(request), httponly=name.endswith("session"), samesite="lax")
response.headers["Cache-Control"] = "no-store"
def _origin(value: str) -> tuple[str, str, int] | None:
try:
parsed = urlsplit(value)
if parsed.scheme not in {"https", "http"} or not parsed.hostname or parsed.username or parsed.password:
return None
if parsed.path not in {"", "/"} or parsed.query or parsed.fragment:
return None
return parsed.scheme, parsed.hostname.lower(), parsed.port or (443 if parsed.scheme == "https" else 80)
except ValueError:
return None
def _foreign_origin(request: Request) -> bool:
fetch_site = request.headers.get("sec-fetch-site", "").lower()
if fetch_site not in {"", "same-origin", "none"}:
return True
supplied = request.headers.get("origin")
if supplied is None:
return False # CLI clients have no Origin; cookie writes still require signed CSRF.
actual = _origin(supplied)
configured = os.environ.get("SINETTI_PUBLIC_BASE_URL", "").strip().rstrip("/")
if not secure_cookies(request):
# A real loopback desk may inherit the production URL used in outbound mail.
configured = ""
expected = _origin(configured) if configured else _origin(str(request.base_url).rstrip("/"))
if configured and secure_cookies(request) and expected and expected[0] != "https":
return True
if not configured and secure_cookies(request) and expected:
# HF terminates TLS outside ASGI. Only the same hostname on HTTPS is accepted.
expected = ("https", expected[1], 443 if expected[2] == 80 else expected[2])
return actual is None or expected is None or actual != expected
class BrowserSessionMiddleware:
def __init__(self, app):
self.app = app
async def __call__(self, scope, receive, send):
if scope["type"] == "http" and scope["method"].upper() not in _SAFE and scope["path"].startswith("/api/"):
request = Request(scope)
cookie_token = request.cookies.get(cookie_names(request)[0], "")
cookie_auth = bool(cookie_token) and "authorization" not in request.headers
reason = None
if (cookie_auth or scope["path"].startswith("/api/auth/")) and _foreign_origin(request):
reason = "Request origin is not allowed"
elif cookie_auth and scope["path"] not in _PREAUTH:
supplied = request.headers.get("x-sinetti-csrf", "")
try:
valid = bool(supplied) and hmac.compare_digest(supplied, _csrf(cookie_token))
except (RuntimeError, ValueError, TypeError):
valid = False
if not valid:
reason = "Session request verification failed"
if reason:
response = JSONResponse({"ok": False, "error": {"code": "csrf_rejected", "message": reason}}, status_code=403)
await response(scope, receive, send)
return
await self.app(scope, receive, send)
|