Spaces:
Running
Running
| """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) | |