From f9e35e6e9490b2219c9cee599c80fa9e17fcc398 Mon Sep 17 00:00:00 2001 From: mark Date: Thu, 9 Jul 2026 22:33:26 -0400 Subject: [PATCH] fix(auth): route session refresh with provider hint cookie --- hermes_cli/dashboard_auth/cookies.py | 38 ++ hermes_cli/dashboard_auth/middleware.py | 31 +- hermes_cli/dashboard_auth/routes.py | 2 + .../test_dashboard_auth_401_reauth.py | 56 ++ .../hermes_cli/test_dashboard_auth_cookies.py | 485 +++++++++--------- 5 files changed, 374 insertions(+), 238 deletions(-) diff --git a/hermes_cli/dashboard_auth/cookies.py b/hermes_cli/dashboard_auth/cookies.py index ef7f79b27d3f..8bcd9db78eb6 100644 --- a/hermes_cli/dashboard_auth/cookies.py +++ b/hermes_cli/dashboard_auth/cookies.py @@ -66,6 +66,10 @@ from fastapi.responses import Response # request's HTTPS + prefix combination. SESSION_AT_COOKIE = "hermes_session_at" SESSION_RT_COOKIE = "hermes_session_rt" +# Provider that minted the session. This non-secret routing hint prevents a +# refresh token from being handed to the wrong provider when several dashboard +# auth plugins are enabled (for example Basic + Nous OAuth). +SESSION_PROVIDER_COOKIE = "hermes_session_provider" PKCE_COOKIE = "hermes_session_pkce" # One-shot loop-guard marker for the auto-SSO redirect (Phase 1, # cloud-auto-discovery). Set when the gate auto-initiates the portal OAuth @@ -141,6 +145,24 @@ def _common_attrs(*, use_https: bool, prefix: str) -> dict: return attrs +def set_session_provider_cookie( + response: Response, + *, + provider: str, + use_https: bool, + prefix: str = "", +) -> None: + """Persist the non-secret provider routing hint for token refresh.""" + if not provider: + return + response.set_cookie( + _resolved_name(SESSION_PROVIDER_COOKIE, use_https=use_https, prefix=prefix), + provider, + max_age=_RT_MAX_AGE, + **_common_attrs(use_https=use_https, prefix=prefix), + ) + + def set_session_cookies( response: Response, *, @@ -149,6 +171,7 @@ def set_session_cookies( access_token_expires_in: int, use_https: bool, prefix: str = "", + provider: str = "", ) -> None: """Set the session cookies on the response. @@ -181,6 +204,12 @@ def set_session_cookies( max_age=_RT_MAX_AGE, **_common_attrs(use_https=use_https, prefix=prefix), ) + set_session_provider_cookie( + response, + provider=provider, + use_https=use_https, + prefix=prefix, + ) def clear_session_cookies(response: Response, *, prefix: str = "") -> None: @@ -202,6 +231,10 @@ def clear_session_cookies(response: Response, *, prefix: str = "") -> None: f"{variant}{SESSION_RT_COOKIE}", "", max_age=0, path=path, httponly=True, samesite="lax", ) + response.set_cookie( + f"{variant}{SESSION_PROVIDER_COOKIE}", "", max_age=0, + path=path, httponly=True, samesite="lax", + ) def set_pkce_cookie( @@ -248,6 +281,11 @@ def read_session_cookies(request: Request) -> Tuple[Optional[str], Optional[str] return at, rt +def read_session_provider(request: Request) -> Optional[str]: + """Return the provider routing hint associated with the session cookies.""" + return _read_with_fallback(request, SESSION_PROVIDER_COOKIE) + + def read_pkce_cookie(request: Request) -> Optional[str]: return _read_with_fallback(request, PKCE_COOKIE) diff --git a/hermes_cli/dashboard_auth/middleware.py b/hermes_cli/dashboard_auth/middleware.py index 362ed729d65b..da5f9f1ed23b 100644 --- a/hermes_cli/dashboard_auth/middleware.py +++ b/hermes_cli/dashboard_auth/middleware.py @@ -28,7 +28,9 @@ from hermes_cli.dashboard_auth.base import ProviderError, RefreshExpiredError from hermes_cli.dashboard_auth.cookies import ( clear_sso_attempt_cookie, read_session_cookies, + read_session_provider, read_sso_attempt_cookie, + set_session_provider_cookie, set_sso_attempt_cookie, ) from hermes_cli.dashboard_auth.public_paths import PUBLIC_API_PATHS @@ -276,6 +278,7 @@ async def gated_auth_middleware( return await call_next(request) at, _rt = read_session_cookies(request) + provider_hint = read_session_provider(request) if not at and not _rt: # Neither token present — no session at all. Nothing to verify or # refresh. Before falling back to the /login interstitial, try to @@ -321,7 +324,10 @@ async def gated_auth_middleware( # 503 — distinguishing "transient IDP outage" (don't force re-login) # from "token genuinely invalid" (fall through to refresh/relogin). unreachable_provider: str | None = None - for provider in list_session_providers(): + providers = list_session_providers() + if provider_hint: + providers = [provider for provider in providers if provider.name == provider_hint] + for provider in providers: try: session = provider.verify_session(access_token=at) except ProviderError as e: @@ -355,7 +361,7 @@ async def gated_auth_middleware( # one). On success we re-set the rotated cookies on the response and # serve the request transparently; on RefreshExpiredError (RT dead / # revoked / reuse-detected) we fall through to clear-and-relogin. - refreshed = _attempt_refresh(request, refresh_token=_rt) + refreshed = _attempt_refresh(request, refresh_token=_rt, provider_hint=provider_hint) if refreshed is not None: new_session, refreshing_provider = refreshed request.state.session = new_session @@ -378,6 +384,7 @@ async def gated_auth_middleware( access_token_expires_in=_expires_in_seconds(new_session), use_https=detect_https(request), prefix=prefix_from_request(request), + provider=refreshing_provider, ) audit_log( AuditEvent.REFRESH_SUCCESS, @@ -405,7 +412,18 @@ async def gated_auth_middleware( return response request.state.session = session - return await call_next(request) + response = await call_next(request) + if not provider_hint and session.provider: + from hermes_cli.dashboard_auth.cookies import detect_https + from hermes_cli.dashboard_auth.prefix import prefix_from_request + + set_session_provider_cookie( + response, + provider=session.provider, + use_https=detect_https(request), + prefix=prefix_from_request(request), + ) + return response def _expires_in_seconds(session) -> int: @@ -421,7 +439,7 @@ def _expires_in_seconds(session) -> int: return max(60, int(session.expires_at) - int(time.time())) -def _attempt_refresh(request: Request, *, refresh_token): +def _attempt_refresh(request: Request, *, refresh_token, provider_hint: str | None = None): """Try to rotate an expired session via the refresh token. Returns ``(new_session, provider_name)`` on success, or ``None`` if @@ -435,7 +453,10 @@ def _attempt_refresh(request: Request, *, refresh_token): """ if not refresh_token: return None - for provider in list_session_providers(): + providers = list_session_providers() + if provider_hint: + providers = [provider for provider in providers if provider.name == provider_hint] + for provider in providers: try: new_session = provider.refresh_session(refresh_token=refresh_token) except RefreshExpiredError: diff --git a/hermes_cli/dashboard_auth/routes.py b/hermes_cli/dashboard_auth/routes.py index ee595be7d821..5b833e5df79a 100644 --- a/hermes_cli/dashboard_auth/routes.py +++ b/hermes_cli/dashboard_auth/routes.py @@ -365,6 +365,7 @@ async def auth_callback( access_token_expires_in=expires_in, use_https=detect_https(request), prefix=_prefix(request), + provider=session.provider, ) clear_pkce_cookie(resp, prefix=_prefix(request)) # Clear the one-shot auto-SSO loop-guard marker now that login succeeded, @@ -549,6 +550,7 @@ async def auth_password_login(request: Request, body: _PasswordLoginBody): access_token_expires_in=expires_in, use_https=detect_https(request), prefix=_prefix(request), + provider=session.provider, ) return resp diff --git a/tests/hermes_cli/test_dashboard_auth_401_reauth.py b/tests/hermes_cli/test_dashboard_auth_401_reauth.py index 458be58c7942..769b5d1ccf69 100644 --- a/tests/hermes_cli/test_dashboard_auth_401_reauth.py +++ b/tests/hermes_cli/test_dashboard_auth_401_reauth.py @@ -35,6 +35,7 @@ from hermes_cli import web_server from hermes_cli.dashboard_auth import clear_providers, register_provider from hermes_cli.dashboard_auth.cookies import ( SESSION_AT_COOKIE, + SESSION_PROVIDER_COOKIE, SESSION_RT_COOKIE, clear_session_cookies, set_session_cookies, @@ -252,6 +253,61 @@ class TestTransparentRefreshOnAccessTokenEviction: for c in set_cookies ), f"no rotated RT cookie in {set_cookies!r}" + def test_provider_hint_routes_refresh_to_token_owner(self, gated_app): + """A Nous-style RT must not be rejected by Basic just because Basic + was registered first. The non-secret provider hint routes directly to + the provider that minted the session.""" + class WrongProvider(StubAuthProvider): + name = "basic" + + def __init__(self): + super().__init__() + self.refresh_calls = 0 + + def refresh_session(self, *, refresh_token: str): + self.refresh_calls += 1 + raise AssertionError("foreign refresh token reached Basic provider") + + wrong = WrongProvider() + _provider, valid_rt = self._build_rt_only_app() + clear_providers() + register_provider(wrong) + register_provider(StubAuthProvider(default_ttl=900)) + gated_app.cookies.clear() + gated_app.cookies.set(SESSION_RT_COOKIE, valid_rt) + gated_app.cookies.set(SESSION_PROVIDER_COOKIE, "stub") + + response = gated_app.get("/api/sessions", follow_redirects=False) + + assert response.status_code == 200 + assert wrong.refresh_calls == 0 + assert any( + SESSION_PROVIDER_COOKIE in cookie and "stub" in cookie + for cookie in response.headers.get_list("set-cookie") + ) + + def test_valid_legacy_session_is_migrated_with_provider_hint(self, gated_app): + import time as _t + from tests.hermes_cli.conftest_dashboard_auth import _sign + + valid_at = _sign({ + "sub": "stub-user-1", + "email": "stub@example.test", + "name": "Stub User", + "org_id": "stub-org-1", + "exp": int(_t.time()) + 900, + }) + gated_app.cookies.clear() + gated_app.cookies.set(SESSION_AT_COOKIE, valid_at) + + response = gated_app.get("/api/sessions") + + assert response.status_code == 200 + assert any( + SESSION_PROVIDER_COOKIE in cookie and "stub" in cookie + for cookie in response.headers.get_list("set-cookie") + ) + def test_no_cookies_at_all_still_bounces(self, gated_app): """Guard the fix didn't over-reach: a request with NEITHER cookie must still 401 to login (nothing to verify or refresh).""" diff --git a/tests/hermes_cli/test_dashboard_auth_cookies.py b/tests/hermes_cli/test_dashboard_auth_cookies.py index 7109b7b70993..25ae2b8ff4ae 100644 --- a/tests/hermes_cli/test_dashboard_auth_cookies.py +++ b/tests/hermes_cli/test_dashboard_auth_cookies.py @@ -1,233 +1,252 @@ -"""Tests for the dashboard-auth cookie helpers.""" -from __future__ import annotations - -from fastapi import FastAPI -from fastapi.responses import Response -from fastapi.testclient import TestClient -from starlette.requests import Request - -from hermes_cli.dashboard_auth.cookies import ( - PKCE_COOKIE, - SESSION_AT_COOKIE, - SESSION_RT_COOKIE, - clear_pkce_cookie, - clear_session_cookies, - read_pkce_cookie, - read_session_cookies, - set_pkce_cookie, - set_session_cookies, -) - - -def _build_app(use_https: bool = True, prefix: str = ""): - app = FastAPI() - - @app.get("/set") - def set_endpoint(): - r = Response("ok") - set_session_cookies( - r, access_token="AT", refresh_token="RT", - access_token_expires_in=3600, use_https=use_https, - prefix=prefix, - ) - return r - - @app.get("/set-pkce") - def set_pkce(): - r = Response("ok") - set_pkce_cookie(r, payload="provider=stub;state=s;verifier=v", - use_https=use_https, prefix=prefix) - return r - - @app.get("/clear") - def clear(): - r = Response("ok") - clear_session_cookies(r, prefix=prefix) - clear_pkce_cookie(r, prefix=prefix) - return r - - return app - - -# Cookie name resolution helpers used throughout — the bare name resolves -# to a request-shape-dependent variant (__Host- / __Secure- / bare). -# Tests pin a specific shape so a regression in the name-resolution -# logic fails loudly rather than silently breaking sessions. - - -def test_session_cookies_use_host_prefix_on_https_direct(): - """HTTPS + no proxy prefix → __Host- prefix (strongest spec - hardening: bound to exact origin, requires Path=/, requires Secure).""" - client = TestClient(_build_app(use_https=True, prefix="")) - r = client.get("/set") - cookies = r.headers.get_list("set-cookie") - at = next(c for c in cookies if c.startswith(f"__Host-{SESSION_AT_COOKIE}=")) - rt = next(c for c in cookies if c.startswith(f"__Host-{SESSION_RT_COOKIE}=")) - for c in (at, rt): - assert "HttpOnly" in c - assert "samesite=lax" in c.lower() - assert "Secure" in c - assert "Path=/" in c - - -def test_session_cookies_use_secure_prefix_when_proxied(): - """HTTPS + /hermes prefix → __Secure- prefix (__Host- forbids - Path != "/"; __Secure- keeps the Secure-required hardening).""" - client = TestClient(_build_app(use_https=True, prefix="/hermes")) - r = client.get("/set") - cookies = r.headers.get_list("set-cookie") - at = next(c for c in cookies if c.startswith(f"__Secure-{SESSION_AT_COOKIE}=")) - assert "Path=/hermes" in at - assert "Secure" in at - # __Host- variant must NOT be emitted on the prefix path. - assert not any( - c.startswith(f"__Host-{SESSION_AT_COOKIE}=") for c in cookies - ) - - -def test_session_cookies_use_bare_name_on_http(): - """Loopback HTTP dev: __Host- / __Secure- both require Secure, which - we can't set on HTTP. Use bare cookie names.""" - client = TestClient(_build_app(use_https=False)) - r = client.get("/set") - cookies = r.headers.get_list("set-cookie") - # Bare name present; no __Host- / __Secure- variant emitted. - assert any(c.startswith(f"{SESSION_AT_COOKIE}=") for c in cookies) - assert not any( - c.startswith(f"__Host-{SESSION_AT_COOKIE}=") - or c.startswith(f"__Secure-{SESSION_AT_COOKIE}=") - for c in cookies - ) - # No Secure flag (HTTP). - at = next(c for c in cookies if c.startswith(f"{SESSION_AT_COOKIE}=")) - assert "Secure" not in at - - -def test_session_cookies_have_30day_rt_and_token_ttl_at(): - client = TestClient(_build_app(use_https=True)) - r = client.get("/set") - cookies = r.headers.get_list("set-cookie") - at = next(c for c in cookies if c.startswith(f"__Host-{SESSION_AT_COOKIE}=")) - rt = next(c for c in cookies if c.startswith(f"__Host-{SESSION_RT_COOKIE}=")) - assert "Max-Age=3600" in at - assert "Max-Age=2592000" in rt # 30 days = 30 * 86400 - - -def test_clear_session_cookies_emits_expired_at_and_rt(): - """``clear_session_cookies`` emits Max-Age=0 deletions for every - plausible cookie-name variant under the active prefix so we flush - stale cookies that an older deploy may have set under a different - prefix.""" - client = TestClient(_build_app()) - r = client.get("/clear") - cookies = r.headers.get_list("set-cookie") - # At least one variant of each session cookie should be deleted. - assert any( - SESSION_AT_COOKIE in c and "Max-Age=0" in c for c in cookies - ) - assert any( - SESSION_RT_COOKIE in c and "Max-Age=0" in c for c in cookies - ) - - -def test_pkce_cookie_short_ttl_and_path_root(): - client = TestClient(_build_app(use_https=True)) - r = client.get("/set-pkce") - pkce = next( - c for c in r.headers.get_list("set-cookie") - if PKCE_COOKIE in c - ) - assert "HttpOnly" in pkce - assert "Max-Age=600" in pkce # 10 minutes - assert "Path=/" in pkce - assert "Secure" in pkce - - -def test_read_session_cookies_from_request_bare_name(): - """Reader accepts the bare name (loopback) by default.""" - scope = { - "type": "http", - "method": "GET", - "path": "/", - "headers": [( - b"cookie", - f"{SESSION_AT_COOKIE}=at_value; {SESSION_RT_COOKIE}=rt_value".encode(), - )], - } - req = Request(scope) - at, rt = read_session_cookies(req) - assert at == "at_value" - assert rt == "rt_value" - - -def test_read_session_cookies_from_request_host_prefix(): - """Reader also finds cookies set with the __Host- variant - (HTTPS direct deploy).""" - scope = { - "type": "http", - "method": "GET", - "path": "/", - "headers": [( - b"cookie", - f"__Host-{SESSION_AT_COOKIE}=at_value; " - f"__Host-{SESSION_RT_COOKIE}=rt_value".encode(), - )], - } - req = Request(scope) - at, rt = read_session_cookies(req) - assert at == "at_value" - assert rt == "rt_value" - - -def test_read_session_cookies_from_request_secure_prefix(): - """Reader also finds cookies set with the __Secure- variant - (HTTPS behind a proxy prefix).""" - scope = { - "type": "http", - "method": "GET", - "path": "/", - "headers": [( - b"cookie", - f"__Secure-{SESSION_AT_COOKIE}=at_value; " - f"__Secure-{SESSION_RT_COOKIE}=rt_value".encode(), - )], - } - req = Request(scope) - at, rt = read_session_cookies(req) - assert at == "at_value" - assert rt == "rt_value" - - -def test_read_session_cookies_missing_returns_none(): - req = Request({"type": "http", "method": "GET", "path": "/", "headers": []}) - assert read_session_cookies(req) == (None, None) - - -def test_read_pkce_cookie_round_trip(): - scope = { - "type": "http", - "method": "GET", - "path": "/", - "headers": [(b"cookie", f"{PKCE_COOKIE}=state=s;verifier=v".encode())], - } - req = Request(scope) - assert read_pkce_cookie(req) == "state=s" # NB: cookie value stops at ';' - - -def test_detect_https_via_scheme(): - """``detect_https`` reads from request.url.scheme. - - Under uvicorn proxy_headers=True the scheme is rewritten from - ``X-Forwarded-Proto``; that's an integration concern, not unit. - """ - from hermes_cli.dashboard_auth.cookies import detect_https - http_req = Request({ - "type": "http", "method": "GET", "path": "/", "scheme": "http", - "headers": [], "server": ("x", 80), - }) - https_req = Request({ - "type": "http", "method": "GET", "path": "/", "scheme": "https", - "headers": [], "server": ("x", 443), - }) - assert detect_https(http_req) is False - assert detect_https(https_req) is True +"""Tests for the dashboard-auth cookie helpers.""" +from __future__ import annotations + +from fastapi import FastAPI +from fastapi.responses import Response +from fastapi.testclient import TestClient +from starlette.requests import Request + +from hermes_cli.dashboard_auth.cookies import ( + PKCE_COOKIE, + SESSION_AT_COOKIE, + SESSION_PROVIDER_COOKIE, + SESSION_RT_COOKIE, + clear_pkce_cookie, + clear_session_cookies, + read_pkce_cookie, + read_session_cookies, + read_session_provider, + set_pkce_cookie, + set_session_cookies, +) + + +def _build_app(use_https: bool = True, prefix: str = ""): + app = FastAPI() + + @app.get("/set") + def set_endpoint(): + r = Response("ok") + set_session_cookies( + r, access_token="AT", refresh_token="RT", + access_token_expires_in=3600, use_https=use_https, + prefix=prefix, provider="nous", + ) + return r + + @app.get("/set-pkce") + def set_pkce(): + r = Response("ok") + set_pkce_cookie(r, payload="provider=stub;state=s;verifier=v", + use_https=use_https, prefix=prefix) + return r + + @app.get("/clear") + def clear(): + r = Response("ok") + clear_session_cookies(r, prefix=prefix) + clear_pkce_cookie(r, prefix=prefix) + return r + + return app + + +# Cookie name resolution helpers used throughout — the bare name resolves +# to a request-shape-dependent variant (__Host- / __Secure- / bare). +# Tests pin a specific shape so a regression in the name-resolution +# logic fails loudly rather than silently breaking sessions. + + +def test_session_cookies_use_host_prefix_on_https_direct(): + """HTTPS + no proxy prefix → __Host- prefix (strongest spec + hardening: bound to exact origin, requires Path=/, requires Secure).""" + client = TestClient(_build_app(use_https=True, prefix="")) + r = client.get("/set") + cookies = r.headers.get_list("set-cookie") + at = next(c for c in cookies if c.startswith(f"__Host-{SESSION_AT_COOKIE}=")) + rt = next(c for c in cookies if c.startswith(f"__Host-{SESSION_RT_COOKIE}=")) + provider = next(c for c in cookies if c.startswith(f"__Host-{SESSION_PROVIDER_COOKIE}=nous")) + for c in (at, rt, provider): + assert "HttpOnly" in c + assert "samesite=lax" in c.lower() + assert "Secure" in c + assert "Path=/" in c + + +def test_session_cookies_use_secure_prefix_when_proxied(): + """HTTPS + /hermes prefix → __Secure- prefix (__Host- forbids + Path != "/"; __Secure- keeps the Secure-required hardening).""" + client = TestClient(_build_app(use_https=True, prefix="/hermes")) + r = client.get("/set") + cookies = r.headers.get_list("set-cookie") + at = next(c for c in cookies if c.startswith(f"__Secure-{SESSION_AT_COOKIE}=")) + assert "Path=/hermes" in at + assert "Secure" in at + # __Host- variant must NOT be emitted on the prefix path. + assert not any( + c.startswith(f"__Host-{SESSION_AT_COOKIE}=") for c in cookies + ) + + +def test_session_cookies_use_bare_name_on_http(): + """Loopback HTTP dev: __Host- / __Secure- both require Secure, which + we can't set on HTTP. Use bare cookie names.""" + client = TestClient(_build_app(use_https=False)) + r = client.get("/set") + cookies = r.headers.get_list("set-cookie") + # Bare name present; no __Host- / __Secure- variant emitted. + assert any(c.startswith(f"{SESSION_AT_COOKIE}=") for c in cookies) + assert not any( + c.startswith(f"__Host-{SESSION_AT_COOKIE}=") + or c.startswith(f"__Secure-{SESSION_AT_COOKIE}=") + for c in cookies + ) + # No Secure flag (HTTP). + at = next(c for c in cookies if c.startswith(f"{SESSION_AT_COOKIE}=")) + assert "Secure" not in at + + +def test_session_cookies_have_30day_rt_and_token_ttl_at(): + client = TestClient(_build_app(use_https=True)) + r = client.get("/set") + cookies = r.headers.get_list("set-cookie") + at = next(c for c in cookies if c.startswith(f"__Host-{SESSION_AT_COOKIE}=")) + rt = next(c for c in cookies if c.startswith(f"__Host-{SESSION_RT_COOKIE}=")) + assert "Max-Age=3600" in at + assert "Max-Age=2592000" in rt # 30 days = 30 * 86400 + + +def test_clear_session_cookies_emits_expired_at_and_rt(): + """``clear_session_cookies`` emits Max-Age=0 deletions for every + plausible cookie-name variant under the active prefix so we flush + stale cookies that an older deploy may have set under a different + prefix.""" + client = TestClient(_build_app()) + r = client.get("/clear") + cookies = r.headers.get_list("set-cookie") + # At least one variant of each session cookie should be deleted. + assert any( + SESSION_AT_COOKIE in c and "Max-Age=0" in c for c in cookies + ) + assert any( + SESSION_RT_COOKIE in c and "Max-Age=0" in c for c in cookies + ) + assert any( + SESSION_PROVIDER_COOKIE in c and "Max-Age=0" in c for c in cookies + ) + + +def test_pkce_cookie_short_ttl_and_path_root(): + client = TestClient(_build_app(use_https=True)) + r = client.get("/set-pkce") + pkce = next( + c for c in r.headers.get_list("set-cookie") + if PKCE_COOKIE in c + ) + assert "HttpOnly" in pkce + assert "Max-Age=600" in pkce # 10 minutes + assert "Path=/" in pkce + assert "Secure" in pkce + + +def test_read_session_cookies_from_request_bare_name(): + """Reader accepts the bare name (loopback) by default.""" + scope = { + "type": "http", + "method": "GET", + "path": "/", + "headers": [( + b"cookie", + f"{SESSION_AT_COOKIE}=at_value; {SESSION_RT_COOKIE}=rt_value".encode(), + )], + } + req = Request(scope) + at, rt = read_session_cookies(req) + assert at == "at_value" + assert rt == "rt_value" + + +def test_read_session_provider_from_request(): + scope = { + "type": "http", + "method": "GET", + "path": "/", + "headers": [( + b"cookie", + f"__Host-{SESSION_PROVIDER_COOKIE}=nous".encode(), + )], + } + assert read_session_provider(Request(scope)) == "nous" + + +def test_read_session_cookies_from_request_host_prefix(): + """Reader also finds cookies set with the __Host- variant + (HTTPS direct deploy).""" + scope = { + "type": "http", + "method": "GET", + "path": "/", + "headers": [( + b"cookie", + f"__Host-{SESSION_AT_COOKIE}=at_value; " + f"__Host-{SESSION_RT_COOKIE}=rt_value".encode(), + )], + } + req = Request(scope) + at, rt = read_session_cookies(req) + assert at == "at_value" + assert rt == "rt_value" + + +def test_read_session_cookies_from_request_secure_prefix(): + """Reader also finds cookies set with the __Secure- variant + (HTTPS behind a proxy prefix).""" + scope = { + "type": "http", + "method": "GET", + "path": "/", + "headers": [( + b"cookie", + f"__Secure-{SESSION_AT_COOKIE}=at_value; " + f"__Secure-{SESSION_RT_COOKIE}=rt_value".encode(), + )], + } + req = Request(scope) + at, rt = read_session_cookies(req) + assert at == "at_value" + assert rt == "rt_value" + + +def test_read_session_cookies_missing_returns_none(): + req = Request({"type": "http", "method": "GET", "path": "/", "headers": []}) + assert read_session_cookies(req) == (None, None) + + +def test_read_pkce_cookie_round_trip(): + scope = { + "type": "http", + "method": "GET", + "path": "/", + "headers": [(b"cookie", f"{PKCE_COOKIE}=state=s;verifier=v".encode())], + } + req = Request(scope) + assert read_pkce_cookie(req) == "state=s" # NB: cookie value stops at ';' + + +def test_detect_https_via_scheme(): + """``detect_https`` reads from request.url.scheme. + + Under uvicorn proxy_headers=True the scheme is rewritten from + ``X-Forwarded-Proto``; that's an integration concern, not unit. + """ + from hermes_cli.dashboard_auth.cookies import detect_https + http_req = Request({ + "type": "http", "method": "GET", "path": "/", "scheme": "http", + "headers": [], "server": ("x", 80), + }) + https_req = Request({ + "type": "http", "method": "GET", "path": "/", "scheme": "https", + "headers": [], "server": ("x", 443), + }) + assert detect_https(http_req) is False + assert detect_https(https_req) is True