Authentication (JWT)

What Your Agent Inherits

Your agent’s route handlers receive an authenticated Principal object through FastAPI dependency injection. Every concern below the route signature, including token parsing, key resolution, signature verification, claim validation, and clock-skew tolerance, is already handled by the chassis’s JWTAuthService.

The agent never inspects the Authorization header directly. Instead, it declares a dependency, receives a typed Principal, and makes business decisions based on subject, scopes, and roles.

Three Validation Modes

The chassis supports three sources of key material for JWT signature verification, all selected through environment variables:

Only one mode is active at a time, and the configuration validator rejects mixed setups at startup. The JWTAuthService constructor stores the settings and prepares the JWKS cache lock for the async runtime:

class JWTAuthService:
  """Validate externally-issued JWTs against local config or JWKS."""

  def __init__(self, settings: Settings, http_client: httpx.AsyncClient) -> None:
      self.settings = settings
      self.http_client = http_client
      self._jwks_lock = asyncio.Lock()
      self._jwks_cache: dict[str, Any] | None = None
      self._jwks_loaded_at = 0.0
      self._jwks_last_fetch_used_stale_cache = False

Key Resolution

When a token arrives, _resolve_key() walks the three modes in priority order: JWKS first, then static key, then shared secret. Because of this ordering, a production deployment with APP_AUTH_JWKS_URL set will always use rotating keys, even if a fallback secret happens to be configured:

async def _resolve_key(self, token: str) -> Any:
  if self.settings.auth_jwks_url:
      header = jwt.get_unverified_header(token)
      kid = header.get("kid")
      if not kid:
          raise AuthenticationError("JWT is missing kid header required for JWKS")

      jwks = await self._fetch_jwks(force_refresh=False)
      key = _get_jwk_key_for_kid(jwks, kid)
      if key is not None:
          return key

      # Force one refresh on kid misses so key rotation does not require
      # a process restart or cache TTL expiry before new tokens work.
      refreshed_jwks = await self._fetch_jwks(force_refresh=True)
      refreshed_key = _get_jwk_key_for_kid(refreshed_jwks, kid)
      if refreshed_key is not None:
          return refreshed_key
      raise AuthenticationError("No matching signing key found")

  if self.settings.auth_jwt_public_key:
      return self.settings.auth_jwt_public_key

  if self._uses_shared_secret():
      return self.settings.auth_jwt_secret

  raise AuthenticationError("JWT validation is enabled but no key material is configured")

Notice the forced refresh on kid miss (lines 131-134). When an identity provider rotates its signing key, the first token signed with the new key will naturally miss the cache. Rather than rejecting the request outright, the service forces a single JWKS refresh. If the new kid appears in the refreshed key set, the token validates without any operator intervention or a process restart.

JWKS with Graceful Degradation

The _fetch_jwks() method implements a cache-aside pattern backed by an asyncio.Lock to prevent thundering herd fetches:

async def _fetch_jwks(self, force_refresh: bool) -> dict[str, Any]:
  if not force_refresh and self._jwks_cache and not self._jwks_cache_expired():
      self._jwks_last_fetch_used_stale_cache = False
      return self._jwks_cache

  async with self._jwks_lock:
      if not force_refresh and self._jwks_cache and not self._jwks_cache_expired():
          self._jwks_last_fetch_used_stale_cache = False
          return self._jwks_cache

      try:
          response = await self.http_client.get(self.settings.auth_jwks_url)
          response.raise_for_status()
          self._jwks_cache = _validate_jwks_payload(response.json())
          self._jwks_loaded_at = monotonic()
          self._jwks_last_fetch_used_stale_cache = False
          return self._jwks_cache
      except Exception:
          if self._jwks_cache is not None:
              stale_seconds = monotonic() - self._jwks_loaded_at
              max_stale = self.settings.auth_jwks_max_stale_seconds
              if max_stale > 0 and stale_seconds <= max_stale:
                  self._jwks_last_fetch_used_stale_cache = True
                  logger.warning(
                      "JWKS refresh failed; using stale cache (%.0fs old, max %ds)",
                      stale_seconds,
                      max_stale,
                  )
                  return self._jwks_cache
              logger.error(
                  "JWKS refresh failed and stale cache exceeded max age "
                  "(%.0fs old, max %ds); rejecting",
                  stale_seconds,
                  max_stale,
              )
          raise

Key behaviors:

• Double-check locking. The fast path returns the cached value without acquiring the lock. Inside the lock body, staleness is re-checked to avoid redundant fetches.
• TTL-based refresh. The cache expires after auth_jwks_cache_ttl_seconds (default 300 s), which you can tune via APP_AUTH_JWKS_CACHE_TTL_SECONDS.
• Stale-cache fallback with age limits. If the JWKS endpoint is temporarily unreachable but a stale cache exists, the service keeps validating tokens with the last known key set instead of rejecting every request. This grace period lasts up to auth_jwks_max_stale_seconds (default 3600 s). Once the stale cache exceeds that age, the service starts rejecting requests rather than relying on dangerously old keys. The readiness check will report the degraded state.

The Principal Model

Once the JWT passes signature and claim validation, the service maps its raw claims to a typed Pydantic model. This gives route handlers a stable, well-defined domain object rather than an arbitrary dictionary:

class Principal(BaseModel):
  """Authenticated principal extracted from a validated JWT."""

  subject: str = Field(description="JWT subject claim")
  issuer: str | None = Field(default=None, description="Token issuer")
  audience: list[str] = Field(default_factory=list, description="Normalized audience values")
  scopes: list[str] = Field(default_factory=list, description="Granted OAuth scopes")
  roles: list[str] = Field(default_factory=list, description="Application roles")
  claims: dict[str, Any] = Field(default_factory=dict, description="Full validated claim set")

The scopes and roles fields are normalized from several common formats (space-delimited strings, JSON arrays, and comma-separated values), so your agent can use simple in checks regardless of how the identity provider structures its claims.

Configuration Validation

Authentication misconfigurations are caught at startup, not at runtime. The _validate_auth_settings() function enforces a set of consistency rules:

def _validate_auth_settings(settings: Settings) -> None:
  """Validate JWT/auth-related configuration after defaults resolve."""
  if settings.auth_jwks_url and not settings.auth_jwks_url.startswith("https://"):
      raise ValueError("APP_AUTH_JWKS_URL must use https://")

  if not settings.auth_enabled:
      return

  algorithm_families = {
      _jwt_algorithm_family(algorithm) for algorithm in settings.auth_jwt_algorithms
  }
  if len(algorithm_families) != 1:
      raise ValueError("APP_AUTH_JWT_ALGORITHMS must all belong to the same algorithm family")

  uses_shared_secret = "HS" in algorithm_families
  _validate_auth_key_material(settings, uses_shared_secret)
  _validate_auth_claim_requirements(settings)

Here is what gets caught before the first request ever arrives:

• JWKS URL served over plain HTTP (must be HTTPS)
• Mixed algorithm families (e.g., HS256 combined with RS256)
• Missing key material for the chosen algorithm family
• Shared secrets shorter than 32 characters when using HS* algorithms
• Missing issuer or audience values when the corresponding auth_require_* flags are enabled

Dependency Injection

The final piece wires the JWTAuthService into FastAPI’s dependency injection system. Route handlers simply declare what authorization level they need, and the framework takes care of the rest:

async def get_optional_principal(
  request: Request,
  credentials: HTTPAuthorizationCredentials | None = Depends(bearer_scheme),
) -> Principal | None:
  """Return the authenticated principal when a bearer token is provided."""
  if credentials is None:
      return None

  auth_service = get_auth_service(request)
  try:
      return await auth_service.authenticate_token(credentials.credentials)
  except AuthenticationError as exc:
      raise HTTPException(
          status_code=status.HTTP_401_UNAUTHORIZED,
          detail="Invalid bearer token",
          headers={"WWW-Authenticate": "Bearer"},
      ) from exc


def get_current_principal(
  principal: Principal | None = Depends(get_optional_principal),
) -> Principal:
  """Require and return an authenticated principal."""
  if principal is None:
      raise HTTPException(
          status_code=status.HTTP_401_UNAUTHORIZED,
          detail="Missing bearer token",
          headers={"WWW-Authenticate": "Bearer"},
      )
  return principal

Four dependency options are available to your agent’s routes:

A typical protected endpoint looks like this:

@router.get("/items")
async def list_items(principal: Principal = Depends(get_current_principal)):
    # principal.subject, principal.scopes, principal.roles are all available
    return await service.list_items_for(principal.subject)

Best Practices

• Always use JWKS for production JWT validation to support automatic key rotation without process restarts or redeployments.
• Never store private keys or shared secrets in environment variables for production. Use JWKS endpoints backed by your identity provider (Auth0, Keycloak, Entra ID) for automatic key management.
• Always validate iss and aud claims in production. Without issuer and audience validation, tokens from unrelated services can authenticate against your API.
• Prefer typed Principal models over raw claim dictionaries. Typed models provide IDE autocompletion, runtime validation, and protect against typos in claim field names.
• Always implement clock-skew tolerance (the default leeway in PyJWT) for JWT exp validation. Clock drift between the identity provider and your application is inevitable in distributed systems.
• Never mix algorithm families in a single deployment. HS256 (symmetric) and RS256 (asymmetric) have fundamentally different security properties and key management requirements.

Further Reading

• RFC 7519 — JSON Web Token (JWT)
• RFC 7517 — JSON Web Key (JWK)
• FastAPI Security Documentation
• OWASP JWT Security Cheat Sheet
• Auth0 — JWKS Best Practices

What the Agent Never Implements

The following concerns live entirely in the chassis infrastructure. Your agent’s generated code should never contain logic for any of these:

• Token parsing. The chassis extracts the bearer token from the Authorization header.
• Key management. Secrets, public keys, and JWKS endpoints are loaded automatically.
• JWKS rotation. Cache refresh, stale-cache fallback, and kid-miss retry are built in.
• Signature verification. The chassis calls jwt.decode with the correct algorithm and key.
• Claim validation. Checks for exp, iss, aud, and any required claims are automatic.
• Principal mapping. Scopes, roles, and audience values are normalized from varying formats.
• HTTP 401/403 responses. The dependency chain raises the correct status codes on its own.