test(mcp-oauth): bind the 400-recovery path to issuer binding, guard live-TTL on an empty context

The 400-recovery reload installs a disk pair and only then re-runs issuer
binding on it. When that pair was minted by a different issuer the enforcer
strips its refresh token, and the reload must report "no recovery" so the
session is cleared like any other dead grant. No test pinned that verdict:
a reload that ignored the install result would keep a stripped pair in the
context and return True. The new invariant drives a real 400 through
_handle_refresh_response against a foreign-bound disk pair and asserts the
result is False, the context is cleared, and the foreign refresh token does
not survive on disk.

_hermes_live_ttl read expires_in off current_tokens without checking for
None; getattr(None, ...) happened to yield the default and report "live".
Both current callers install a pair first, but the helper is now explicit
that an empty context is never live, so a future call site cannot adopt
nothing.
This commit is contained in:
kshitijk4poor
2026-09-15 01:21:02 +05:30
committed by kshitij
parent f4bf786c57
commit 8ad7ae06e7
2 changed files with 26 additions and 0 deletions

View File

@@ -952,3 +952,27 @@ async def test_refresh_fence_surfaces_non_contention_lock_errors_immediately(tmp
with pytest.raises(mcp_oauth.RefreshFenceTimeout, match="unavailable on this filesystem"):
await mcp_oauth.acquire_refresh_fence(tmp_path / "srv.json", timeout=5.0)
assert time.monotonic() - started < 1.0, "must fail fast, not wait out the deadline"
@pytest.mark.asyncio
async def test_refresh_400_recovery_rejects_disk_pair_from_another_issuer(tmp_path, monkeypatch):
"""A 400 must not be "recovered" with a disk pair bound to a different issuer.
The enforcer strips that pair's refresh token on install; a stripped pair
is not a recovery, so the session is cleared as on any dead grant and the
foreign refresh token never survives on disk.
"""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
endpoint = "https://idp.example.com/oauth/token"
provider = _fenced_provider(tmp_path, monkeypatch, endpoint)
storage = provider.context.storage
storage.bind_issuer("https://other-idp.example")
await storage.set_tokens(_token("A2", "R2"))
recovered = await provider._handle_refresh_response(
_fake_response(400, endpoint, b'{"error":"invalid_grant"}')
)
assert recovered is False
assert provider.context.current_tokens is None
assert (await storage.get_tokens()).refresh_token is None, "foreign refresh token must not survive on disk"

View File

@@ -176,6 +176,8 @@ class HermesProviderMixin:
``expires_in`` is optional in RFC 6749: None means no expiry was
issued, which the SDK treats as valid, so it counts as live here too.
"""
if self.context.current_tokens is None:
return False
exp = getattr(self.context.current_tokens, "expires_in", None)
return exp is None or int(exp) > 0