fix: clear stale refresh_token_expires_at for Keycloak offline tokens on subsequent refreshes

When Keycloak returns refresh_expires_in=0 (offline token, never expires) on
every token response, the previous code only guarded the initial exchange path.
On subsequent exchange_refresh_token and transparent refresh cycles, the code
would fall through to 'keep existing expiry' — inheriting the wall-clock
timestamp set at initial exchange.  After ~1 year that decayed to ~0 seconds,
issuing FastMCP RTs with 1-second TTL and forcing re-auth even though the
Keycloak offline token was still valid.

Fix: add elif val == 0 to both refresh paths that clears refresh_token_expires_at
to None.  The fallback branch then issues a fresh full fallback-TTL FastMCP RT
on every cycle, matching the 'always-valid' semantics of offline tokens.

Also improves the debug log message to distinguish 'never expires' from
'expiry not provided' so operators can see exactly what Keycloak sent.

Adds a regression test: test_refresh_expires_in_zero_subsequent_refresh_does_not_shrink

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
strawgate 2026-05-13 12:38:50 -05:00
commit 08baf0763b
2 changed files with 128 additions and 2 deletions

View file

@ -1100,8 +1100,19 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
"Upstream refresh token expires in %d seconds",
refresh_expires_in,
)
elif val == 0:
# Keycloak offline_access tokens use 0 to mean "never expires".
# refresh_token_expires_at stays None (no upstream expiry to track).
# We still need a finite FastMCP RT TTL; use the configured fallback.
# The FastMCP RT is renewed on every transparent refresh, so active
# sessions roll forward automatically.
logger.debug(
"Upstream refresh_expires_in=0 (never expires); "
"FastMCP RT will use fallback TTL of %d seconds",
self._fallback_refresh_token_expiry_seconds,
)
if refresh_expires_in is None:
# Upstream didn't specify; use configured fallback (default 1 year).
# Upstream didn't specify expiry (or signalled never-expires with 0).
refresh_expires_in = self._fallback_refresh_token_expiry_seconds
refresh_token_expires_at = time.time() + refresh_expires_in
logger.debug(
@ -1417,6 +1428,11 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
"Upstream refresh token expires in %d seconds",
new_refresh_expires_in,
)
elif val == 0:
# Keycloak offline token — never expires upstream.
# Clear any stale expires_at so the fallback below issues
# a fresh full-length FastMCP RT instead of a decaying one.
upstream_token_set.refresh_token_expires_at = None
if new_refresh_expires_in is None:
if upstream_token_set.refresh_token_expires_at:
# Keep existing expiry if upstream doesn't provide new one
@ -1629,6 +1645,10 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
upstream_token_set.refresh_token_expires_at = (
time.time() + new_refresh_expires_in
)
elif val == 0:
# Keycloak offline token — never expires upstream.
# Clear stale expires_at so fallback issues a fresh full-length RT.
upstream_token_set.refresh_token_expires_at = None
if new_refresh_expires_in is None:
if upstream_token_set.refresh_token_expires_at:
new_refresh_expires_in = int(

View file

@ -879,8 +879,114 @@ class TestUpstreamTokenStorageTTL:
)
assert refresh_meta is not None
async def test_refresh_expires_in_zero_subsequent_refresh_does_not_shrink(
self, proxy
):
"""Repeated refresh cycles with refresh_expires_in=0 must not shrink the RT TTL.
Keycloak offline tokens return refresh_expires_in=0 on every token response.
The first exchange correctly stores a 1-year FastMCP RT. Subsequent
exchange_refresh_token calls (triggered by the MCP client refreshing its tokens)
must each issue a new FastMCP RT with the same full fallback TTL not a
gradually decaying value computed from the original wall-clock timestamp.
"""
import jwt as pyjwt
client = OAuthClientInformationFull(
client_id="kc-shrink-client",
client_secret="test-secret",
redirect_uris=[AnyUrl("http://localhost:12345/callback")],
)
await proxy.register_client(client)
# ── step 1: initial authorization code exchange ──────────────────────
client_code = ClientCode(
code="kc-offline-code",
client_id="kc-shrink-client",
redirect_uri="http://localhost:12345/callback",
code_challenge="test-challenge",
code_challenge_method="S256",
scopes=["read"],
idp_tokens={
"access_token": "upstream-at-v1",
"refresh_token": "upstream-rt-v1",
"expires_in": 3600,
"refresh_expires_in": 0, # Keycloak offline sentinel
"token_type": "Bearer",
},
expires_at=time.time() + 300,
created_at=time.time(),
)
await proxy._code_store.put(key=client_code.code, value=client_code)
auth_code = AuthorizationCode(
code="kc-offline-code",
scopes=["read"],
expires_at=time.time() + 300,
client_id="kc-shrink-client",
code_challenge="test-challenge",
redirect_uri=AnyUrl("http://localhost:12345/callback"),
redirect_uri_provided_explicitly=True,
)
result1 = await proxy.exchange_authorization_code(
client=client,
authorization_code=auth_code,
)
assert result1.refresh_token is not None
first_rt = result1.refresh_token
first_payload = pyjwt.decode(
first_rt, options={"verify_signature": False, "verify_exp": False}
)
first_ttl = first_payload["exp"] - first_payload["iat"]
# First RT should be the full fallback TTL (1 year ± some seconds)
assert first_ttl > 60 * 60 * 24 * 300, (
f"First RT TTL {first_ttl}s should be close to 1 year"
)
# ── step 2: simulate a later refresh cycle (Keycloak returns 0 again) ─
async def fake_refresh(url, refresh_token, scope=None, **_kwargs):
return {
"access_token": "upstream-at-v2",
"refresh_token": "upstream-rt-v2",
"expires_in": 3600,
"refresh_expires_in": 0, # same Keycloak sentinel
"token_type": "Bearer",
}
mock_client = Mock()
mock_client.refresh_token = AsyncMock(side_effect=fake_refresh)
with patch.object(
proxy, "_create_upstream_oauth_client", return_value=mock_client
):
first_rt_meta = await proxy._refresh_token_store.get(
key=_hash_token(first_rt)
)
assert first_rt_meta is not None
result2 = await proxy.exchange_refresh_token(
client=client,
refresh_token=RefreshToken(
token=first_rt,
client_id="kc-shrink-client",
scopes=["read"],
expires_at=first_rt_meta.expires_at,
),
scopes=["read"],
)
assert result2.refresh_token is not None
second_payload = pyjwt.decode(
result2.refresh_token,
options={"verify_signature": False, "verify_exp": False},
)
second_ttl = second_payload["exp"] - second_payload["iat"]
# After a refresh cycle the new RT must still have a full fallback TTL,
# not a value that has shrunk from the original timestamp.
assert second_ttl > 60 * 60 * 24 * 300, (
f"Refreshed RT TTL {second_ttl}s shrank — expected close to 1 year"
)
class TestTransparentUpstreamRefresh:
"""Tests for transparent upstream token refresh in load_access_token.
When the upstream token expires but a refresh token is available, the proxy