mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-20 15:33:54 +00:00
fix(auth): route session refresh with provider hint cookie
This commit is contained in:
mark
Teknium
parent
72ac0d6af0
commit
f9e35e6e94
5 changed files with 374 additions and 238 deletions
|
|
@ -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)
|
||||
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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)."""
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue