Pin Clerk issuer to prevent JWKS endpoint spoofing

packages/fastapi/.env.example

@@ -1,4 +1,7 @@
 OPENROUTER_API_KEY=
 CONVEX_URL=https://your-deployment.convex.cloud
 CONVEX_DEPLOY_KEY=
-FRONTEND_URL=http://localhost:3001
+FRONTEND_URL=http://localhost:3001
+# Clerk issuer URL — set to your Clerk Frontend API URL, e.g.:
+# https://<your-clerk-subdomain>.clerk.accounts.dev
+CLERK_ISSUER=

packages/fastapi/app/auth.py

@@ -4,19 +4,21 @@
 import jwt
 from fastapi import HTTPException, Request
 
+from app.config import settings
+
 logger = logging.getLogger(__name__)
 
 _jwks_cache: dict | None = None
 
 
-async def _get_jwks(client: httpx.AsyncClient, issuer: str) -> dict:
-    """Fetch and cache Clerk's JWKS keys."""
+async def _get_jwks(client: httpx.AsyncClient, jwks_url: str) -> dict:
+    """Fetch and cache Clerk's JWKS keys from a pinned URL."""
     global _jwks_cache
     if _jwks_cache is not None:
         return _jwks_cache
 
-    logger.debug("Fetching JWKS from %s", issuer)
-    resp = await client.get(f"{issuer}/.well-known/jwks.json", timeout=10.0)
+    logger.debug("Fetching JWKS from %s", jwks_url)
+    resp = await client.get(jwks_url, timeout=10.0)
     resp.raise_for_status()
     _jwks_cache = resp.json()
     return _jwks_cache
@@ -34,21 +36,31 @@ async def verify_token(request: Request) -> dict:
     token = auth_header[7:]
     http_client = request.app.state.http_client
 
+    # Require a pinned issuer to prevent JWKS spoofing attacks
+    expected_issuer = settings.clerk_issuer
+    if not expected_issuer:
+        logger.error("CLERK_ISSUER is not configured")
+        raise HTTPException(status_code=500, detail="Server authentication misconfigured")
+
+    jwks_url = f"{expected_issuer.rstrip('/')}/.well-known/jwks.json"
+
     try:
         # Decode header to get key ID
         unverified_header = jwt.get_unverified_header(token)
         kid = unverified_header.get("kid")
         if not kid:
             raise HTTPException(status_code=401, detail="Token missing key ID")
 
-        # Get issuer from unverified claims to fetch JWKS
+        # Validate `iss` from unverified payload against the pinned issuer
+        # before any JWKS lookup, so attacker-controlled issuers are rejected early.
         unverified_claims = jwt.decode(token, options={"verify_signature": False})
-        issuer = unverified_claims.get("iss", "")
-        if not issuer:
-            raise HTTPException(status_code=401, detail="Token missing issuer")
+        token_issuer = unverified_claims.get("iss", "")
+        if token_issuer != expected_issuer:
+            logger.warning("Token issuer mismatch: expected %s, got %s", expected_issuer, token_issuer)
+            raise HTTPException(status_code=401, detail="Invalid token issuer")
 
-        # Fetch JWKS and find matching key
-        jwks = await _get_jwks(http_client, issuer)
+        # Fetch JWKS from the pinned URL (not from the token's iss)
+        jwks = await _get_jwks(http_client, jwks_url)
         key = None
         for k in jwks.get("keys", []):
             if k.get("kid") == kid:
@@ -60,7 +72,7 @@ async def verify_token(request: Request) -> dict:
             global _jwks_cache
             _jwks_cache = None
             logger.info("JWKS key %s not found in cache, refetching", kid)
-            jwks = await _get_jwks(http_client, issuer)
+            jwks = await _get_jwks(http_client, jwks_url)
             for k in jwks.get("keys", []):
                 if k.get("kid") == kid:
                     key = jwt.algorithms.RSAAlgorithm.from_jwk(k)
@@ -73,6 +85,7 @@ async def verify_token(request: Request) -> dict:
             token,
             key,
             algorithms=["RS256"],
+            issuer=expected_issuer,
             options={"verify_aud": False},
         )
         return payload

packages/fastapi/app/config.py

@@ -13,6 +13,9 @@ class Settings(BaseSettings):
     frontend_url: str = "http://localhost:3000"
     # Public base URL of the FastAPI backend, used for OAuth redirect URIs.
     fastapi_base_url: str = "http://localhost:8000"
+    # Clerk issuer URL used to pin and validate JWT `iss` claims.
+    # Must be set to your Clerk Frontend API URL, e.g. https://<your-clerk-subdomain>.clerk.accounts.dev
+    clerk_issuer: str = ""
     # Pre-registered GitHub OAuth App credentials (for GitHub MCP server).
     # Create one at https://github.com/settings/applications/new
     github_oauth_client_id: str = ""
@@ -30,6 +33,10 @@ def validate_startup(self) -> None:
             logger.warning(
                 "CONVEX_DEPLOY_KEY is not set — backend cannot save messages to Convex"
             )
+        if not self.clerk_issuer:
+            logger.warning(
+                "CLERK_ISSUER is not set — JWT authentication will fail for all requests"
+            )
 
 
 settings = Settings()