OAuth proxy issues its own tokens (#2109)

* OAuth proxy issues its own tokens

Implement token factory pattern where proxy issues FastMCP JWTs
instead of forwarding upstream tokens. Tokens are minimal references
(JTI) that map to encrypted upstream credentials stored server-side.

* Update run-tests.yml

* Update secret generation and docs

* Add upgrade guide

.github/workflows/run-tests.yml

@@ -31,7 +31,7 @@ jobs:
         os: [ubuntu-latest, windows-latest]
         python-version: ["3.10"]
       fail-fast: false
-    timeout-minutes: 5
+    timeout-minutes: 10
 
     steps:
       - uses: actions/checkout@v5

docs/deployment/http.mdx

@@ -3,6 +3,7 @@ title: HTTP Deployment
 sidebarTitle: HTTP Deployment
 description: Deploy your FastMCP server over HTTP for remote access
 icon: server
+tag: NEW
 ---
 
 import { VersionBadge } from "/snippets/version-badge.mdx";
@@ -479,6 +480,44 @@ Deploy with your secrets safely stored in environment variables:
 MCP_AUTH_TOKEN=secret uvicorn app:app --host 0.0.0.0 --port 8000
 ```
 
+### OAuth Token Security
+
+<VersionBadge version="2.13.0" />
+
+If you're using the [OAuth Proxy](/servers/auth/oauth-proxy), FastMCP issues its own JWT tokens to clients instead of forwarding upstream provider tokens. This maintains proper OAuth 2.0 token boundaries, but requires specific production configuration to ensure tokens survive server restarts.
+
+**Development vs Production:**
+
+By default, token cryptographic keys are ephemeral—generated from a random salt at startup and not persisted anywhere. This means keys change on every restart, invalidating all tokens and triggering client re-authentication. This works fine for development and testing where re-auth after restart is acceptable.
+
+For production, tokens should survive restarts to avoid disrupting clients. This requires four things working together:
+
+1. **Explicit JWT signing key** for signing tokens issued to clients
+2. **Explicit token encryption key** for encrypting upstream OAuth tokens at rest
+3. **Persistent storage** so encrypted upstream tokens survive restart
+4. **HTTPS deployment** for secure cookie handling
+
+The two keys can be any secret strings (environment variables, secret manager, etc.) and should be different from each other. FastMCP derives proper cryptographic keys from whatever you provide using HKDF.
+
+**Configuration:**
+
+Add two parameters to your auth provider and use persistent storage and HTTPS:
+
+```python {4-7}
+auth = GitHubProvider(
+    client_id=os.environ["GITHUB_CLIENT_ID"],
+    client_secret=os.environ["GITHUB_CLIENT_SECRET"],
+    jwt_signing_key=os.environ["JWT_SIGNING_KEY"],
+    token_encryption_key=os.environ["TOKEN_ENCRYPTION_KEY"],
+    client_storage=RedisStore(host="redis.example.com", ...),
+    base_url="https://your-server.com"  # use HTTPS 
+)
+```
+
+Without explicit keys, new keys are generated each time the server starts. Without persistent storage, encrypted tokens are lost. Both cause token validation to fail after restart, requiring all clients to re-authenticate.
+
+For more details on the token architecture, see [OAuth Proxy Token Architecture](/servers/auth/oauth-proxy#token-architecture).
+
 ## Testing Your Deployment
 
 Once your server is deployed, you'll need to verify it's accessible and functioning correctly. For comprehensive testing strategies including connectivity tests, client testing, and authentication testing, see the [Testing Your Server](/development/tests) guide.

docs/development/upgrade-guide.mdx

@@ -0,0 +1,44 @@
+---
+title: Upgrade Guide
+sidebarTitle: Upgrade Guide
+description: Migration instructions for upgrading between FastMCP versions
+icon: up
+---
+
+This guide provides migration instructions for breaking changes and major updates when upgrading between FastMCP versions.
+
+## v2.13.0
+
+### OAuth Token Key Management
+
+The OAuth proxy now issues its own JWT tokens to clients instead of forwarding upstream provider tokens. This improves security by maintaining proper token audience boundaries.
+
+**What changed:**
+
+The OAuth proxy now implements a token factory pattern - it receives tokens from your OAuth provider (GitHub, Google, etc.), encrypts and stores them, then issues its own FastMCP JWT tokens to clients. This requires cryptographic keys for JWT signing and token encryption.
+
+By default, these keys are ephemeral (random salt at startup, not persisted). For most users (development/testing), this works fine since re-authentication after restart is acceptable. For production deployments where you want tokens to persist across restarts, provide explicit keys via parameters.
+
+**Production deployments:**
+
+If you want tokens to survive server restarts, add two new parameters:
+
+```python
+auth = GitHubProvider(
+    client_id=os.environ["GITHUB_CLIENT_ID"],
+    client_secret=os.environ["GITHUB_CLIENT_SECRET"],
+    base_url="https://your-server.com",
+
+    # Add these for production token persistence
+    jwt_signing_key=os.environ["JWT_SIGNING_KEY"],
+    token_encryption_key=os.environ["TOKEN_ENCRYPTION_KEY"],
+
+    client_storage=RedisStore(...)  # Persistent storage
+)
+```
+
+Both keys accept any secret string. Make sure they're different from each other.
+
+**More information:**
+- [OAuth Token Security](/deployment/http#oauth-token-security) - Complete production setup guide
+- [OAuth Proxy Parameters](/servers/auth/oauth-proxy#configuration-parameters) - Parameter documentation

docs/docs.json

@@ -243,6 +243,7 @@
               "development/contributing",
               "development/tests",
               "development/releases",
+              "development/upgrade-guide",
               "changelog"
             ]
           }

docs/servers/auth/oauth-proxy.mdx

@@ -211,7 +211,9 @@ These parameters are included in all token requests to the upstream provider.
 </ParamField>
 
 <ParamField body="client_storage" type="KVStorage | None">
-  Storage backend for persisting OAuth client registrations. By default, clients are automatically persisted to disk in `~/.config/fastmcp/oauth-proxy-clients/`, allowing them to survive server restarts as long as the filesystem remains accessible. This means MCP clients only need to register once and can reconnect seamlessly after your server restarts.
+  Storage backend for persisting OAuth client registrations and encrypted upstream tokens. By default, clients are automatically persisted to disk in `~/.config/fastmcp/oauth-proxy-clients/`, allowing them to survive server restarts as long as the filesystem remains accessible. This means MCP clients only need to register once and can reconnect seamlessly after your server restarts.
+
+For production deployments with token persistence, use this with `jwt_signing_key` and `token_encryption_key` - all three work together to ensure tokens survive restarts. See [OAuth Token Security](/deployment/http#oauth-token-security).
 
 ```python
 from fastmcp.utilities.storage import InMemoryStorage
@@ -221,6 +223,47 @@ auth = OAuthProxy(..., client_storage=InMemoryStorage())
 ```
 
 </ParamField>
+
+<ParamField body="jwt_signing_key" type="str | bytes | None">
+  Secret used to sign FastMCP JWT tokens issued to clients. Accepts any string or bytes - will be derived into a proper 32-byte cryptographic key using HKDF.
+
+  **Default behavior (None):**
+  Keys are ephemeral (random salt at startup). Tokens become invalid on server restart, triggering client re-authentication. This is fine for development and testing.
+
+  **Production:**
+  Provide an explicit secret (e.g., from environment variable). Works with `token_encryption_key` and `client_storage` to ensure tokens survive restarts - all three parameters are required for production token persistence.
+
+  ```python
+  import os
+
+  auth = OAuthProxy(
+      ...,
+      jwt_signing_key=os.environ["JWT_SIGNING_KEY"],  # Any string!
+      token_encryption_key=os.environ["TOKEN_ENCRYPTION_KEY"],
+      client_storage=RedisStore(...)  # Persistent storage
+  )
+  ```
+
+  See [HTTP Deployment - OAuth Token Security](/deployment/http#oauth-token-security) for complete production setup.
+</ParamField>
+
+<ParamField body="token_encryption_key" type="str | bytes | None">
+  Secret used to encrypt upstream tokens at rest in `client_storage`. Accepts any string or bytes - will be derived into a proper 32-byte cryptographic key using HKDF.
+
+  **Default behavior (None):**
+  Like `jwt_signing_key`, this is ephemeral. However, without a valid JWT signing key, encrypted tokens are useless anyway (JWT validation fails first).
+
+  **Production:**
+  Provide an explicit secret distinct from `jwt_signing_key`. Works with `jwt_signing_key` and persistent `client_storage` - all three are required for production token persistence.
+
+  ```python
+  # Use different secrets for each key
+  jwt_signing_key="my-jwt-secret-v1"
+  token_encryption_key="my-encryption-secret-v1"
+  ```
+
+  See [HTTP Deployment - OAuth Token Security](/deployment/http#oauth-token-security).
+</ParamField>
 </Card>
 
 ### Using Built-in Providers
@@ -352,6 +395,44 @@ Finally, the client exchanges its authorization code with the proxy to receive t
 
 This entire flow is transparent to the MCP client—it experiences a standard OAuth flow with dynamic registration, unaware that a proxy is managing the complexity behind the scenes.
 
+### Token Architecture
+
+The OAuth proxy implements a **token factory pattern**: instead of directly forwarding tokens from the upstream OAuth provider, it issues its own JWT tokens to MCP clients. This maintains proper OAuth 2.0 token audience boundaries and enables better security controls.
+
+**How it works:**
+
+When an MCP client completes authorization, the proxy:
+
+1. **Receives upstream tokens** from the OAuth provider (GitHub, Google, etc.)
+2. **Encrypts and stores** these tokens using Fernet encryption (AES-128-CBC + HMAC-SHA256)
+3. **Issues FastMCP JWT tokens** to the client, signed with HS256
+
+The FastMCP JWT contains minimal claims: issuer, audience, client ID, scopes, expiration, and a unique token identifier (JTI). The JTI acts as a reference linking to the encrypted upstream token.
+
+**Token validation:**
+
+When a client makes an MCP request with its FastMCP token:
+
+1. **FastMCP validates the JWT** signature, expiration, issuer, and audience
+2. **Looks up the upstream token** using the JTI from the validated JWT
+3. **Decrypts and validates** the upstream token with the provider
+
+This two-tier validation ensures that FastMCP tokens can only be used with this server (via audience validation) while maintaining full upstream token security.
+
+**Token expiry alignment:**
+
+FastMCP token lifetimes match the upstream token lifetimes. When the upstream token expires, the FastMCP token also expires, maintaining consistent security boundaries.
+
+**Refresh tokens:**
+
+The proxy issues its own refresh tokens that map to upstream refresh tokens. When a client uses a FastMCP refresh token, the proxy refreshes the upstream token and issues a new FastMCP access token.
+
+**Key and storage configuration:**
+
+The token architecture requires cryptographic keys for JWT signing and token encryption. By default, these keys are ephemeral—generated from a random salt at startup and not persisted. This means tokens become invalid on server restart, requiring clients to re-authenticate. For development and testing, this is acceptable.
+
+For production, configure three parameters together: `jwt_signing_key` (for signing FastMCP JWTs), `token_encryption_key` (for encrypting upstream tokens at rest), and persistent `client_storage` (for storing encrypted tokens). All three are required for tokens to survive server restarts. The keys accept any secret string and derive proper cryptographic keys using HKDF. See [OAuth Token Security](/deployment/http#oauth-token-security) for complete production setup.
+
 ### PKCE Forwarding
 
 The OAuth proxy automatically handles PKCE (Proof Key for Code Exchange) when working with providers that support or require it. The proxy generates its own PKCE parameters to send upstream while separately validating the client's PKCE, ensuring end-to-end security at both layers.

src/fastmcp/server/auth/jwt_issuer.py

@@ -0,0 +1,289 @@
+"""JWT token issuance and verification for FastMCP OAuth Proxy.
+
+This module implements the token factory pattern for OAuth proxies, where the proxy
+issues its own JWT tokens to clients instead of forwarding upstream provider tokens.
+This maintains proper OAuth 2.0 token audience boundaries.
+"""
+
+from __future__ import annotations
+
+import base64
+import time
+from typing import Any
+
+from authlib.jose import JsonWebToken
+from authlib.jose.errors import JoseError
+from cryptography.fernet import Fernet
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives.kdf.hkdf import HKDF
+
+from fastmcp.utilities.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+def derive_jwt_key(upstream_secret: str, server_salt: str) -> bytes:
+    """Derive JWT signing key from upstream client secret and server salt.
+
+    Uses HKDF (RFC 5869) to derive a cryptographically secure signing key from
+    the upstream OAuth client secret combined with a server-specific salt.
+
+    Args:
+        upstream_secret: The OAuth client secret from upstream provider
+        server_salt: Random salt unique to this server instance
+
+    Returns:
+        32-byte key suitable for HS256 JWT signing
+    """
+    return HKDF(
+        algorithm=hashes.SHA256(),
+        length=32,
+        salt=f"fastmcp-jwt-signing-v1-{server_salt}".encode(),
+        info=b"HS256",
+    ).derive(upstream_secret.encode())
+
+
+def derive_encryption_key(upstream_secret: str) -> bytes:
+    """Derive Fernet encryption key from upstream client secret.
+
+    Uses HKDF to derive a cryptographically secure encryption key for
+    encrypting upstream tokens at rest.
+
+    Args:
+        upstream_secret: The OAuth client secret from upstream provider
+
+    Returns:
+        32-byte Fernet key (base64url-encoded)
+    """
+    key_material = HKDF(
+        algorithm=hashes.SHA256(),
+        length=32,
+        salt=b"fastmcp-token-encryption-v1",
+        info=b"Fernet",
+    ).derive(upstream_secret.encode())
+    return base64.urlsafe_b64encode(key_material)
+
+
+def derive_key_from_secret(secret: str | bytes, salt: str, info: bytes) -> bytes:
+    """Derive 32-byte key from user-provided secret (string or bytes).
+
+    Accepts any length input and derives a proper cryptographic key.
+    Uses HKDF to stretch weak inputs into strong keys.
+
+    Args:
+        secret: User-provided secret (any string or bytes)
+        salt: Application-specific salt string
+        info: Key purpose identifier
+
+    Returns:
+        32-byte key suitable for HS256 JWT signing or Fernet encryption
+    """
+    secret_bytes = secret.encode() if isinstance(secret, str) else secret
+    return HKDF(
+        algorithm=hashes.SHA256(),
+        length=32,
+        salt=salt.encode(),
+        info=info,
+    ).derive(secret_bytes)
+
+
+class JWTIssuer:
+    """Issues and validates FastMCP-signed JWT tokens using HS256.
+
+    This issuer creates JWT tokens for MCP clients with proper audience claims,
+    maintaining OAuth 2.0 token boundaries. Tokens are signed with HS256 using
+    a key derived from the upstream client secret.
+    """
+
+    def __init__(
+        self,
+        issuer: str,
+        audience: str,
+        signing_key: bytes,
+    ):
+        """Initialize JWT issuer.
+
+        Args:
+            issuer: Token issuer (FastMCP server base URL)
+            audience: Token audience (typically {base_url}/mcp)
+            signing_key: HS256 signing key (32 bytes)
+        """
+        self.issuer = issuer
+        self.audience = audience
+        self._signing_key = signing_key
+        self._jwt = JsonWebToken(["HS256"])
+
+    def issue_access_token(
+        self,
+        client_id: str,
+        scopes: list[str],
+        jti: str,
+        expires_in: int = 3600,
+    ) -> str:
+        """Issue a minimal FastMCP access token.
+
+        FastMCP tokens are reference tokens containing only the minimal claims
+        needed for validation and lookup. The JTI maps to the upstream token
+        which contains actual user identity and authorization data.
+
+        Args:
+            client_id: MCP client ID
+            scopes: Token scopes
+            jti: Unique token identifier (maps to upstream token)
+            expires_in: Token lifetime in seconds
+
+        Returns:
+            Signed JWT token
+        """
+        now = int(time.time())
+
+        header = {"alg": "HS256", "typ": "JWT"}
+        payload = {
+            "iss": self.issuer,
+            "aud": self.audience,
+            "client_id": client_id,
+            "scope": " ".join(scopes),
+            "exp": now + expires_in,
+            "iat": now,
+            "jti": jti,
+        }
+
+        token_bytes = self._jwt.encode(header, payload, self._signing_key)
+        token = token_bytes.decode("utf-8")
+
+        logger.debug(
+            "Issued access token for client=%s jti=%s exp=%d",
+            client_id,
+            jti[:8],
+            payload["exp"],
+        )
+
+        return token
+
+    def issue_refresh_token(
+        self,
+        client_id: str,
+        scopes: list[str],
+        jti: str,
+        expires_in: int,
+    ) -> str:
+        """Issue a minimal FastMCP refresh token.
+
+        FastMCP refresh tokens are reference tokens containing only the minimal
+        claims needed for validation and lookup. The JTI maps to the upstream
+        token which contains actual user identity and authorization data.
+
+        Args:
+            client_id: MCP client ID
+            scopes: Token scopes
+            jti: Unique token identifier (maps to upstream token)
+            expires_in: Token lifetime in seconds (should match upstream refresh expiry)
+
+        Returns:
+            Signed JWT token
+        """
+        now = int(time.time())
+
+        header = {"alg": "HS256", "typ": "JWT"}
+        payload = {
+            "iss": self.issuer,
+            "aud": self.audience,
+            "client_id": client_id,
+            "scope": " ".join(scopes),
+            "exp": now + expires_in,
+            "iat": now,
+            "jti": jti,
+            "token_use": "refresh",
+        }
+
+        token_bytes = self._jwt.encode(header, payload, self._signing_key)
+        token = token_bytes.decode("utf-8")
+
+        logger.debug(
+            "Issued refresh token for client=%s jti=%s exp=%d",
+            client_id,
+            jti[:8],
+            payload["exp"],
+        )
+
+        return token
+
+    def verify_token(self, token: str) -> dict[str, Any]:
+        """Verify and decode a FastMCP token.
+
+        Validates JWT signature, expiration, issuer, and audience.
+
+        Args:
+            token: JWT token to verify
+
+        Returns:
+            Decoded token payload
+
+        Raises:
+            JoseError: If token is invalid, expired, or has wrong claims
+        """
+        try:
+            # Decode and verify signature
+            payload = self._jwt.decode(token, self._signing_key)
+
+            # Validate expiration
+            exp = payload.get("exp")
+            if exp and exp < time.time():
+                logger.debug("Token expired")
+                raise JoseError("Token has expired")
+
+            # Validate issuer
+            if payload.get("iss") != self.issuer:
+                logger.debug("Token has invalid issuer")
+                raise JoseError("Invalid token issuer")
+
+            # Validate audience
+            if payload.get("aud") != self.audience:
+                logger.debug("Token has invalid audience")
+                raise JoseError("Invalid token audience")
+
+            logger.debug(
+                "Token verified successfully for subject=%s", payload.get("sub")
+            )
+            return payload
+
+        except JoseError as e:
+            logger.debug("Token validation failed: %s", e)
+            raise
+
+
+class TokenEncryption:
+    """Handles encryption/decryption of upstream OAuth tokens at rest."""
+
+    def __init__(self, encryption_key: bytes):
+        """Initialize token encryption.
+
+        Args:
+            encryption_key: Fernet encryption key (32 bytes, base64url-encoded)
+        """
+        self._fernet = Fernet(encryption_key)
+
+    def encrypt(self, token: str) -> bytes:
+        """Encrypt a token for storage.
+
+        Args:
+            token: Plain text token
+
+        Returns:
+            Encrypted token bytes
+        """
+        return self._fernet.encrypt(token.encode())
+
+    def decrypt(self, encrypted_token: bytes) -> str:
+        """Decrypt a token from storage.
+
+        Args:
+            encrypted_token: Encrypted token bytes
+
+        Returns:
+            Plain text token
+
+        Raises:
+            cryptography.fernet.InvalidToken: If token is corrupted or key is wrong
+        """
+        return self._fernet.decrypt(encrypted_token).decode()

src/fastmcp/server/auth/oauth_proxy.py

@@ -57,6 +57,10 @@ from starlette.responses import HTMLResponse, RedirectResponse
 from starlette.routing import Route
 
 from fastmcp.server.auth.auth import OAuthProvider, TokenVerifier
+from fastmcp.server.auth.jwt_issuer import (
+    JWTIssuer,
+    TokenEncryption,
+)
 from fastmcp.server.auth.redirect_validation import (
     validate_redirect_uri,
 )
@@ -134,6 +138,39 @@ class ClientCode(BaseModel):
     created_at: float
 
 
+class UpstreamTokenSet(BaseModel):
+    """Stored upstream OAuth tokens from identity provider.
+
+    These tokens are obtained from the upstream provider (Google, GitHub, etc.)
+    and are stored encrypted at rest. They are never exposed to MCP clients.
+    """
+
+    upstream_token_id: str  # Unique ID for this token set
+    access_token: bytes  # Encrypted upstream access token
+    refresh_token: bytes | None  # Encrypted upstream refresh token
+    refresh_token_expires_at: (
+        float | None
+    )  # Unix timestamp when refresh token expires (if known)
+    expires_at: float  # Unix timestamp when access token expires
+    token_type: str  # Usually "Bearer"
+    scope: str  # Space-separated scopes
+    client_id: str  # MCP client this is bound to
+    created_at: float  # Unix timestamp
+    raw_token_data: dict[str, Any] = Field(default_factory=dict)  # Full token response
+
+
+class JTIMapping(BaseModel):
+    """Maps FastMCP token JTI to upstream token ID.
+
+    This allows stateless JWT validation while still being able to look up
+    the corresponding upstream token when tools need to access upstream APIs.
+    """
+
+    jti: str  # JWT ID from FastMCP-issued token
+    upstream_token_id: str  # References UpstreamTokenSet
+    created_at: float  # Unix timestamp
+
+
 class ProxyDCRClient(OAuthClientInformationFull):
     """Client for DCR proxy with configurable redirect URI validation.
 
@@ -474,6 +511,10 @@ class OAuthProxy(OAuthProvider):
         extra_token_params: dict[str, str] | None = None,
         # Client storage
         client_storage: AsyncKeyValue | None = None,
+        # JWT signing key (optional, ephemeral if not provided)
+        jwt_signing_key: str | bytes | None = None,
+        # Token encryption key (optional, ephemeral if not provided)
+        token_encryption_key: str | bytes | None = None,
     ):
         """Initialize the OAuth proxy provider.
 
@@ -508,6 +549,12 @@ class OAuthProxy(OAuthProvider):
             extra_token_params: Additional parameters to forward to the upstream token endpoint.
                 Useful for provider-specific parameters during token exchange.
             client_storage: An AsyncKeyValue-compatible store for client registrations, registrations are stored in memory if not provided
+            jwt_signing_key: Optional secret for signing FastMCP JWT tokens (accepts any string or bytes).
+                Default: ephemeral (random salt at startup, won't survive restart).
+                Production: provide explicit key from environment variable.
+            token_encryption_key: Optional secret for encrypting upstream tokens at rest (accepts any string or bytes).
+                Default: ephemeral (random salt at startup, won't survive restart).
+                Production: provide explicit key from environment variable.
         """
         # Always enable DCR since we implement it locally for MCP clients
         client_registration_options = ClientRegistrationOptions(
@@ -579,8 +626,10 @@ class OAuthProxy(OAuthProvider):
         # Warn if using MemoryStore in production
         if isinstance(client_storage, MemoryStore):
             logger.warning(
-                "Using in-memory storage - all OAuth state will be lost on restart. "
-                "For production, configure persistent storage (Redis, PostgreSQL, etc.)."
+                "Using in-memory storage - all OAuth state (clients, tokens) will be lost on restart. "
+                "Additionally, without explicit jwt_signing_key and token_encryption_key, "
+                "keys are ephemeral and tokens won't survive restart even with persistent storage. "
+                "For production, configure persistent storage AND explicit keys."
             )
 
         # Cache HTTPS check to avoid repeated logging
@@ -613,6 +662,29 @@ class OAuthProxy(OAuthProvider):
             raise_on_validation_error=True,
         )
 
+        # Storage for upstream tokens (encrypted at rest)
+        self._upstream_token_store = PydanticAdapter[UpstreamTokenSet](
+            key_value=self._client_storage,
+            pydantic_model=UpstreamTokenSet,
+            default_collection="mcp-upstream-tokens",
+            raise_on_validation_error=True,
+        )
+
+        # Storage for JTI mappings (FastMCP token -> upstream token)
+        self._jti_mapping_store = PydanticAdapter[JTIMapping](
+            key_value=self._client_storage,
+            pydantic_model=JTIMapping,
+            default_collection="mcp-jti-mappings",
+            raise_on_validation_error=True,
+        )
+
+        # JWT issuer and encryption (initialized lazily on first use)
+        self._custom_jwt_key = jwt_signing_key
+        self._custom_encryption_key = token_encryption_key
+        self._jwt_issuer: JWTIssuer | None = None
+        self._token_encryption: TokenEncryption | None = None
+        self._jwt_initialized = False
+
         # Local state for token bookkeeping only (no client caching)
         self._access_tokens: dict[str, AccessToken] = {}
         self._refresh_tokens: dict[str, RefreshToken] = {}
@@ -648,6 +720,87 @@ class OAuthProxy(OAuthProvider):
 
         return code_verifier, code_challenge
 
+    # -------------------------------------------------------------------------
+    # JWT Token Factory Initialization
+    # -------------------------------------------------------------------------
+
+    async def _ensure_jwt_initialized(self) -> None:
+        """Initialize JWT issuer and token encryption (lazy initialization).
+
+        Key derivation strategy:
+        - Default: Generate random salt at startup, derive ephemeral keys
+          → Keys change on restart, all tokens become invalid
+          → Perfect for development/testing where re-auth is acceptable
+
+        - Production: User provides explicit keys via parameters
+          → Keys stable across restarts when combined with persistent storage
+          → Tokens survive restart, seamless client reconnection
+        """
+        if self._jwt_initialized:
+            return
+
+        # Generate random salt for this server instance (NOT persisted)
+        server_salt = secrets.token_urlsafe(32)
+
+        # Derive or use custom JWT signing key
+        from fastmcp.server.auth.jwt_issuer import derive_key_from_secret
+
+        if self._custom_jwt_key:
+            jwt_key = derive_key_from_secret(
+                secret=self._custom_jwt_key,
+                salt="fastmcp-jwt-signing-v1",
+                info=b"HS256",
+            )
+            logger.info("Using explicit JWT signing key (will survive restarts)")
+        else:
+            # Ephemeral key from random salt + upstream secret
+            upstream_secret = self._upstream_client_secret.get_secret_value()
+            jwt_key = derive_key_from_secret(
+                secret=upstream_secret,
+                salt=f"fastmcp-jwt-signing-v1-{server_salt}",
+                info=b"HS256",
+            )
+            logger.info(
+                "Using ephemeral JWT signing key - tokens will NOT survive server restart. "
+                "For production, provide explicit jwt_signing_key parameter."
+            )
+
+        # Initialize JWT issuer
+        issuer = str(self.base_url)
+        audience = f"{str(self.base_url).rstrip('/')}/mcp"
+        self._jwt_issuer = JWTIssuer(
+            issuer=issuer,
+            audience=audience,
+            signing_key=jwt_key,
+        )
+
+        # Derive or use custom encryption key
+        if self._custom_encryption_key:
+            encryption_key = derive_key_from_secret(
+                secret=self._custom_encryption_key,
+                salt="fastmcp-token-encryption-v1",
+                info=b"Fernet",
+            )
+            # Fernet needs base64url-encoded key
+            encryption_key = base64.urlsafe_b64encode(encryption_key)
+            logger.info("Using explicit token encryption key (will survive restarts)")
+        else:
+            # Ephemeral key from random salt + upstream secret
+            upstream_secret = self._upstream_client_secret.get_secret_value()
+            key_material = derive_key_from_secret(
+                secret=upstream_secret,
+                salt=f"fastmcp-token-encryption-v1-{server_salt}",
+                info=b"Fernet",
+            )
+            encryption_key = base64.urlsafe_b64encode(key_material)
+            logger.info(
+                "Using ephemeral token encryption key - encrypted tokens will NOT survive server restart. "
+                "For production, provide explicit token_encryption_key parameter."
+            )
+
+        self._token_encryption = TokenEncryption(encryption_key)
+        self._jwt_initialized = True
+
     # -------------------------------------------------------------------------
     # Client Registration (Local Implementation)
     # -------------------------------------------------------------------------
@@ -753,6 +906,7 @@ class OAuthProxy(OAuthProvider):
                 resource=getattr(params, "resource", None),
                 proxy_code_verifier=proxy_code_verifier,
             ),
+            ttl=15 * 60,  # Auto-expire after 15 minutes
         )
 
         consent_url = f"{str(self.base_url).rstrip('/')}/consent?txn_id={txn_id}"
@@ -816,11 +970,22 @@ class OAuthProxy(OAuthProvider):
         client: OAuthClientInformationFull,
         authorization_code: AuthorizationCode,
     ) -> OAuthToken:
-        """Exchange authorization code for stored IdP tokens.
+        """Exchange authorization code for FastMCP-issued tokens.
 
-        For the DCR-compliant proxy flow, we return the IdP tokens that were obtained
-        during the IdP callback exchange. PKCE validation is handled by the MCP framework.
+        Implements the token factory pattern:
+        1. Retrieves upstream tokens from stored authorization code
+        2. Extracts user identity from upstream token
+        3. Encrypts and stores upstream tokens
+        4. Issues FastMCP-signed JWT tokens
+        5. Returns FastMCP tokens (NOT upstream tokens)
+
+        PKCE validation is handled by the MCP framework before this method is called.
         """
+        # Ensure JWT issuer is initialized
+        await self._ensure_jwt_initialized()
+        assert self._jwt_issuer is not None
+        assert self._token_encryption is not None
+
         # Look up stored code data
         code_model = await self._code_store.get(key=authorization_code.code)
         if not code_model:
@@ -830,49 +995,141 @@ class OAuthProxy(OAuthProvider):
             )
             raise TokenError("invalid_grant", "Authorization code not found")
 
-        # Get stored IdP tokens
+        # Get stored upstream tokens
         idp_tokens = code_model.idp_tokens
 
         # Clean up client code (one-time use)
         await self._code_store.delete(key=authorization_code.code)
 
-        # Extract token information for local tracking
-        access_token_value = idp_tokens["access_token"]
-        refresh_token_value = idp_tokens.get("refresh_token")
+        # Generate IDs for token storage
+        upstream_token_id = secrets.token_urlsafe(32)
+        access_jti = secrets.token_urlsafe(32)
+        refresh_jti = (
+            secrets.token_urlsafe(32) if idp_tokens.get("refresh_token") else None
+        )
+
+        # Calculate token expiry times
         expires_in = int(
             idp_tokens.get("expires_in", DEFAULT_ACCESS_TOKEN_EXPIRY_SECONDS)
         )
-        expires_at = int(time.time() + expires_in)
 
-        # Store access token locally for tracking
-        access_token = AccessToken(
-            token=access_token_value,
+        # Calculate refresh token expiry if provided by upstream
+        # Some providers include refresh_expires_in, some don't
+        refresh_expires_in = None
+        refresh_token_expires_at = None
+        if idp_tokens.get("refresh_token"):
+            if "refresh_expires_in" in idp_tokens:
+                refresh_expires_in = int(idp_tokens["refresh_expires_in"])
+                refresh_token_expires_at = time.time() + refresh_expires_in
+                logger.debug(
+                    "Upstream refresh token expires in %d seconds", refresh_expires_in
+                )
+            else:
+                # Default to 30 days if upstream doesn't specify
+                # This is conservative - most providers use longer expiry
+                refresh_expires_in = 60 * 60 * 24 * 30  # 30 days
+                refresh_token_expires_at = time.time() + refresh_expires_in
+                logger.debug(
+                    "Upstream refresh token expiry unknown, using 30-day default"
+                )
+
+        # Encrypt and store upstream tokens
+        upstream_token_set = UpstreamTokenSet(
+            upstream_token_id=upstream_token_id,
+            access_token=self._token_encryption.encrypt(idp_tokens["access_token"]),
+            refresh_token=self._token_encryption.encrypt(idp_tokens["refresh_token"])
+            if idp_tokens.get("refresh_token")
+            else None,
+            refresh_token_expires_at=refresh_token_expires_at,
+            expires_at=time.time() + expires_in,
+            token_type=idp_tokens.get("token_type", "Bearer"),
+            scope=" ".join(authorization_code.scopes),
+            client_id=client.client_id,
+            created_at=time.time(),
+            raw_token_data=idp_tokens,
+        )
+        await self._upstream_token_store.put(
+            key=upstream_token_id,
+            value=upstream_token_set,
+            ttl=expires_in,  # Auto-expire when access token expires
+        )
+        logger.debug("Stored encrypted upstream tokens (jti=%s)", access_jti[:8])
+
+        # Issue minimal FastMCP access token (just a reference via JTI)
+        fastmcp_access_token = self._jwt_issuer.issue_access_token(
             client_id=client.client_id,
             scopes=authorization_code.scopes,
-            expires_at=expires_at,
+            jti=access_jti,
+            expires_in=expires_in,
         )
-        self._access_tokens[access_token_value] = access_token
 
-        # Store refresh token if provided
-        if refresh_token_value:
-            refresh_token = RefreshToken(
-                token=refresh_token_value,
+        # Issue minimal FastMCP refresh token if upstream provided one
+        # Use upstream refresh token expiry to align lifetimes
+        fastmcp_refresh_token = None
+        if refresh_jti and refresh_expires_in:
+            fastmcp_refresh_token = self._jwt_issuer.issue_refresh_token(
                 client_id=client.client_id,
                 scopes=authorization_code.scopes,
-                expires_at=None,  # Refresh tokens typically don't expire
+                jti=refresh_jti,
+                expires_in=refresh_expires_in,
             )
-            self._refresh_tokens[refresh_token_value] = refresh_token
 
-            # Maintain token relationships for cleanup
-            self._access_to_refresh[access_token_value] = refresh_token_value
-            self._refresh_to_access[refresh_token_value] = access_token_value
+        # Store JTI mappings
+        await self._jti_mapping_store.put(
+            key=access_jti,
+            value=JTIMapping(
+                jti=access_jti,
+                upstream_token_id=upstream_token_id,
+                created_at=time.time(),
+            ),
+            ttl=expires_in,  # Auto-expire with access token
+        )
+        if refresh_jti:
+            await self._jti_mapping_store.put(
+                key=refresh_jti,
+                value=JTIMapping(
+                    jti=refresh_jti,
+                    upstream_token_id=upstream_token_id,
+                    created_at=time.time(),
+                ),
+                ttl=60 * 60 * 24 * 30,  # Auto-expire with refresh token (30 days)
+            )
 
-        logger.debug(
-            "Successfully exchanged client code for stored IdP tokens (client: %s)",
-            client.client_id,
+        # Store FastMCP access token for MCP framework validation
+        self._access_tokens[fastmcp_access_token] = AccessToken(
+            token=fastmcp_access_token,
+            client_id=client.client_id,
+            scopes=authorization_code.scopes,
+            expires_at=int(time.time() + expires_in),
         )
 
-        return OAuthToken(**idp_tokens)  # type: ignore[arg-type]
+        # Store FastMCP refresh token if provided
+        if fastmcp_refresh_token:
+            self._refresh_tokens[fastmcp_refresh_token] = RefreshToken(
+                token=fastmcp_refresh_token,
+                client_id=client.client_id,
+                scopes=authorization_code.scopes,
+                expires_at=None,
+            )
+            # Maintain token relationships for cleanup
+            self._access_to_refresh[fastmcp_access_token] = fastmcp_refresh_token
+            self._refresh_to_access[fastmcp_refresh_token] = fastmcp_access_token
+
+        logger.debug(
+            "Issued FastMCP tokens for client=%s (access_jti=%s, refresh_jti=%s)",
+            client.client_id,
+            access_jti[:8],
+            refresh_jti[:8] if refresh_jti else "none",
+        )
+
+        # Return FastMCP-issued tokens (NOT upstream tokens!)
+        return OAuthToken(
+            access_token=fastmcp_access_token,
+            token_type="Bearer",
+            expires_in=expires_in,
+            refresh_token=fastmcp_refresh_token,
+            scope=" ".join(authorization_code.scopes),
+        )
 
     # -------------------------------------------------------------------------
     # Refresh Token Flow
@@ -892,9 +1149,54 @@ class OAuthProxy(OAuthProvider):
         refresh_token: RefreshToken,
         scopes: list[str],
     ) -> OAuthToken:
-        """Exchange refresh token for new access token using authlib."""
+        """Exchange FastMCP refresh token for new FastMCP access token.
 
-        # Use authlib's AsyncOAuth2Client for refresh token exchange
+        Implements two-tier refresh:
+        1. Verify FastMCP refresh token
+        2. Look up upstream token via JTI mapping
+        3. Refresh upstream token with upstream provider
+        4. Update stored upstream token
+        5. Issue new FastMCP access token
+        6. Keep same FastMCP refresh token (unless upstream rotates)
+        """
+        # Ensure JWT issuer is initialized
+        await self._ensure_jwt_initialized()
+        assert self._jwt_issuer is not None
+        assert self._token_encryption is not None
+
+        # Verify FastMCP refresh token
+        try:
+            refresh_payload = self._jwt_issuer.verify_token(refresh_token.token)
+            refresh_jti = refresh_payload["jti"]
+        except Exception as e:
+            logger.debug("FastMCP refresh token validation failed: %s", e)
+            raise TokenError("invalid_grant", "Invalid refresh token") from e
+
+        # Look up upstream token via JTI mapping
+        jti_mapping = await self._jti_mapping_store.get(key=refresh_jti)
+        if not jti_mapping:
+            logger.error("JTI mapping not found for refresh token: %s", refresh_jti[:8])
+            raise TokenError("invalid_grant", "Refresh token mapping not found")
+
+        upstream_token_set = await self._upstream_token_store.get(
+            key=jti_mapping.upstream_token_id
+        )
+        if not upstream_token_set:
+            logger.error(
+                "Upstream token set not found: %s", jti_mapping.upstream_token_id[:8]
+            )
+            raise TokenError("invalid_grant", "Upstream token not found")
+
+        # Decrypt upstream refresh token
+        if not upstream_token_set.refresh_token:
+            logger.error("No upstream refresh token available")
+            raise TokenError("invalid_grant", "Refresh not supported for this token")
+
+        upstream_refresh_token = self._token_encryption.decrypt(
+            upstream_token_set.refresh_token
+        )
+
+        # Refresh upstream token using authlib
         oauth_client = AsyncOAuth2Client(
             client_id=self._upstream_client_id,
             client_secret=self._upstream_client_secret.get_secret_value(),
@@ -903,76 +1205,217 @@ class OAuthProxy(OAuthProvider):
         )
 
         try:
-            logger.debug("Using authlib to refresh token from upstream")
-
-            # Let authlib handle the refresh token exchange
+            logger.debug("Refreshing upstream token (jti=%s)", refresh_jti[:8])
             token_response: dict[str, Any] = await oauth_client.refresh_token(  # type: ignore[misc]
                 url=self._upstream_token_endpoint,
-                refresh_token=refresh_token.token,
+                refresh_token=upstream_refresh_token,
                 scope=" ".join(scopes) if scopes else None,
             )
-
-            logger.debug(
-                "Successfully refreshed access token via authlib (client: %s)",
-                client.client_id,
-            )
-
+            logger.debug("Successfully refreshed upstream token")
         except Exception as e:
-            logger.error("Authlib refresh token exchange failed: %s", e)
-            raise TokenError(
-                "invalid_grant", f"Upstream refresh token exchange failed: {e}"
-            ) from e
+            logger.error("Upstream token refresh failed: %s", e)
+            raise TokenError("invalid_grant", f"Upstream refresh failed: {e}") from e
 
-        # Update local token storage
-        new_access_token = token_response["access_token"]
-        expires_in = int(
+        # Update stored upstream token
+        new_expires_in = int(
             token_response.get("expires_in", DEFAULT_ACCESS_TOKEN_EXPIRY_SECONDS)
         )
+        upstream_token_set.access_token = self._token_encryption.encrypt(
+            token_response["access_token"]
+        )
+        upstream_token_set.expires_at = time.time() + new_expires_in
 
-        self._access_tokens[new_access_token] = AccessToken(
-            token=new_access_token,
-            client_id=client.client_id,
-            scopes=scopes,
-            expires_at=int(time.time() + expires_in),
+        # Handle upstream refresh token rotation and expiry
+        new_refresh_expires_in = None
+        if new_upstream_refresh := token_response.get("refresh_token"):
+            if new_upstream_refresh != upstream_refresh_token:
+                upstream_token_set.refresh_token = self._token_encryption.encrypt(
+                    new_upstream_refresh
+                )
+                logger.debug("Upstream refresh token rotated")
+
+            # Update refresh token expiry if provided
+            if "refresh_expires_in" in token_response:
+                new_refresh_expires_in = int(token_response["refresh_expires_in"])
+                upstream_token_set.refresh_token_expires_at = (
+                    time.time() + new_refresh_expires_in
+                )
+                logger.debug(
+                    "Upstream refresh token expires in %d seconds",
+                    new_refresh_expires_in,
+                )
+            elif upstream_token_set.refresh_token_expires_at:
+                # Keep existing expiry if upstream doesn't provide new one
+                new_refresh_expires_in = int(
+                    upstream_token_set.refresh_token_expires_at - time.time()
+                )
+            else:
+                # Default to 30 days if unknown
+                new_refresh_expires_in = 60 * 60 * 24 * 30
+                upstream_token_set.refresh_token_expires_at = (
+                    time.time() + new_refresh_expires_in
+                )
+
+        upstream_token_set.raw_token_data = token_response
+        await self._upstream_token_store.put(
+            key=upstream_token_set.upstream_token_id,
+            value=upstream_token_set,
+            ttl=new_expires_in,  # Auto-expire when refreshed access token expires
         )
 
-        # Handle refresh token rotation if new one provided
-        if new_refresh_token := token_response.get("refresh_token"):
-            if new_refresh_token != refresh_token.token:
-                # Remove old refresh token
-                self._refresh_tokens.pop(refresh_token.token, None)
-                old_access = self._refresh_to_access.pop(refresh_token.token, None)
-                if old_access:
-                    self._access_to_refresh.pop(old_access, None)
+        # Issue new minimal FastMCP access token (just a reference via JTI)
+        new_access_jti = secrets.token_urlsafe(32)
+        new_fastmcp_access = self._jwt_issuer.issue_access_token(
+            client_id=client.client_id,
+            scopes=scopes,
+            jti=new_access_jti,
+            expires_in=new_expires_in,
+        )
 
-                # Store new refresh token
-                self._refresh_tokens[new_refresh_token] = RefreshToken(
-                    token=new_refresh_token,
-                    client_id=client.client_id,
-                    scopes=scopes,
-                    expires_at=None,
-                )
-                self._access_to_refresh[new_access_token] = new_refresh_token
-                self._refresh_to_access[new_refresh_token] = new_access_token
+        # Store new access token JTI mapping
+        await self._jti_mapping_store.put(
+            key=new_access_jti,
+            value=JTIMapping(
+                jti=new_access_jti,
+                upstream_token_id=upstream_token_set.upstream_token_id,
+                created_at=time.time(),
+            ),
+            ttl=new_expires_in,  # Auto-expire with refreshed access token
+        )
 
-        return OAuthToken(**token_response)  # type: ignore[arg-type]
+        # Issue NEW minimal FastMCP refresh token (rotation for security)
+        # Use upstream refresh token expiry to align lifetimes
+        new_refresh_jti = secrets.token_urlsafe(32)
+        new_fastmcp_refresh = self._jwt_issuer.issue_refresh_token(
+            client_id=client.client_id,
+            scopes=scopes,
+            jti=new_refresh_jti,
+            expires_in=new_refresh_expires_in
+            or 60 * 60 * 24 * 30,  # Fallback to 30 days
+        )
+
+        # Store new refresh token JTI mapping with aligned expiry
+        refresh_ttl = new_refresh_expires_in or 60 * 60 * 24 * 30
+        await self._jti_mapping_store.put(
+            key=new_refresh_jti,
+            value=JTIMapping(
+                jti=new_refresh_jti,
+                upstream_token_id=upstream_token_set.upstream_token_id,
+                created_at=time.time(),
+            ),
+            ttl=refresh_ttl,  # Align with upstream refresh token expiry
+        )
+
+        # Invalidate old refresh token (refresh token rotation - enforces one-time use)
+        await self._jti_mapping_store.delete(key=refresh_jti)
+        logger.debug(
+            "Rotated refresh token (old JTI invalidated - one-time use enforced)"
+        )
+
+        # Update local token tracking
+        self._access_tokens[new_fastmcp_access] = AccessToken(
+            token=new_fastmcp_access,
+            client_id=client.client_id,
+            scopes=scopes,
+            expires_at=int(time.time() + new_expires_in),
+        )
+        self._refresh_tokens[new_fastmcp_refresh] = RefreshToken(
+            token=new_fastmcp_refresh,
+            client_id=client.client_id,
+            scopes=scopes,
+            expires_at=None,
+        )
+
+        # Update token relationship mappings
+        self._access_to_refresh[new_fastmcp_access] = new_fastmcp_refresh
+        self._refresh_to_access[new_fastmcp_refresh] = new_fastmcp_access
+
+        # Clean up old token from in-memory tracking
+        self._refresh_tokens.pop(refresh_token.token, None)
+        old_access = self._refresh_to_access.pop(refresh_token.token, None)
+        if old_access:
+            self._access_tokens.pop(old_access, None)
+            self._access_to_refresh.pop(old_access, None)
+
+        logger.info(
+            "Issued new FastMCP tokens (rotated refresh) for client=%s (access_jti=%s, refresh_jti=%s)",
+            client.client_id,
+            new_access_jti[:8],
+            new_refresh_jti[:8],
+        )
+
+        # Return new FastMCP tokens (both access AND refresh are new)
+        return OAuthToken(
+            access_token=new_fastmcp_access,
+            token_type="Bearer",
+            expires_in=new_expires_in,
+            refresh_token=new_fastmcp_refresh,  # NEW refresh token (rotated)
+            scope=" ".join(scopes),
+        )
 
     # -------------------------------------------------------------------------
     # Token Validation
     # -------------------------------------------------------------------------
 
     async def load_access_token(self, token: str) -> AccessToken | None:
-        """Validate access token using upstream JWKS.
+        """Validate FastMCP JWT by swapping for upstream token.
 
-        Delegates to the JWT verifier which handles signature validation,
-        expiration checking, and claims validation using the upstream JWKS.
+        This implements the token swap pattern:
+        1. Verify FastMCP JWT signature (proves it's our token)
+        2. Look up upstream token via JTI mapping
+        3. Decrypt upstream token
+        4. Validate upstream token with provider (GitHub API, JWT validation, etc.)
+        5. Return upstream validation result
+
+        The FastMCP JWT is a reference token - all authorization data comes
+        from validating the upstream token via the TokenVerifier.
         """
-        result = await self._token_validator.verify_token(token)
-        if result:
-            logger.debug("Token validated successfully")
-        else:
-            logger.debug("Token validation failed")
-        return result
+        # Ensure JWT issuer and encryption are initialized
+        await self._ensure_jwt_initialized()
+        assert self._jwt_issuer is not None
+        assert self._token_encryption is not None
+
+        try:
+            # 1. Verify FastMCP JWT signature and claims
+            payload = self._jwt_issuer.verify_token(token)
+            jti = payload["jti"]
+
+            # 2. Look up upstream token via JTI mapping
+            jti_mapping = await self._jti_mapping_store.get(key=jti)
+            if not jti_mapping:
+                logger.debug("JTI mapping not found: %s", jti)
+                return None
+
+            upstream_token_set = await self._upstream_token_store.get(
+                key=jti_mapping.upstream_token_id
+            )
+            if not upstream_token_set:
+                logger.debug(
+                    "Upstream token not found: %s", jti_mapping.upstream_token_id
+                )
+                return None
+
+            # 3. Decrypt upstream token
+            upstream_token = self._token_encryption.decrypt(
+                upstream_token_set.access_token
+            )
+
+            # 4. Validate with upstream provider (delegated to TokenVerifier)
+            # This calls the real token validator (GitHub API, JWKS, etc.)
+            validated = await self._token_validator.verify_token(upstream_token)
+
+            if not validated:
+                logger.debug("Upstream token validation failed")
+                return None
+
+            logger.debug(
+                "Token swap successful for JTI=%s (upstream validated)", jti[:8]
+            )
+            return validated
+
+        except Exception as e:
+            logger.debug("Token swap validation failed: %s", e)
+            return None
 
     # -------------------------------------------------------------------------
     # Token Revocation
@@ -1220,6 +1663,7 @@ class OAuthProxy(OAuthProvider):
                     expires_at=code_expires_at,
                     created_at=time.time(),
                 ),
+                ttl=DEFAULT_AUTH_CODE_EXPIRY_SECONDS,  # Auto-expire after 5 minutes
             )
 
             # Clean up transaction
@@ -1439,7 +1883,9 @@ class OAuthProxy(OAuthProvider):
         # Update transaction with CSRF token
         txn_model.csrf_token = csrf_token
         txn_model.csrf_expires_at = csrf_expires_at
-        await self._transaction_store.put(key=txn_id, value=txn_model)
+        await self._transaction_store.put(
+            key=txn_id, value=txn_model, ttl=15 * 60
+        )  # Auto-expire after 15 minutes
 
         # Update dict for use in HTML generation
         txn["csrf_token"] = csrf_token

tests/server/auth/test_jwt_issuer.py

@@ -0,0 +1,297 @@
+"""Unit tests for JWT issuer and token encryption."""
+
+import base64
+import time
+
+import pytest
+from authlib.jose.errors import JoseError
+
+from fastmcp.server.auth.jwt_issuer import (
+    JWTIssuer,
+    TokenEncryption,
+    derive_encryption_key,
+    derive_jwt_key,
+)
+
+
+class TestKeyDerivation:
+    """Tests for HKDF key derivation functions."""
+
+    def test_derive_jwt_key_produces_32_bytes(self):
+        """Test that JWT key derivation produces 32-byte key."""
+        key = derive_jwt_key("test-secret", "test-salt")
+        assert len(key) == 32
+        assert isinstance(key, bytes)
+
+    def test_derive_jwt_key_with_different_secrets_produces_different_keys(self):
+        """Test that different secrets produce different keys."""
+        key1 = derive_jwt_key("secret1", "salt")
+        key2 = derive_jwt_key("secret2", "salt")
+        assert key1 != key2
+
+    def test_derive_jwt_key_with_different_salts_produces_different_keys(self):
+        """Test that different salts produce different keys."""
+        key1 = derive_jwt_key("secret", "salt1")
+        key2 = derive_jwt_key("secret", "salt2")
+        assert key1 != key2
+
+    def test_derive_jwt_key_is_deterministic(self):
+        """Test that same inputs always produce same key."""
+        key1 = derive_jwt_key("secret", "salt")
+        key2 = derive_jwt_key("secret", "salt")
+        assert key1 == key2
+
+    def test_derive_encryption_key_produces_base64_key(self):
+        """Test that encryption key is base64url-encoded."""
+        key = derive_encryption_key("test-secret")
+        assert len(key) == 44  # 32 bytes base64url-encoded = 44 chars
+        assert isinstance(key, bytes)
+        # Should be valid base64url (no padding issues)
+        import base64
+
+        decoded = base64.urlsafe_b64decode(key)
+        assert len(decoded) == 32
+
+    def test_derive_encryption_key_with_different_secrets_produces_different_keys(
+        self,
+    ):
+        """Test that different secrets produce different encryption keys."""
+        key1 = derive_encryption_key("secret1")
+        key2 = derive_encryption_key("secret2")
+        assert key1 != key2
+
+    def test_derive_encryption_key_is_deterministic(self):
+        """Test that same input always produces same encryption key."""
+        key1 = derive_encryption_key("secret")
+        key2 = derive_encryption_key("secret")
+        assert key1 == key2
+
+    def test_jwt_and_encryption_keys_are_different(self):
+        """Test that JWT and encryption keys derived from same secret are different."""
+        jwt_key = derive_jwt_key("secret", "salt")
+        enc_key_raw = base64.urlsafe_b64decode(derive_encryption_key("secret"))
+        assert jwt_key != enc_key_raw
+
+
+class TestJWTIssuer:
+    """Tests for JWT token issuance and verification."""
+
+    @pytest.fixture
+    def issuer(self):
+        """Create a JWT issuer for testing."""
+        signing_key = derive_jwt_key("test-secret", "test-salt")
+        return JWTIssuer(
+            issuer="https://test-server.com",
+            audience="https://test-server.com/mcp",
+            signing_key=signing_key,
+        )
+
+    def test_issue_access_token_creates_valid_jwt(self, issuer):
+        """Test that access token is a minimal JWT with correct structure."""
+        token = issuer.issue_access_token(
+            client_id="client-abc",
+            scopes=["read", "write"],
+            jti="token-id-123",
+            expires_in=3600,
+        )
+
+        # Should be a JWT with 3 segments
+        assert len(token.split(".")) == 3
+
+        # Should be verifiable
+        payload = issuer.verify_token(token)
+        # Minimal token should only have required claims
+        assert payload["client_id"] == "client-abc"
+        assert payload["scope"] == "read write"
+        assert payload["jti"] == "token-id-123"
+        assert payload["iss"] == "https://test-server.com"
+        assert payload["aud"] == "https://test-server.com/mcp"
+        # Should NOT have user identity claims
+        assert "sub" not in payload
+        assert "azp" not in payload
+
+    def test_minimal_token_has_no_user_identity(self, issuer):
+        """Test that minimal tokens contain no user identity or custom claims."""
+        token = issuer.issue_access_token(
+            client_id="client-abc",
+            scopes=["read"],
+            jti="token-id",
+            expires_in=3600,
+        )
+
+        payload = issuer.verify_token(token)
+        # Should only have minimal required claims
+        assert "sub" not in payload
+        assert "azp" not in payload
+        assert "groups" not in payload
+        assert "roles" not in payload
+        assert "email" not in payload
+        # Should have exactly these claims
+        expected_keys = {"iss", "aud", "client_id", "scope", "exp", "iat", "jti"}
+        assert set(payload.keys()) == expected_keys
+
+    def test_issue_refresh_token_creates_valid_jwt(self, issuer):
+        """Test that refresh token is a minimal JWT with token_use claim."""
+        token = issuer.issue_refresh_token(
+            client_id="client-abc",
+            scopes=["read"],
+            jti="refresh-token-id",
+            expires_in=60 * 60 * 24 * 30,  # 30 days
+        )
+
+        payload = issuer.verify_token(token)
+        assert payload["client_id"] == "client-abc"
+        assert payload["token_use"] == "refresh"
+        assert payload["jti"] == "refresh-token-id"
+        # Should NOT have user identity
+        assert "sub" not in payload
+
+    def test_verify_token_validates_signature(self, issuer):
+        """Test that token verification fails with wrong signing key."""
+        # Create token with one issuer
+        token = issuer.issue_access_token(
+            client_id="client-abc",
+            scopes=["read"],
+            jti="token-id",
+        )
+
+        # Try to verify with different issuer (different key)
+        other_key = derive_jwt_key("different-secret", "different-salt")
+        other_issuer = JWTIssuer(
+            issuer="https://test-server.com",
+            audience="https://test-server.com/mcp",
+            signing_key=other_key,
+        )
+
+        with pytest.raises(JoseError):
+            other_issuer.verify_token(token)
+
+    def test_verify_token_validates_expiration(self, issuer):
+        """Test that expired tokens are rejected."""
+        # Create token that expires in 1 second
+        token = issuer.issue_access_token(
+            client_id="client-abc",
+            scopes=["read"],
+            jti="token-id",
+            expires_in=1,
+        )
+
+        # Should be valid immediately
+        payload = issuer.verify_token(token)
+        assert payload["client_id"] == "client-abc"
+
+        # Wait for token to expire
+        time.sleep(1.1)
+
+        # Should be rejected
+        with pytest.raises(JoseError, match="expired"):
+            issuer.verify_token(token)
+
+    def test_verify_token_validates_issuer(self, issuer):
+        """Test that tokens from different issuers are rejected."""
+        token = issuer.issue_access_token(
+            client_id="client-abc",
+            scopes=["read"],
+            jti="token-id",
+        )
+
+        # Create issuer with different issuer URL but same key
+        other_issuer = JWTIssuer(
+            issuer="https://other-server.com",  # Different issuer
+            audience="https://test-server.com/mcp",
+            signing_key=issuer._signing_key,  # Same key
+        )
+
+        with pytest.raises(JoseError, match="issuer"):
+            other_issuer.verify_token(token)
+
+    def test_verify_token_validates_audience(self, issuer):
+        """Test that tokens for different audiences are rejected."""
+        token = issuer.issue_access_token(
+            client_id="client-abc",
+            scopes=["read"],
+            jti="token-id",
+        )
+
+        # Create issuer with different audience but same key
+        other_issuer = JWTIssuer(
+            issuer="https://test-server.com",
+            audience="https://other-server.com/mcp",  # Different audience
+            signing_key=issuer._signing_key,  # Same key
+        )
+
+        with pytest.raises(JoseError, match="audience"):
+            other_issuer.verify_token(token)
+
+    def test_verify_token_rejects_malformed_tokens(self, issuer):
+        """Test that malformed tokens are rejected."""
+        with pytest.raises(JoseError):
+            issuer.verify_token("not-a-jwt")
+
+        with pytest.raises(JoseError):
+            issuer.verify_token("too.few.segments")
+
+        with pytest.raises(JoseError):
+            issuer.verify_token("header.payload")  # Missing signature
+
+
+class TestTokenEncryption:
+    """Tests for token encryption/decryption."""
+
+    @pytest.fixture
+    def encryption(self):
+        """Create token encryption instance for testing."""
+        key = derive_encryption_key("test-secret")
+        return TokenEncryption(key)
+
+    def test_encrypt_decrypt_roundtrip(self, encryption):
+        """Test that encryption and decryption work correctly."""
+        plaintext = "sensitive-token-value"
+        encrypted = encryption.encrypt(plaintext)
+        decrypted = encryption.decrypt(encrypted)
+        assert decrypted == plaintext
+
+    def test_encrypt_produces_different_ciphertext_each_time(self, encryption):
+        """Test that encrypting the same plaintext produces different ciphertext."""
+        plaintext = "token-value"
+        ciphertext1 = encryption.encrypt(plaintext)
+        ciphertext2 = encryption.encrypt(plaintext)
+        # Fernet includes timestamp and IV, so ciphertext differs each time
+        assert ciphertext1 != ciphertext2
+        # But both decrypt to same plaintext
+        assert encryption.decrypt(ciphertext1) == plaintext
+        assert encryption.decrypt(ciphertext2) == plaintext
+
+    def test_decrypt_with_wrong_key_fails(self, encryption):
+        """Test that decryption with wrong key fails."""
+        plaintext = "token-value"
+        encrypted = encryption.encrypt(plaintext)
+
+        # Create different encryption instance with different key
+        other_key = derive_encryption_key("different-secret")
+        other_encryption = TokenEncryption(other_key)
+
+        from cryptography.fernet import InvalidToken
+
+        with pytest.raises(InvalidToken):
+            other_encryption.decrypt(encrypted)
+
+    def test_encrypt_handles_unicode(self, encryption):
+        """Test that encryption handles unicode strings correctly."""
+        plaintext = "token-with-émojis-🔒"
+        encrypted = encryption.encrypt(plaintext)
+        decrypted = encryption.decrypt(encrypted)
+        assert decrypted == plaintext
+
+    def test_decrypt_rejects_tampered_ciphertext(self, encryption):
+        """Test that tampered ciphertext is rejected."""
+        plaintext = "token-value"
+        encrypted = encryption.encrypt(plaintext)
+
+        # Tamper with ciphertext
+        tampered = encrypted[:-1] + b"X"
+
+        from cryptography.fernet import InvalidToken
+
+        with pytest.raises(InvalidToken):
+            encryption.decrypt(tampered)

tests/server/auth/test_oauth_proxy.py

@@ -630,33 +630,103 @@ class TestOAuthProxyTokenEndpointAuth:
             token_endpoint_auth_method="client_secret_post",
         )
 
+        # First, create a valid FastMCP token via full OAuth flow
+        client = OAuthClientInformationFull(
+            client_id="test-client",
+            client_secret="test-secret",
+            redirect_uris=[AnyUrl("http://localhost:12345/callback")],
+        )
+
+        # Mock the upstream OAuth provider response
         with patch("fastmcp.server.auth.oauth_proxy.AsyncOAuth2Client") as MockClient:
             mock_client = AsyncMock()
+
+            # Mock initial token exchange (authorization code flow)
+            mock_client.fetch_token = AsyncMock(
+                return_value={
+                    "access_token": "upstream-access-token",
+                    "refresh_token": "upstream-refresh-token",
+                    "expires_in": 3600,
+                    "token_type": "Bearer",
+                }
+            )
+
+            # Mock token refresh
             mock_client.refresh_token = AsyncMock(
                 return_value={
-                    "access_token": "new-token",
-                    "refresh_token": "new-refresh",
+                    "access_token": "new-upstream-token",
+                    "refresh_token": "new-upstream-refresh",
                     "expires_in": 3600,
+                    "token_type": "Bearer",
                 }
             )
             MockClient.return_value = mock_client
 
-            client = OAuthClientInformationFull(
+            # Register client and do initial OAuth flow to get valid FastMCP tokens
+            await proxy.register_client(client)
+
+            # Store client code that would be created during OAuth callback
+            from fastmcp.server.auth.oauth_proxy import ClientCode
+
+            client_code = ClientCode(
+                code="test-auth-code",
                 client_id="test-client",
-                client_secret="test-secret",
-                redirect_uris=[AnyUrl("http://localhost:12345/callback")],
+                redirect_uri="http://localhost:12345/callback",
+                code_challenge="",
+                code_challenge_method="S256",
+                scopes=["read"],
+                idp_tokens={
+                    "access_token": "upstream-access-token",
+                    "refresh_token": "upstream-refresh-token",
+                    "expires_in": 3600,
+                    "token_type": "Bearer",
+                },
+                expires_at=time.time() + 300,
+                created_at=time.time(),
+            )
+            await proxy._code_store.put(key=client_code.code, value=client_code)
+
+            # Exchange authorization code to get FastMCP tokens
+            from mcp.server.auth.provider import AuthorizationCode
+
+            auth_code = AuthorizationCode(
+                code="test-auth-code",
+                scopes=["read"],
+                expires_at=time.time() + 300,
+                client_id="test-client",
+                code_challenge="",
+                redirect_uri=AnyUrl("http://localhost:12345/callback"),
+                redirect_uri_provided_explicitly=True,
+            )
+            result = await proxy.exchange_authorization_code(
+                client=client,
+                authorization_code=auth_code,
             )
 
-            refresh_token = RefreshToken(
-                token="old-refresh",
+            # Now test refresh with the valid FastMCP refresh token
+            assert result.refresh_token is not None
+            fastmcp_refresh = RefreshToken(
+                token=result.refresh_token,
                 client_id="test-client",
                 scopes=["read"],
                 expires_at=None,
             )
 
-            await proxy.exchange_refresh_token(client, refresh_token, ["read"])
+            # Reset mock to check refresh call
+            MockClient.reset_mock()
+            mock_client.refresh_token = AsyncMock(
+                return_value={
+                    "access_token": "new-upstream-token-2",
+                    "refresh_token": "new-upstream-refresh-2",
+                    "expires_in": 3600,
+                    "token_type": "Bearer",
+                }
+            )
+            MockClient.return_value = mock_client
 
-            # Verify auth method was passed
+            await proxy.exchange_refresh_token(client, fastmcp_refresh, ["read"])
+
+            # Verify auth method was passed to OAuth client
             MockClient.assert_called_with(
                 client_id="client-id",
                 client_secret="client-secret",
@@ -735,9 +805,18 @@ class TestOAuthProxyE2E:
             base_url="http://localhost:8000",
         )
 
-        # Mock initial tokens in provider
-        refresh_token = "mock_refresh_initial"
-        mock_oauth_provider.refresh_tokens[refresh_token] = {
+        client = OAuthClientInformationFull(
+            client_id="test-client",
+            client_secret="test-secret",
+            redirect_uris=[AnyUrl("http://localhost:12345/callback")],
+        )
+
+        # Register client first
+        await proxy.register_client(client)
+
+        # Set up initial upstream tokens in mock provider
+        upstream_refresh_token = "mock_refresh_initial"
+        mock_oauth_provider.refresh_tokens[upstream_refresh_token] = {
             "client_id": "mock-client",
             "scope": "read write",
         }
@@ -745,14 +824,24 @@ class TestOAuthProxyE2E:
         with patch("fastmcp.server.auth.oauth_proxy.AsyncOAuth2Client") as MockClient:
             mock_client = AsyncMock()
 
-            # Configure mock to call real provider
+            # Mock initial token exchange to get FastMCP tokens
+            mock_client.fetch_token = AsyncMock(
+                return_value={
+                    "access_token": "upstream-access-initial",
+                    "refresh_token": upstream_refresh_token,
+                    "expires_in": 3600,
+                    "token_type": "Bearer",
+                }
+            )
+
+            # Configure mock to call real provider for refresh
             async def mock_refresh(*args, **kwargs):
                 async with httpx.AsyncClient() as http:
                     response = await http.post(
                         mock_oauth_provider.token_endpoint,
                         data={
                             "grant_type": "refresh_token",
-                            "refresh_token": refresh_token,
+                            "refresh_token": upstream_refresh_token,
                         },
                     )
                     return response.json()
@@ -760,23 +849,61 @@ class TestOAuthProxyE2E:
             mock_client.refresh_token = mock_refresh
             MockClient.return_value = mock_client
 
-            # Test refresh
-            client = OAuthClientInformationFull(
+            # Store client code that would be created during OAuth callback
+            from fastmcp.server.auth.oauth_proxy import ClientCode
+
+            client_code = ClientCode(
+                code="test-auth-code",
                 client_id="test-client",
-                client_secret="test-secret",
-                redirect_uris=[AnyUrl("http://localhost:12345/callback")],
+                redirect_uri="http://localhost:12345/callback",
+                code_challenge="",
+                code_challenge_method="S256",
+                scopes=["read", "write"],
+                idp_tokens={
+                    "access_token": "upstream-access-initial",
+                    "refresh_token": upstream_refresh_token,
+                    "expires_in": 3600,
+                    "token_type": "Bearer",
+                },
+                expires_at=time.time() + 300,
+                created_at=time.time(),
+            )
+            await proxy._code_store.put(key=client_code.code, value=client_code)
+
+            # Exchange authorization code to get FastMCP tokens
+            from mcp.server.auth.provider import AuthorizationCode
+
+            auth_code = AuthorizationCode(
+                code="test-auth-code",
+                scopes=["read", "write"],
+                expires_at=time.time() + 300,
+                client_id="test-client",
+                code_challenge="",
+                redirect_uri=AnyUrl("http://localhost:12345/callback"),
+                redirect_uri_provided_explicitly=True,
+            )
+            initial_result = await proxy.exchange_authorization_code(
+                client=client,
+                authorization_code=auth_code,
             )
 
-            refresh = RefreshToken(
-                token=refresh_token,
+            # Now test refresh with the valid FastMCP refresh token
+            assert initial_result.refresh_token is not None
+            fastmcp_refresh = RefreshToken(
+                token=initial_result.refresh_token,
                 client_id="test-client",
                 scopes=["read"],
                 expires_at=None,
             )
 
-            result = await proxy.exchange_refresh_token(client, refresh, ["read"])
+            result = await proxy.exchange_refresh_token(
+                client, fastmcp_refresh, ["read"]
+            )
 
-            assert result.access_token.startswith("mock_access_")
+            # Should return new FastMCP tokens (not upstream tokens)
+            assert result.access_token != "upstream-access-initial"
+            # FastMCP tokens are JWTs (have 3 segments)
+            assert len(result.access_token.split(".")) == 3
             assert mock_oauth_provider.refresh_called
 
     @pytest.mark.asyncio