5dea46d13d
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>
116 lines
4.5 KiB
Python
116 lines
4.5 KiB
Python
"""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
|