feat: add KeycloakOAuthProxy with offline token support

Keycloak returns refresh_expires_in=0 for offline_access tokens to signal
'this refresh token never expires'. Base OAuthProxy has no way to know this
is intentional rather than a malformed response, so it falls through to the
standard 1-year wall-clock fallback — which causes the FastMCP refresh token
TTL to shrink on every subsequent refresh cycle until it reaches ~0 after
one year, forcing re-authentication even though the Keycloak offline token
is still valid.

This commit:
- Adds KeycloakOAuthProxy(OAuthProxy) to providers/keycloak.py with a
  convenience __init__ that derives OIDC endpoints from realm_url
- Adds _zero_refresh_expiry_means_never_expires: bool = False class attr
  on OAuthProxy; KeycloakOAuthProxy sets it to True
- Adds refresh_token_never_expires: bool = False to UpstreamTokenSet so
  the intent is visible in stored state
- When the flag is set and val==0: marks the token as never-expiring and
  clears refresh_token_expires_at so subsequent refresh cycles always get
  a fresh full fallback-TTL FastMCP RT instead of a decaying one
- Base OAuthProxy is completely unchanged for val==0: falls through to
  the existing 1-year wall-clock fallback as before

Tests:
- test_refresh_expires_in_zero_issues_refresh_token: KeycloakOAuthProxy
  correctly issues a refresh token and marks upstream as never-expiring
- test_refresh_expires_in_zero_subsequent_refresh_does_not_shrink: TTL
  stays at ~1 year after repeated refresh cycles
- test_base_proxy_does_not_treat_zero_as_never_expires: confirms base
  OAuthProxy behaviour is unaffected

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

fastmcp_slim/fastmcp/server/auth/oauth_proxy/models.py

@@ -92,6 +92,7 @@ class UpstreamTokenSet(BaseModel):
     refresh_token_expires_at: (
         float | None
     )  # Unix timestamp when refresh token expires (if known)
+    refresh_token_never_expires: bool = False  # True = upstream explicitly said "no expiry" (e.g. Keycloak offline_access)
     expires_at: float  # Unix timestamp when access token expires
     token_type: str  # Usually "Bearer"
     scope: str  # Space-separated scopes

fastmcp_slim/fastmcp/server/auth/oauth_proxy/proxy.py

@@ -231,6 +231,12 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
     - Generic: Works with any spec-compliant provider
     """
 
+    # Subclasses can set this to True to treat refresh_expires_in=0 from the
+    # upstream as "this refresh token never expires" rather than an unknown expiry.
+    # RFC 6749 does not define refresh_expires_in; 0 is a Keycloak-specific
+    # convention for offline_access tokens.  Do not enable for other providers.
+    _zero_refresh_expiry_means_never_expires: bool = False
+
     def __init__(
         self,
         *,
@@ -1090,6 +1096,7 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
         # Some providers include refresh_expires_in, some don't
         refresh_expires_in = None
         refresh_token_expires_at = None
+        refresh_token_never_expires = False
         if idp_tokens.get("refresh_token"):
             if "refresh_expires_in" in idp_tokens:
                 val = int(idp_tokens["refresh_expires_in"])
@@ -1100,21 +1107,24 @@ 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".
+                elif val == 0 and self._zero_refresh_expiry_means_never_expires:
+                    # Provider explicitly signals "no expiry" (e.g. Keycloak offline_access).
                     # 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.
+                    # sessions roll forward automatically without ever hitting a hard wall.
+                    refresh_token_never_expires = True
                     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 expiry (or signalled never-expires with 0).
+                # Upstream didn't specify expiry; use configured fallback (default 1 year).
                 refresh_expires_in = self._fallback_refresh_token_expiry_seconds
-                refresh_token_expires_at = time.time() + refresh_expires_in
+                if not refresh_token_never_expires:
+                    # Track wall-clock expiry only for tokens with a real deadline.
+                    refresh_token_expires_at = time.time() + refresh_expires_in
                 logger.debug(
                     "Upstream refresh token expiry unknown, using fallback %d seconds",
                     refresh_expires_in,
@@ -1128,6 +1138,7 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
             if idp_tokens.get("refresh_token")
             else None,
             refresh_token_expires_at=refresh_token_expires_at,
+            refresh_token_never_expires=refresh_token_never_expires,
             expires_at=time.time() + expires_in,
             token_type=idp_tokens.get("token_type", "Bearer"),
             scope=" ".join(granted_scopes),
@@ -1428,10 +1439,10 @@ 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.
+                elif val == 0 and self._zero_refresh_expiry_means_never_expires:
+                    # Provider signals "no expiry" — mark and clear stale wall-clock time
+                    # so the fallback below always issues a fresh full-length FastMCP RT.
+                    upstream_token_set.refresh_token_never_expires = True
                     upstream_token_set.refresh_token_expires_at = None
             if new_refresh_expires_in is None:
                 if upstream_token_set.refresh_token_expires_at:
@@ -1440,11 +1451,12 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
                         upstream_token_set.refresh_token_expires_at - time.time()
                     )
                 else:
-                    # Upstream rotated the refresh token but gave no expiry; use fallback
+                    # Upstream rotated the refresh token but gave no expiry (or never-expires); use fallback
                     new_refresh_expires_in = self._fallback_refresh_token_expiry_seconds
-                    upstream_token_set.refresh_token_expires_at = (
-                        time.time() + new_refresh_expires_in
-                    )
+                    if not upstream_token_set.refresh_token_never_expires:
+                        upstream_token_set.refresh_token_expires_at = (
+                            time.time() + new_refresh_expires_in
+                        )
 
         upstream_token_set.raw_token_data = {
             **upstream_token_set.raw_token_data,
@@ -1645,9 +1657,9 @@ 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.
+                elif val == 0 and self._zero_refresh_expiry_means_never_expires:
+                    # Provider signals "no expiry" — mark and clear stale wall-clock time.
+                    upstream_token_set.refresh_token_never_expires = True
                     upstream_token_set.refresh_token_expires_at = None
             if new_refresh_expires_in is None:
                 if upstream_token_set.refresh_token_expires_at:
@@ -1656,9 +1668,10 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
                     )
                 else:
                     new_refresh_expires_in = self._fallback_refresh_token_expiry_seconds
-                    upstream_token_set.refresh_token_expires_at = (
-                        time.time() + new_refresh_expires_in
-                    )
+                    if not upstream_token_set.refresh_token_never_expires:
+                        upstream_token_set.refresh_token_expires_at = (
+                            time.time() + new_refresh_expires_in
+                        )
 
         upstream_token_set.raw_token_data = {
             **upstream_token_set.raw_token_data,

fastmcp_slim/fastmcp/server/auth/providers/keycloak.py

@@ -2,13 +2,19 @@
 
 from __future__ import annotations
 
+from typing import TYPE_CHECKING, Literal
+
 from pydantic import AnyHttpUrl
 
 from fastmcp.server.auth import RemoteAuthProvider, TokenVerifier
+from fastmcp.server.auth.oauth_proxy import OAuthProxy
 from fastmcp.server.auth.providers.jwt import JWTVerifier
 from fastmcp.utilities.auth import parse_scopes
 from fastmcp.utilities.logging import get_logger
 
+if TYPE_CHECKING:
+    from key_value.aio.protocols import AsyncKeyValue
+
 logger = get_logger(__name__)
 
 
@@ -72,3 +78,137 @@ class KeycloakAuthProvider(RemoteAuthProvider):
             authorization_servers=[AnyHttpUrl(self.realm_url)],
             base_url=AnyHttpUrl(str(base_url).rstrip("/")),
         )
+
+
+class KeycloakOAuthProxy(OAuthProxy):
+    """OAuth proxy for Keycloak identity providers.
+
+    Use this instead of `OAuthProxy` when proxying to Keycloak. It handles
+    Keycloak-specific token response conventions, most importantly
+    `refresh_expires_in=0`, which Keycloak uses to indicate that a refresh
+    token obtained with the `offline_access` scope never expires. Standard
+    `OAuthProxy` treats `0` as an unknown expiry, which causes the FastMCP
+    refresh token TTL to shrink on every refresh cycle until it hits zero
+    and forces the user to re-authenticate — even though the Keycloak
+    offline token is still valid.
+
+    All other behaviour is identical to `OAuthProxy`. Pass `realm_url` for
+    automatic endpoint discovery, or supply the individual endpoint URLs
+    directly.
+
+    Example:
+        ```python
+        from fastmcp import FastMCP
+        from fastmcp.server.auth.providers.keycloak import KeycloakOAuthProxy
+
+        auth = KeycloakOAuthProxy(
+            realm_url="https://keycloak.example.com/realms/myrealm",
+            upstream_client_id="my-client",
+            upstream_client_secret="my-secret",
+            base_url="https://my-mcp-server.example.com",
+            jwt_signing_key="some-secret",
+        )
+
+        mcp = FastMCP("My App", auth=auth)
+        ```
+    """
+
+    # Keycloak uses refresh_expires_in=0 for offline_access tokens ("never expires").
+    _zero_refresh_expiry_means_never_expires: bool = True
+
+    def __init__(
+        self,
+        *,
+        realm_url: AnyHttpUrl | str | None = None,
+        upstream_client_id: str,
+        upstream_client_secret: str | None = None,
+        base_url: AnyHttpUrl | str,
+        required_scopes: list[str] | str | None = None,
+        audience: str | list[str] | None = None,
+        token_verifier: TokenVerifier | None = None,
+        # Direct endpoint overrides (optional if realm_url is provided)
+        upstream_authorization_endpoint: str | None = None,
+        upstream_token_endpoint: str | None = None,
+        upstream_revocation_endpoint: str | None = None,
+        # Pass-through OAuthProxy options
+        jwt_signing_key: str | bytes | None = None,
+        client_storage: AsyncKeyValue | None = None,
+        require_authorization_consent: bool | Literal["remember", "external"] = True,
+        allowed_client_redirect_uris: list[str] | None = None,
+        fallback_refresh_token_expiry_seconds: int | None = None,
+    ):
+        """Initialize the Keycloak OAuth proxy.
+
+        Args:
+            realm_url: Keycloak realm URL (e.g., "https://keycloak.example.com/realms/myrealm").
+                Used to derive authorization and token endpoints automatically. Required unless
+                `upstream_authorization_endpoint` and `upstream_token_endpoint` are both provided.
+            upstream_client_id: Client ID of the application registered in Keycloak.
+            upstream_client_secret: Client secret. Optional for public clients.
+            base_url: Public URL of this FastMCP server.
+            required_scopes: Scopes to require on incoming tokens. Defaults to `["openid"]`.
+            audience: Optional JWT audience for token validation. Recommended for production.
+            token_verifier: Custom token verifier. Defaults to a JWTVerifier configured
+                for the Keycloak realm's JWKS endpoint.
+            upstream_authorization_endpoint: Override the authorization endpoint URL.
+                Required if `realm_url` is not provided.
+            upstream_token_endpoint: Override the token endpoint URL.
+                Required if `realm_url` is not provided.
+            upstream_revocation_endpoint: Optional token revocation endpoint.
+            jwt_signing_key: Secret for signing FastMCP JWTs.
+            client_storage: Storage backend for OAuth state.
+            require_authorization_consent: Consent screen behaviour (default True).
+            allowed_client_redirect_uris: Allowed MCP client redirect URI patterns.
+            fallback_refresh_token_expiry_seconds: FastMCP RT lifetime when Keycloak
+                returns `refresh_expires_in=0`. Defaults to 1 year. The token is
+                re-issued automatically on every refresh cycle, so active sessions
+                remain valid indefinitely.
+        """
+        if realm_url is None and (
+            upstream_authorization_endpoint is None or upstream_token_endpoint is None
+        ):
+            raise ValueError(
+                "Either realm_url or both upstream_authorization_endpoint and "
+                "upstream_token_endpoint must be provided."
+            )
+
+        realm = str(realm_url).rstrip("/") if realm_url else None
+        oidc_base = f"{realm}/protocol/openid-connect" if realm else None
+
+        resolved_auth_endpoint = upstream_authorization_endpoint or f"{oidc_base}/auth"
+        resolved_token_endpoint = upstream_token_endpoint or f"{oidc_base}/token"
+        resolved_revocation_endpoint = upstream_revocation_endpoint or (
+            f"{oidc_base}/revoke" if oidc_base else None
+        )
+
+        parsed_scopes = (
+            parse_scopes(required_scopes) if required_scopes is not None else ["openid"]
+        )
+
+        if token_verifier is None:
+            if realm is None:
+                raise ValueError(
+                    "token_verifier must be provided when realm_url is not set."
+                )
+            token_verifier = JWTVerifier(
+                jwks_uri=f"{realm}/protocol/openid-connect/certs",
+                issuer=realm,
+                algorithm="RS256",
+                required_scopes=parsed_scopes,
+                audience=audience,
+            )
+
+        super().__init__(
+            upstream_authorization_endpoint=resolved_auth_endpoint,
+            upstream_token_endpoint=resolved_token_endpoint,
+            upstream_revocation_endpoint=resolved_revocation_endpoint,
+            upstream_client_id=upstream_client_id,
+            upstream_client_secret=upstream_client_secret,
+            token_verifier=token_verifier,
+            base_url=base_url,
+            jwt_signing_key=jwt_signing_key,
+            client_storage=client_storage,
+            require_authorization_consent=require_authorization_consent,
+            allowed_client_redirect_uris=allowed_client_redirect_uris,
+            fallback_refresh_token_expiry_seconds=fallback_refresh_token_expiry_seconds,
+        )

tests/server/auth/oauth_proxy/test_tokens.py

@@ -29,6 +29,7 @@ from fastmcp.server.auth.oauth_proxy.models import (
     _hash_token,
 )
 from fastmcp.server.auth.providers.jwt import JWTVerifier
+from fastmcp.server.auth.providers.keycloak import KeycloakOAuthProxy
 
 
 class TestOAuthProxyTokenEndpointAuth:
@@ -822,6 +823,37 @@ class TestUpstreamTokenStorageTTL:
         )
         assert upstream_tokens is not None
 
+
+class TestKeycloakOAuthProxy:
+    """Tests specific to KeycloakOAuthProxy's offline token handling.
+
+    Keycloak returns refresh_expires_in=0 for offline_access tokens to signal
+    "no time-based expiry".  KeycloakOAuthProxy must handle this correctly:
+    issue a refresh token, and never let the FastMCP RT TTL shrink across
+    repeated refresh cycles.
+    """
+
+    @pytest.fixture
+    def jwt_verifier(self):
+        verifier = Mock(spec=TokenVerifier)
+        verifier.required_scopes = ["read", "write"]
+        verifier.verify_token = AsyncMock(return_value=None)
+        return verifier
+
+    @pytest.fixture
+    def proxy(self, jwt_verifier):
+        proxy = KeycloakOAuthProxy(
+            realm_url="https://keycloak.example.com/realms/test",
+            upstream_client_id="test-client",
+            upstream_client_secret="test-secret",
+            token_verifier=jwt_verifier,
+            base_url="https://proxy.example.com",
+            jwt_signing_key="test-secret-key",
+            client_storage=MemoryStore(),
+        )
+        proxy.set_mcp_path("/mcp")
+        return proxy
+
     async def test_refresh_expires_in_zero_issues_refresh_token(self, proxy):
         """refresh_expires_in=0 should fall back to the configured default.
 
@@ -879,6 +911,20 @@ class TestUpstreamTokenStorageTTL:
         )
         assert refresh_meta is not None
 
+        # Upstream token set should be marked as never-expiring
+        jti_mapping = await proxy._jti_mapping_store.get(
+            key=proxy.jwt_issuer.verify_token(
+                result.refresh_token, expected_token_use="refresh"
+            )["jti"]
+        )
+        assert jti_mapping is not None
+        upstream_token_set = await proxy._upstream_token_store.get(
+            key=jti_mapping.upstream_token_id
+        )
+        assert upstream_token_set is not None
+        assert upstream_token_set.refresh_token_never_expires is True
+        assert upstream_token_set.refresh_token_expires_at is None
+
     async def test_refresh_expires_in_zero_subsequent_refresh_does_not_shrink(
         self, proxy
     ):
@@ -938,7 +984,6 @@ class TestUpstreamTokenStorageTTL:
             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"
         )
@@ -979,14 +1024,90 @@ class TestUpstreamTokenStorageTTL:
             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"
         )
 
+    async def test_base_proxy_does_not_treat_zero_as_never_expires(self):
+        """Base OAuthProxy must not interpret refresh_expires_in=0 as never-expires.
+
+        Only KeycloakOAuthProxy opts in to that behaviour.  The base proxy should
+        fall through to the standard fallback (1-year wall-clock timestamp), preserving
+        existing behaviour for all other providers.
+        """
+        verifier = Mock(spec=TokenVerifier)
+        verifier.required_scopes = ["read"]
+        verifier.verify_token = AsyncMock(return_value=None)
+
+        base_proxy = OAuthProxy(
+            upstream_authorization_endpoint="https://idp.example.com/authorize",
+            upstream_token_endpoint="https://idp.example.com/token",
+            upstream_client_id="test-client",
+            upstream_client_secret="test-secret",
+            token_verifier=verifier,
+            base_url="https://proxy.example.com",
+            jwt_signing_key="test-secret-key",
+            client_storage=MemoryStore(),
+        )
+        base_proxy.set_mcp_path("/mcp")
+
+        client = OAuthClientInformationFull(
+            client_id="test-client",
+            client_secret="test-secret",
+            redirect_uris=[AnyUrl("http://localhost:12345/callback")],
+        )
+        await base_proxy.register_client(client)
+
+        client_code = ClientCode(
+            code="base-zero-code",
+            client_id="test-client",
+            redirect_uri="http://localhost:12345/callback",
+            code_challenge="test-challenge",
+            code_challenge_method="S256",
+            scopes=["read"],
+            idp_tokens={
+                "access_token": "upstream-at",
+                "refresh_token": "upstream-rt",
+                "expires_in": 3600,
+                "refresh_expires_in": 0,
+                "token_type": "Bearer",
+            },
+            expires_at=time.time() + 300,
+            created_at=time.time(),
+        )
+        await base_proxy._code_store.put(key=client_code.code, value=client_code)
+
+        auth_code = AuthorizationCode(
+            code="base-zero-code",
+            scopes=["read"],
+            expires_at=time.time() + 300,
+            client_id="test-client",
+            code_challenge="test-challenge",
+            redirect_uri=AnyUrl("http://localhost:12345/callback"),
+            redirect_uri_provided_explicitly=True,
+        )
+        result = await base_proxy.exchange_authorization_code(
+            client=client,
+            authorization_code=auth_code,
+        )
+        assert result.refresh_token is not None
+
+        jti_mapping = await base_proxy._jti_mapping_store.get(
+            key=base_proxy.jwt_issuer.verify_token(
+                result.refresh_token, expected_token_use="refresh"
+            )["jti"]
+        )
+        assert jti_mapping is not None
+        upstream_token_set = await base_proxy._upstream_token_store.get(
+            key=jti_mapping.upstream_token_id
+        )
+        assert upstream_token_set is not None
+        # Base proxy: never_expires stays False, expires_at is set (wall-clock fallback)
+        assert upstream_token_set.refresh_token_never_expires is False
+        assert upstream_token_set.refresh_token_expires_at is not None
 
 
+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