fix(dashboard-auth): one refresh single-flight for the cookie gate and the native route, off the event loop
The cookie gate (middleware._attempt_refresh) never coalesced concurrent requests carrying the same stale refresh token, so a browser/desktop burst after access-token expiry replayed a just-rotated RT into the provider's reuse detection and the whole session was revoked (#55712). Both refresh paths also ran the synchronous provider HTTP call on the ASGI event loop, which wedged /api/status behind a slow IdP. Generalise Doud-FR's native-route single-flight (#71548) into refresh_singleflight.refresh_session_coalesced and use it from both paths, each in run_in_threadpool. The replay key is the RT alone: a burst that straddles a network change must still coalesce, and whoever presents the RT already owns the session. Middleware keeps its refresh_expired / provider_unreachable audit events via callbacks. Live E2E (evals/dashboard_auth/refresh_singleflight_live_e2e.py, real uvicorn, stub rotating IdP with reuse detection): base 1/4 requests survive each burst, 4 provider calls, /api/status 2.7 s behind one refresh; fixed 4/4, 1 call, 10 ms. Co-authored-by: liuhao1024 <liuhao1024@users.noreply.github.com>
This commit is contained in:
teknium1
115
evals/dashboard_auth/refresh_singleflight_live_e2e.py
Normal file
115
evals/dashboard_auth/refresh_singleflight_live_e2e.py
Normal file
@@ -0,0 +1,115 @@
|
||||
"""Live E2E for #55712: real uvicorn, stub rotating IdP with reuse detection.
|
||||
|
||||
Proves (a) N concurrent stale-RT requests on BOTH refresh paths rotate exactly once, and
|
||||
(b) /api/status answers while a slow provider refresh is in flight (the event-loop wedge).
|
||||
Run against origin/main to see both fail; against the fix to see both pass.
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import socket
|
||||
import sys
|
||||
import tempfile
|
||||
import threading
|
||||
import time
|
||||
import urllib.request
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
|
||||
ROOT = sys.argv[1]
|
||||
sys.path.insert(0, ROOT)
|
||||
os.environ["HERMES_HOME"] = tempfile.mkdtemp(prefix="hermes-e2e-55712-")
|
||||
for m in [k for k in sys.modules if k.startswith(("hermes", "tools", "plugins"))]:
|
||||
del sys.modules[m]
|
||||
|
||||
import uvicorn # noqa: E402
|
||||
|
||||
from hermes_cli import web_server # noqa: E402
|
||||
from hermes_cli.dashboard_auth import register_provider # noqa: E402
|
||||
from hermes_cli.dashboard_auth.base import RefreshExpiredError, Session # noqa: E402
|
||||
from tests.hermes_cli.conftest_dashboard_auth import StubAuthProvider # noqa: E402
|
||||
|
||||
|
||||
class SlowRotatingIdP(StubAuthProvider):
|
||||
name = "stub"
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.calls = 0
|
||||
self.rotated = set()
|
||||
self.delay = 0.0
|
||||
self.lock = threading.Lock()
|
||||
|
||||
def verify_session(self, *, access_token):
|
||||
return None if access_token.startswith("expired") else super().verify_session(access_token=access_token)
|
||||
|
||||
def refresh_session(self, *, refresh_token):
|
||||
with self.lock:
|
||||
self.calls += 1
|
||||
if refresh_token in self.rotated:
|
||||
raise RefreshExpiredError("reuse detected -> session revoked")
|
||||
self.rotated.add(refresh_token)
|
||||
time.sleep(self.delay)
|
||||
return Session(user_id="u1", email="u@x.test", display_name="U", org_id="o", provider="stub",
|
||||
expires_at=int(time.time()) + 900, access_token="fresh-" + refresh_token,
|
||||
refresh_token="rotated-" + refresh_token)
|
||||
|
||||
|
||||
idp = SlowRotatingIdP()
|
||||
register_provider(idp)
|
||||
app = web_server.app
|
||||
app.state.bound_host = "127.0.0.1"
|
||||
app.state.auth_required = True
|
||||
|
||||
s = socket.socket(); s.bind(("127.0.0.1", 0)); port = s.getsockname()[1]; s.close()
|
||||
server = uvicorn.Server(uvicorn.Config(app, host="127.0.0.1", port=port, log_level="warning"))
|
||||
threading.Thread(target=server.run, daemon=True).start()
|
||||
base = f"http://127.0.0.1:{port}"
|
||||
for _ in range(100):
|
||||
try:
|
||||
urllib.request.urlopen(base + "/api/status", timeout=1); break
|
||||
except Exception:
|
||||
time.sleep(0.1)
|
||||
|
||||
|
||||
def http(path, *, method="GET", body=None, headers=None):
|
||||
req = urllib.request.Request(base + path, method=method, headers=headers or {},
|
||||
data=json.dumps(body).encode() if body is not None else None)
|
||||
if body is not None:
|
||||
req.add_header("content-type", "application/json")
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=15) as r:
|
||||
return r.status
|
||||
except urllib.error.HTTPError as e:
|
||||
return e.code
|
||||
|
||||
|
||||
results = {}
|
||||
|
||||
# (1) native bearer path: 4 concurrent refreshes with the same stale RT
|
||||
idp.calls = 0
|
||||
with ThreadPoolExecutor(4) as pool:
|
||||
codes = sorted(pool.map(lambda _: http("/auth/native/refresh", method="POST",
|
||||
body={"refresh_token": "native-stale", "provider": "stub"}), range(4)))
|
||||
results["native_burst"] = {"codes": codes, "provider_calls": idp.calls}
|
||||
|
||||
# (2) cookie gate path: 4 concurrent gated requests with an expired AT + one stale RT
|
||||
idp.calls = 0
|
||||
ck = "hermes_session_at=expired-at; hermes_session_rt=cookie-stale; hermes_session_provider=stub"
|
||||
with ThreadPoolExecutor(4) as pool:
|
||||
codes = sorted(pool.map(lambda _: http("/api/auth/me", headers={"cookie": ck}), range(4)))
|
||||
results["cookie_burst"] = {"codes": codes, "provider_calls": idp.calls}
|
||||
|
||||
# (3) event-loop wedge: a 3s provider refresh must not block /api/status
|
||||
idp.delay = 3.0
|
||||
slow = threading.Thread(target=lambda: http("/auth/native/refresh", method="POST",
|
||||
body={"refresh_token": "slow-stale", "provider": "stub"}))
|
||||
slow.start(); time.sleep(0.3)
|
||||
t0 = time.monotonic(); code = http("/api/status"); dt = time.monotonic() - t0
|
||||
slow.join()
|
||||
results["status_during_slow_refresh"] = {"code": code, "seconds": round(dt, 2)}
|
||||
|
||||
ok = (results["native_burst"] == {"codes": [200] * 4, "provider_calls": 1}
|
||||
and results["cookie_burst"] == {"codes": [200] * 4, "provider_calls": 1}
|
||||
and code == 200 and dt < 1.0)
|
||||
print(json.dumps(results, indent=1))
|
||||
print("VERDICT:", "PASS" if ok else "FAIL")
|
||||
server.should_exit = True
|
||||
@@ -15,16 +15,18 @@ from urllib.parse import quote
|
||||
|
||||
from fastapi import Request
|
||||
from fastapi.responses import JSONResponse, RedirectResponse, Response
|
||||
from starlette.concurrency import run_in_threadpool
|
||||
|
||||
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 ProviderError
|
||||
from hermes_cli.dashboard_auth.cookies import (
|
||||
clear_session_cookies, clear_sso_attempt_cookie, detect_https, read_session_cookies,
|
||||
read_session_provider, read_sso_attempt_cookie, set_session_cookies,
|
||||
set_session_provider_cookie, set_sso_attempt_cookie)
|
||||
from hermes_cli.dashboard_auth.prefix import prefix_from_request
|
||||
from hermes_cli.dashboard_auth.public_paths import PUBLIC_API_PATHS
|
||||
from hermes_cli.dashboard_auth.refresh_singleflight import refresh_session_coalesced
|
||||
from hermes_cli.dashboard_auth.request_utils import (
|
||||
access_token_max_age as _expires_in_seconds, client_ip as _client_ip,
|
||||
extract_bearer as _extract_bearer, is_safe_next_path, scan_session_providers,
|
||||
@@ -187,7 +189,8 @@ async def gated_auth_middleware(
|
||||
# Rotate via the refresh token before forcing re-login; on success the request is
|
||||
# served transparently with the rotated cookies re-set.
|
||||
try:
|
||||
refreshed = _attempt_refresh(request, refresh_token=_rt, provider_hint=provider_hint)
|
||||
refreshed = await run_in_threadpool(
|
||||
_attempt_refresh, request, refresh_token=_rt, provider_hint=provider_hint)
|
||||
except ProviderError as e:
|
||||
# Uncertain (provider unreachable), not rejected: keep the cookies.
|
||||
return unreachable_response(str(e))
|
||||
@@ -206,9 +209,12 @@ async def gated_auth_middleware(
|
||||
|
||||
def _attempt_refresh(request: Request, *, refresh_token, provider_hint: str | None = None):
|
||||
"""Rotate an expired session via the refresh token; ``(Session, provider_name)`` or ``None``.
|
||||
``RefreshExpiredError`` rejects that candidate only (Basic raises it for foreign opaque tokens
|
||||
too); if none succeeds and any raised ``ProviderError`` it is re-raised so the caller returns
|
||||
503 without clearing cookies."""
|
||||
Concurrent requests carrying the same stale RT are coalesced (``refresh_singleflight``): a
|
||||
burst of parallel fetches after AT expiry must not replay a rotated RT into the provider's
|
||||
reuse detection. ``RefreshExpiredError`` rejects that candidate only (Basic raises it for
|
||||
foreign opaque tokens too); if none succeeds and any raised ``ProviderError`` it is
|
||||
re-raised so the caller returns 503 without clearing cookies. Synchronous: the gate runs it
|
||||
in the threadpool so a slow IdP never blocks the event loop."""
|
||||
if not refresh_token:
|
||||
return None
|
||||
|
||||
@@ -217,13 +223,9 @@ def _attempt_refresh(request: Request, *, refresh_token, provider_hint: str | No
|
||||
AuditEvent.REFRESH_FAILURE, provider=provider.name, reason=reason,
|
||||
ip=_client_ip(request))
|
||||
|
||||
def _refresh(provider):
|
||||
new_session = provider.refresh_session(refresh_token=refresh_token)
|
||||
return None if new_session is None else (new_session, provider.name)
|
||||
|
||||
return scan_session_providers(
|
||||
provider_hint, _refresh, phase="refresh", log=_log, swallow=(RefreshExpiredError,),
|
||||
on_swallow=_audit_failure("refresh_expired"),
|
||||
return refresh_session_coalesced(
|
||||
refresh_token, provider_hint or "", phase="refresh", log=_log,
|
||||
on_rejected=_audit_failure("refresh_expired"),
|
||||
on_unreachable=_audit_failure("provider_unreachable"))
|
||||
|
||||
|
||||
|
||||
@@ -1,78 +0,0 @@
|
||||
"""Short-lived native refresh replay, scoped to the concrete provider and caller.
|
||||
|
||||
A provider hint only orders discovery: it must neither split one rotating credential's
|
||||
lock nor let an unrelated provider reuse its result. Raw refresh tokens are never keys.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import logging
|
||||
import threading
|
||||
import time
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
from hermes_cli.dashboard_auth.base import DashboardAuthProvider, RefreshExpiredError, Session
|
||||
from hermes_cli.dashboard_auth.request_utils import scan_session_providers
|
||||
|
||||
_SUCCESS_TTL = 30.0
|
||||
_FAILURE_TTL = 5.0
|
||||
_MAX_ENTRIES = 256
|
||||
_guard = threading.Lock()
|
||||
|
||||
|
||||
@dataclass
|
||||
class _Flight:
|
||||
lock: threading.Lock = field(default_factory=threading.Lock)
|
||||
users: int = 0
|
||||
|
||||
|
||||
# Hold the provider itself while caching: replacement (including same-name scoped
|
||||
# registrations) invalidates identity, and Python cannot recycle its id under a live entry.
|
||||
_cache: dict[tuple[int, bytes], tuple[float, DashboardAuthProvider, Session | None]] = {}
|
||||
_flights: dict[tuple[int, bytes], _Flight] = {}
|
||||
|
||||
|
||||
def _prune(now: float) -> None:
|
||||
for key, (expires, _, _) in list(_cache.items()):
|
||||
if expires <= now:
|
||||
del _cache[key]
|
||||
while len(_cache) > _MAX_ENTRIES:
|
||||
del _cache[min(_cache, key=lambda key: _cache[key][0])]
|
||||
|
||||
|
||||
def _refresh_provider(provider: DashboardAuthProvider, token: str, client_ip: str) -> Session | None:
|
||||
digest = hashlib.sha256(client_ip.encode() + b"\0" + token.encode()).digest()
|
||||
key = (id(provider), digest)
|
||||
with _guard:
|
||||
_prune(time.monotonic())
|
||||
flight = _flights.setdefault(key, _Flight())
|
||||
flight.users += 1
|
||||
try:
|
||||
with flight.lock:
|
||||
with _guard:
|
||||
cached = _cache.get(key)
|
||||
if cached is not None and cached[0] > time.monotonic():
|
||||
return cached[2]
|
||||
try:
|
||||
session = provider.refresh_session(refresh_token=token)
|
||||
except RefreshExpiredError:
|
||||
session = None
|
||||
# ProviderError and unexpected execution failures are deliberately not cached.
|
||||
with _guard:
|
||||
now = time.monotonic()
|
||||
_cache[key] = (now + (_SUCCESS_TTL if session is not None else _FAILURE_TTL), provider, session)
|
||||
_prune(now)
|
||||
return session
|
||||
finally:
|
||||
with _guard:
|
||||
flight.users -= 1
|
||||
if flight.users == 0:
|
||||
_flights.pop(key, None)
|
||||
|
||||
|
||||
def refresh_native_session(token: str, provider_hint: str, client_ip: str) -> Session | None:
|
||||
"""Preserve upstream provider fallback/503 behavior while coalescing each actual issuer."""
|
||||
return scan_session_providers(
|
||||
provider_hint, lambda provider: _refresh_provider(provider, token, client_ip),
|
||||
phase="native refresh", log=logging.getLogger(__name__),
|
||||
)
|
||||
107
hermes_cli/dashboard_auth/refresh_singleflight.py
Normal file
107
hermes_cli/dashboard_auth/refresh_singleflight.py
Normal file
@@ -0,0 +1,107 @@
|
||||
"""Single-flight + short replay cache for rotating refresh tokens (both refresh paths).
|
||||
|
||||
Rotating refresh tokens with reuse detection (Nous Portal, Authelia, most OIDC IdPs) make a
|
||||
replay of an already-rotated RT fatal: the provider revokes the whole session. The desktop and
|
||||
the browser both fire bursts of parallel requests on wake or after the access token lapses,
|
||||
each still carrying the same old RT, so the gateway must let exactly ONE of them reach the
|
||||
provider and hand the rotated session to the rest. The cookie gate (``middleware``) and the
|
||||
native bearer route (``routes.auth_native_refresh``) share this one flight table.
|
||||
|
||||
A provider hint only orders discovery: it must neither split one rotating credential's lock
|
||||
nor let an unrelated provider reuse its result. Raw refresh tokens are never keys.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import logging
|
||||
import threading
|
||||
import time
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Callable, Optional
|
||||
|
||||
from hermes_cli.dashboard_auth.base import DashboardAuthProvider, RefreshExpiredError, Session
|
||||
from hermes_cli.dashboard_auth.request_utils import scan_session_providers
|
||||
|
||||
# The success TTL covers the window between the winning response and the siblings' arrival
|
||||
# (a laptop waking from sleep can deliver its burst over many seconds); the failure TTL only
|
||||
# absorbs a retry storm against a token the provider has already declared dead.
|
||||
_SUCCESS_TTL = 30.0
|
||||
_FAILURE_TTL = 5.0
|
||||
_MAX_ENTRIES = 256
|
||||
_guard = threading.Lock()
|
||||
|
||||
|
||||
@dataclass
|
||||
class _Flight:
|
||||
lock: threading.Lock = field(default_factory=threading.Lock)
|
||||
users: int = 0
|
||||
|
||||
|
||||
# Hold the provider itself while caching: replacement (including same-name scoped
|
||||
# registrations) invalidates identity, and Python cannot recycle its id under a live entry.
|
||||
_cache: dict[tuple[int, bytes], tuple[float, DashboardAuthProvider, Session | None]] = {}
|
||||
_flights: dict[tuple[int, bytes], _Flight] = {}
|
||||
|
||||
|
||||
def _prune(now: float) -> None:
|
||||
for key, (expires, _, _) in list(_cache.items()):
|
||||
if expires <= now:
|
||||
del _cache[key]
|
||||
while len(_cache) > _MAX_ENTRIES:
|
||||
del _cache[min(_cache, key=lambda key: _cache[key][0])]
|
||||
|
||||
|
||||
def _refresh_provider(provider: DashboardAuthProvider, token: str) -> Session | None:
|
||||
# Keyed on the token alone: whoever presents this RT already owns the session, and a client
|
||||
# that changed network between two requests of one burst must still hit the cache.
|
||||
digest = hashlib.sha256(token.encode()).digest()
|
||||
key = (id(provider), digest)
|
||||
with _guard:
|
||||
_prune(time.monotonic())
|
||||
flight = _flights.setdefault(key, _Flight())
|
||||
flight.users += 1
|
||||
try:
|
||||
with flight.lock:
|
||||
with _guard:
|
||||
cached = _cache.get(key)
|
||||
if cached is not None and cached[0] > time.monotonic():
|
||||
return cached[2]
|
||||
try:
|
||||
session = provider.refresh_session(refresh_token=token)
|
||||
except RefreshExpiredError:
|
||||
session = None
|
||||
# ProviderError and unexpected execution failures are deliberately not cached.
|
||||
with _guard:
|
||||
now = time.monotonic()
|
||||
_cache[key] = (now + (_SUCCESS_TTL if session is not None else _FAILURE_TTL), provider, session)
|
||||
_prune(now)
|
||||
return session
|
||||
finally:
|
||||
with _guard:
|
||||
flight.users -= 1
|
||||
if flight.users == 0:
|
||||
_flights.pop(key, None)
|
||||
|
||||
|
||||
def refresh_session_coalesced(
|
||||
token: str, provider_hint: str, *, phase: str, log: logging.Logger,
|
||||
on_rejected: Optional[Callable[[DashboardAuthProvider], None]] = None,
|
||||
on_unreachable: Optional[Callable[[DashboardAuthProvider], None]] = None,
|
||||
) -> Optional[tuple[Session, str]]:
|
||||
"""Rotate ``token`` through the provider stack with per-provider single-flight.
|
||||
|
||||
``(Session, provider_name)`` or ``None`` when every provider rejects the token; a
|
||||
``ProviderError`` propagates when nothing rotated and one provider was unreachable
|
||||
(``scan_session_providers`` semantics, so callers keep their 503-not-relogin handling).
|
||||
Synchronous and network-bound: async callers run it in a threadpool.
|
||||
"""
|
||||
def _call(provider: DashboardAuthProvider):
|
||||
session = _refresh_provider(provider, token)
|
||||
if session is None:
|
||||
if on_rejected is not None:
|
||||
on_rejected(provider)
|
||||
return None
|
||||
return session, provider.name
|
||||
|
||||
return scan_session_providers(
|
||||
provider_hint, _call, phase=phase, log=log, on_unreachable=on_unreachable)
|
||||
@@ -41,7 +41,7 @@ from hermes_cli.dashboard_auth.cookies import (
|
||||
set_session_cookies)
|
||||
from hermes_cli.dashboard_auth.login_page import (
|
||||
render_login_html, render_native_provider_choice_html)
|
||||
from hermes_cli.dashboard_auth.native_refresh import refresh_native_session
|
||||
from hermes_cli.dashboard_auth.refresh_singleflight import refresh_session_coalesced
|
||||
from hermes_cli.dashboard_auth.request_utils import (
|
||||
access_token_max_age, client_ip as _client_ip, is_safe_next_path)
|
||||
|
||||
@@ -501,14 +501,15 @@ async def auth_native_refresh(request: Request, body: _NativeRefreshBody):
|
||||
if not body.refresh_token:
|
||||
raise _http(400, "refresh_token required")
|
||||
try:
|
||||
# Uvicorn validates trusted proxy peers before updating the ASGI client.
|
||||
# Never split replay keys on caller-controlled X-Forwarded-For prefixes.
|
||||
session = await run_in_threadpool(
|
||||
refresh_native_session, body.refresh_token, body.provider,
|
||||
request.client.host if request.client else "")
|
||||
# Off the event loop: the provider call is synchronous network I/O and a slow IdP
|
||||
# otherwise wedges every public endpoint (/api/status) behind it.
|
||||
refreshed = await run_in_threadpool(
|
||||
refresh_session_coalesced, body.refresh_token, body.provider,
|
||||
phase="native refresh", log=_log)
|
||||
except ProviderError as e:
|
||||
raise _http(503, f"Auth provider {str(e)!r} unreachable")
|
||||
if session is not None:
|
||||
if refreshed is not None:
|
||||
session = refreshed[0]
|
||||
_audit(request, AuditEvent.REFRESH_SUCCESS, provider=session.provider,
|
||||
user_id=session.user_id)
|
||||
return _bearer_payload(session)
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
"""Native HTTP replay boundaries and deterministic provider-level concurrency."""
|
||||
import logging
|
||||
import threading
|
||||
import time
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
@@ -8,7 +9,7 @@ from fastapi import FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from hermes_cli.dashboard_auth import clear_providers, register_provider
|
||||
from hermes_cli.dashboard_auth import native_refresh as replay
|
||||
from hermes_cli.dashboard_auth import refresh_singleflight as replay
|
||||
from hermes_cli.dashboard_auth.base import ProviderError, RefreshExpiredError, Session
|
||||
from hermes_cli.dashboard_auth.routes import router
|
||||
from tests.hermes_cli.conftest_dashboard_auth import StubAuthProvider
|
||||
@@ -50,8 +51,8 @@ def isolated_registry():
|
||||
replay._cache.clear()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("case", ["hint-fallback", "negative", "outage", "replacement", "client",
|
||||
"ttl", "capacity", "independent", "xff"])
|
||||
@pytest.mark.parametrize("case", ["hint-fallback", "negative", "outage", "replacement",
|
||||
"ttl", "capacity", "independent", "network-hop"])
|
||||
def test_native_http_refresh_boundaries(case, monkeypatch):
|
||||
owner = Provider("owner", "expired" if case == "negative" else "outage" if case == "outage" else "success")
|
||||
other = Provider("other", "success" if case == "independent" else "expired")
|
||||
@@ -81,10 +82,6 @@ def test_native_http_refresh_boundaries(case, monkeypatch):
|
||||
register_provider(replacement)
|
||||
assert request().status_code == 200
|
||||
assert replacement.calls == 1
|
||||
elif case == "client":
|
||||
with TestClient(app, client=("192.0.2.12", 2345)) as another_client:
|
||||
assert another_client.post("/auth/native/refresh", json={"refresh_token": "opaque-old-token"}).status_code == 200
|
||||
assert owner.calls == 2
|
||||
elif case == "ttl":
|
||||
assert request().json() == first.json()
|
||||
now[0] += replay._SUCCESS_TTL
|
||||
@@ -103,15 +100,20 @@ def test_native_http_refresh_boundaries(case, monkeypatch):
|
||||
assert first.json()["provider"] == "owner"
|
||||
assert owner.calls == other.calls == 1
|
||||
else:
|
||||
for prefix in ("192.0.2.1", "192.0.2.2"):
|
||||
assert request(headers={"x-forwarded-for": f"{prefix}, 192.0.2.100"}).status_code == 200
|
||||
# Only the ASGI peer (validated by Uvicorn), never an arbitrary header, scopes replay.
|
||||
# A burst that straddles a network change (laptop wakes on another Wi-Fi) still
|
||||
# coalesces: the RT identifies the session, the peer address does not.
|
||||
with TestClient(app, client=("192.0.2.12", 2345)) as another_client:
|
||||
assert another_client.post("/auth/native/refresh", json={"refresh_token": "opaque-old-token"}).json() == first.json()
|
||||
assert owner.calls == 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize("outcome, independent", [("success", False), ("expired", False),
|
||||
("outage", False), ("success", True)])
|
||||
def test_concurrent_refresh_uses_concrete_provider_identity(outcome, independent):
|
||||
def coalesced(token, hint):
|
||||
return replay.refresh_session_coalesced(
|
||||
token, hint, phase="test", log=logging.getLogger(__name__))
|
||||
|
||||
owner = Provider("owner", outcome)
|
||||
other = Provider("other", "success" if independent else "expired")
|
||||
owner.release.clear()
|
||||
@@ -120,9 +122,9 @@ def test_concurrent_refresh_uses_concrete_provider_identity(outcome, independent
|
||||
register_provider(owner)
|
||||
register_provider(other)
|
||||
with ThreadPoolExecutor(max_workers=3) as pool:
|
||||
first = pool.submit(replay.refresh_native_session, "same-token", "owner", "client")
|
||||
first = pool.submit(coalesced, "same-token", "owner")
|
||||
assert owner.entered.wait(3)
|
||||
second = pool.submit(replay.refresh_native_session, "same-token", "other", "client")
|
||||
second = pool.submit(coalesced, "same-token", "other")
|
||||
try:
|
||||
if independent:
|
||||
assert other.entered.wait(3), "unrelated providers must not share a lock"
|
||||
@@ -149,6 +151,64 @@ def test_concurrent_refresh_uses_concrete_provider_identity(outcome, independent
|
||||
if outcome == "expired":
|
||||
assert results == [None, None]
|
||||
else:
|
||||
assert [result.provider for result in results] == ["owner", "other" if independent else "owner"]
|
||||
assert [result[1] for result in results] == ["owner", "other" if independent else "owner"]
|
||||
assert owner.calls == 1
|
||||
assert not replay._flights
|
||||
|
||||
|
||||
class _RotatingReuseDetectingProvider(Provider):
|
||||
"""A rotating-RT IdP with reuse detection: replaying a rotated RT kills the session."""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__("stub")
|
||||
self.rotated: set[str] = set()
|
||||
|
||||
def verify_session(self, *, access_token):
|
||||
return None # every AT presented is expired -> the gate must refresh
|
||||
|
||||
def refresh_session(self, *, refresh_token):
|
||||
self.calls += 1
|
||||
if refresh_token in self.rotated:
|
||||
raise RefreshExpiredError("refresh token reuse detected")
|
||||
self.rotated.add(refresh_token)
|
||||
self.entered.set()
|
||||
assert self.release.wait(5), "test provider timed out"
|
||||
return Session(user_id="u", email="u@example.test", display_name="u", org_id="o",
|
||||
provider=self.name, expires_at=int(time.time()) + 900,
|
||||
access_token="fresh-at", refresh_token=f"rt-{self.calls}")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def gated_web_app():
|
||||
from hermes_cli import web_server
|
||||
|
||||
prev = {k: getattr(web_server.app.state, k, None) for k in ("bound_host", "bound_port", "auth_required")}
|
||||
web_server.app.state.bound_host = "gw.example.test"
|
||||
web_server.app.state.bound_port = 443
|
||||
web_server.app.state.auth_required = True
|
||||
yield web_server.app
|
||||
for k, v in prev.items():
|
||||
setattr(web_server.app.state, k, v)
|
||||
|
||||
|
||||
def test_cookie_gate_burst_with_stale_rt_rotates_once(gated_web_app):
|
||||
"""#55712: a browser burst after AT expiry carries one stale RT in N requests; exactly one
|
||||
reaches the provider and every sibling is served under the rotated session."""
|
||||
provider = _RotatingReuseDetectingProvider()
|
||||
provider.release.clear()
|
||||
register_provider(provider)
|
||||
cookies = {"hermes_session_at": "expired-at", "hermes_session_rt": "stale-rt",
|
||||
"hermes_session_provider": "stub"}
|
||||
|
||||
def call():
|
||||
# One TestClient per request: a shared jar would hand later requests the rotated RT.
|
||||
with TestClient(gated_web_app, base_url="http://gw.example.test") as client:
|
||||
return client.get("/api/auth/me", cookies=cookies)
|
||||
|
||||
with ThreadPoolExecutor(max_workers=4) as pool:
|
||||
futures = [pool.submit(call) for _ in range(4)]
|
||||
assert provider.entered.wait(3)
|
||||
provider.release.set()
|
||||
statuses = sorted(f.result(timeout=10).status_code for f in futures)
|
||||
assert statuses == [200, 200, 200, 200]
|
||||
assert provider.calls == 1
|
||||
Reference in New Issue
Block a user