mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-19 15:18:03 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

fix(auth): preserve provider fallback during refresh

Browse source
This commit is contained in:
mark 2026-07-10 18:31:27 -04:00 • committed by Teknium
parent f9e35e6e94
commit 96a0708448
4 changed files with 413 additions and 281 deletions
Download patch file Download diff file Expand all files Collapse all files

16
hermes_cli/dashboard_auth/base.py
View file

@ -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)

78
hermes_cli/dashboard_auth/middleware.py
View file

@ -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

96
tests/hermes_cli/test_dashboard_auth_401_reauth.py
View file

@ -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

504
tests/hermes_cli/test_dashboard_auth_cookies.py
View file

@ -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