From 96a070844898b4fba9a77dda844a875e23e675d7 Mon Sep 17 00:00:00 2001 From: mark Date: Fri, 10 Jul 2026 18:31:27 -0400 Subject: [PATCH] fix(auth): preserve provider fallback during refresh --- hermes_cli/dashboard_auth/base.py | 16 +- hermes_cli/dashboard_auth/middleware.py | 78 ++- .../test_dashboard_auth_401_reauth.py | 96 ++++ .../hermes_cli/test_dashboard_auth_cookies.py | 504 +++++++++--------- 4 files changed, 413 insertions(+), 281 deletions(-) diff --git a/hermes_cli/dashboard_auth/base.py b/hermes_cli/dashboard_auth/base.py index 8f376f352108..e8b8a7730b1d 100644 --- a/hermes_cli/dashboard_auth/base.py +++ b/hermes_cli/dashboard_auth/base.py @@ -94,9 +94,11 @@ class InvalidCredentialsError(Exception): class RefreshExpiredError(Exception): - """The refresh token is dead. + """This provider rejects the refresh token as dead or invalid. - Middleware clears cookies and forces re-login (302 → ``/login``). + In a multi-provider deployment this does not prove token ownership, so + middleware may try remaining providers. It clears cookies and forces + re-login only after every reachable provider rejects the token. """ @@ -125,9 +127,13 @@ class DashboardAuthProvider(ABC): raises ``ProviderError`` if the IDP is unreachable. Middleware treats expiry and unreachable differently (expiry → refresh; unreachable → 503). - * ``refresh_session`` raises ``RefreshExpiredError`` when the - refresh token is also invalid; middleware then forces re-login. - Raises ``ProviderError`` on network failure. + * ``refresh_session`` raises ``RefreshExpiredError`` when the refresh + token is invalid for that provider. Middleware tries the remaining + providers because an opaque foreign token can be indistinguishable + from an expired one; it forces re-login only after every reachable + provider rejects the token. Raises ``ProviderError`` on network + failure; middleware still tries remaining providers, but returns 503 + without clearing cookies if none succeeds and any was unavailable. * ``revoke_session`` is best-effort and must not raise. Subclasses MUST set ``name`` (lowercase identifier, stable forever) diff --git a/hermes_cli/dashboard_auth/middleware.py b/hermes_cli/dashboard_auth/middleware.py index da5f9f1ed23b..caa1b3a6e5dd 100644 --- a/hermes_cli/dashboard_auth/middleware.py +++ b/hermes_cli/dashboard_auth/middleware.py @@ -24,7 +24,11 @@ from fastapi.responses import JSONResponse, RedirectResponse, Response from hermes_cli.dashboard_auth import list_session_providers from hermes_cli.dashboard_auth.audit import AuditEvent, audit_log -from hermes_cli.dashboard_auth.base import ProviderError, RefreshExpiredError +from hermes_cli.dashboard_auth.base import ( + DashboardAuthProvider, + ProviderError, + RefreshExpiredError, +) from hermes_cli.dashboard_auth.cookies import ( clear_sso_attempt_cookie, read_session_cookies, @@ -85,6 +89,22 @@ def _client_ip(request: Request) -> str: return request.client.host if request.client else "" +def _ordered_session_providers( + provider_hint: str | None, +) -> list[DashboardAuthProvider]: + """Prefer the hinted provider without making the hint authoritative. + + The cookie can outlive a provider rename/removal or become stale after a + deployment change. A stable sort moves a matching provider to the front + while preserving registration order for every remaining candidate; an + unknown hint therefore leaves the normal scan unchanged. + """ + providers = list_session_providers() + if provider_hint: + providers.sort(key=lambda provider: provider.name != provider_hint) + return providers + + def _unauth_response(request: Request, *, reason: str) -> Response: """API routes → 401 JSON with ``login_url``; HTML routes → 302 → /login. @@ -324,10 +344,7 @@ 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 - providers = list_session_providers() - if provider_hint: - providers = [provider for provider in providers if provider.name == provider_hint] - for provider in providers: + for provider in _ordered_session_providers(provider_hint): try: session = provider.verify_session(access_token=at) except ProviderError as e: @@ -359,9 +376,22 @@ async def gated_auth_middleware( # Access token is expired/invalid. Before forcing re-login, try to # rotate it using the refresh token (if the session cookie carries # 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, provider_hint=provider_hint) + # serve the request transparently; only after every provider rejects + # the RT do we fall through to clear-and-relogin. + try: + refreshed = _attempt_refresh( + request, + refresh_token=_rt, + provider_hint=provider_hint, + ) + except ProviderError as e: + # At least one provider could not confirm or reject the RT, and no + # other provider refreshed it. Preserve the cookies and surface a + # transient outage instead of turning uncertainty into a logout. + return JSONResponse( + {"detail": f"Auth provider {str(e)!r} unreachable"}, + status_code=503, + ) if refreshed is not None: new_session, refreshing_provider = refreshed request.state.session = new_session @@ -442,33 +472,29 @@ def _expires_in_seconds(session) -> int: 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 - there's no RT or every provider's ``refresh_session`` failed with - ``RefreshExpiredError`` (dead/revoked/reuse-detected RT → force re-login). - - A ``ProviderError`` (Portal unreachable) is NOT swallowed into a re-login - here — re-raising would 500 the request; instead we log and return None so - the caller forces a clean re-login, which is the safer UX than a hard - error on a transient network blip during the narrow refresh window. + The provider hint only changes candidate order. ``RefreshExpiredError`` + rejects the token for that candidate, but cannot prove ownership because + providers such as Basic raise it for foreign opaque tokens too. Likewise, + ``ProviderError`` only makes that candidate unavailable. Both are audited + and the remaining providers are tried. Returns ``None`` only when there is + no RT or every reachable provider rejects it. If no provider succeeds and + at least one raised ``ProviderError``, re-raises with that provider's name + so the caller can return 503 without clearing potentially valid cookies. """ if not refresh_token: return None - providers = list_session_providers() - if provider_hint: - providers = [provider for provider in providers if provider.name == provider_hint] - for provider in providers: + unavailable_provider: str | None = None + for provider in _ordered_session_providers(provider_hint): try: new_session = provider.refresh_session(refresh_token=refresh_token) except RefreshExpiredError: - # This provider owns the RT but it's dead — stop trying others - # (an RT belongs to exactly one provider) and force re-login. audit_log( AuditEvent.REFRESH_FAILURE, provider=provider.name, reason="refresh_expired", ip=_client_ip(request), ) - return None + continue except ProviderError as e: _log.warning( "dashboard-auth: provider %r unreachable during refresh: %s", @@ -480,7 +506,11 @@ def _attempt_refresh(request: Request, *, refresh_token, provider_hint: str | No reason="provider_unreachable", ip=_client_ip(request), ) - return None + if unavailable_provider is None: + unavailable_provider = provider.name + continue if new_session is not None: return new_session, provider.name + if unavailable_provider is not None: + raise ProviderError(unavailable_provider) return None diff --git a/tests/hermes_cli/test_dashboard_auth_401_reauth.py b/tests/hermes_cli/test_dashboard_auth_401_reauth.py index 769b5d1ccf69..63ab76a9a833 100644 --- a/tests/hermes_cli/test_dashboard_auth_401_reauth.py +++ b/tests/hermes_cli/test_dashboard_auth_401_reauth.py @@ -33,6 +33,7 @@ from fastapi.testclient import TestClient from hermes_cli import web_server from hermes_cli.dashboard_auth import clear_providers, register_provider +from hermes_cli.dashboard_auth.base import ProviderError, RefreshExpiredError from hermes_cli.dashboard_auth.cookies import ( SESSION_AT_COOKIE, SESSION_PROVIDER_COOKIE, @@ -286,6 +287,101 @@ class TestTransparentRefreshOnAccessTokenEviction: for cookie in response.headers.get_list("set-cookie") ) + def test_unknown_provider_hint_retains_verify_fallback(self, gated_app): + """A hint for a removed provider must not suppress the normal scan.""" + 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) + gated_app.cookies.set(SESSION_PROVIDER_COOKIE, "removed-provider") + + response = gated_app.get("/api/auth/me") + + assert response.status_code == 200 + assert response.json()["provider"] == "stub" + + @pytest.mark.parametrize( + "error_type", + [RefreshExpiredError, ProviderError], + ids=["token-rejected", "provider-unreachable"], + ) + def test_stale_provider_hint_refresh_error_falls_back( + self, + gated_app, + error_type, + ): + """A stale known hint may reject a foreign RT or be unavailable. + + Either failure applies only to that provider candidate; remaining + providers still get a chance to claim the token. + """ + class StaleHintProvider(StubAuthProvider): + name = "basic" + + def __init__(self): + super().__init__() + self.refresh_calls = 0 + + def refresh_session(self, *, refresh_token: str): + self.refresh_calls += 1 + raise error_type("foreign refresh token") + + stale = StaleHintProvider() + _provider, valid_rt = self._build_rt_only_app() + clear_providers() + register_provider(stale) + 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, "basic") + + response = gated_app.get("/api/sessions", follow_redirects=False) + + assert response.status_code == 200 + assert stale.refresh_calls == 1 + assert any( + SESSION_PROVIDER_COOKIE in cookie and "stub" in cookie + for cookie in response.headers.get_list("set-cookie") + ) + + def test_refresh_outage_returns_503_without_clearing_cookies(self, gated_app): + """Uncertain ownership during an outage must not log the user out.""" + class UnreachableProvider(StubAuthProvider): + name = "unreachable" + + def refresh_session(self, *, refresh_token: str): + raise ProviderError("simulated provider outage") + + class RejectingProvider(StubAuthProvider): + name = "rejecting" + + def refresh_session(self, *, refresh_token: str): + raise RefreshExpiredError("foreign refresh token") + + clear_providers() + register_provider(UnreachableProvider()) + register_provider(RejectingProvider()) + gated_app.cookies.clear() + gated_app.cookies.set(SESSION_RT_COOKIE, "opaque-refresh-token") + gated_app.cookies.set(SESSION_PROVIDER_COOKIE, "unreachable") + + response = gated_app.get("/api/sessions", follow_redirects=False) + + assert response.status_code == 503 + assert gated_app.cookies.get(SESSION_RT_COOKIE) == "opaque-refresh-token" + assert not any( + SESSION_RT_COOKIE in cookie and "Max-Age=0" 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 diff --git a/tests/hermes_cli/test_dashboard_auth_cookies.py b/tests/hermes_cli/test_dashboard_auth_cookies.py index 25ae2b8ff4ae..3f0baaa4dbb2 100644 --- a/tests/hermes_cli/test_dashboard_auth_cookies.py +++ b/tests/hermes_cli/test_dashboard_auth_cookies.py @@ -1,252 +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_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 +"""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