PyJWT — FastAPI Integration & Token Service¶

uv add "pyjwt[crypto]" fastapi pydantic
uv add --dev pytest freezegun httpx2        # httpx2: the client behind fastapi.testclient (Starlette 1.x)

app/__init__.py
app/auth.py          token issuing and verification
app/main.py          FastAPI app and dependencies
tests/__init__.py
tests/conftest.py    keys, token factory, client      (05)
tests/jwks_server.py local fake JWKS endpoint         (06)
tests/test_*.py
pytest.ini

# app/auth.py
from __future__ import annotations

import uuid
from collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Any

import jwt

ACCESS, REFRESH = "access", "refresh"
REQUIRED = ["exp", "iat", "nbf", "iss", "aud", "sub", "jti", "token_use"]


def utc_now() -> datetime:
    return datetime.now(timezone.utc)


@dataclass(frozen=True)
class AuthConfig:
    issuer: str
    audience: str
    algorithm: str = "RS256"
    access_ttl: timedelta = timedelta(minutes=15)
    refresh_ttl: timedelta = timedelta(days=7)
    leeway: timedelta = timedelta(seconds=10)


class TokenService:
    """`verify_key`: PEM public key, PyJWK or PyJWKClient. `private_key` is needed to issue."""

    def __init__(
        self,
        config: AuthConfig,
        *,
        verify_key: Any,
        private_key: str | bytes | None = None,
        kid: str | None = None,
        clock: Callable[[], datetime] = utc_now,
    ) -> None:
        self.config, self.verify_key, self.private_key = config, verify_key, private_key
        self.kid, self.clock = kid, clock

    def _issue(self, sub: str, token_use: str, ttl: timedelta, **extra: Any) -> str:
        now = self.clock()
        claims = {
            "iss": self.config.issuer, "aud": self.config.audience, "sub": sub,
            "iat": now, "nbf": now, "exp": now + ttl,
            "jti": uuid.uuid4().hex, "token_use": token_use, **extra,
        }
        headers = {"kid": self.kid} if self.kid else None
        return jwt.encode(claims, self.private_key, algorithm=self.config.algorithm, headers=headers)

    def issue_pair(self, sub: str, scopes: list[str]) -> dict[str, str]:
        return {
            "access_token": self._issue(sub, ACCESS, self.config.access_ttl, scope=" ".join(scopes)),
            "refresh_token": self._issue(sub, REFRESH, self.config.refresh_ttl),
            "token_type": "bearer",
        }

    def verify(self, token: str, *, token_use: str = ACCESS) -> dict[str, Any]:
        key = self.verify_key
        if isinstance(key, jwt.PyJWKClient):
            key = key.get_signing_key_from_jwt(token)
        claims = jwt.decode(
            token,
            key,
            algorithms=[self.config.algorithm],
            audience=self.config.audience,
            issuer=self.config.issuer,
            leeway=self.config.leeway,
            options={"require": REQUIRED},
        )
        if claims["token_use"] != token_use:           # a refresh token is not an access token
            raise jwt.InvalidTokenError("unexpected token_use")
        return claims

# app/main.py
from __future__ import annotations

import logging

import jwt
from fastapi import Depends, FastAPI, HTTPException, Request
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from pydantic import BaseModel, ValidationError

from app.auth import REFRESH, TokenService

log = logging.getLogger("auth")
bearer = HTTPBearer(auto_error=False)


class Claims(BaseModel):
    sub: str
    jti: str
    scope: str = ""

    @property
    def scopes(self) -> set[str]:
        return set(self.scope.split())


def unauthorized(error: str | None = None) -> HTTPException:
    challenge = "Bearer" if error is None else f'Bearer error="{error}"'
    return HTTPException(401, "Not authenticated", headers={"WWW-Authenticate": challenge})


def get_tokens(request: Request) -> TokenService:
    return request.app.state.tokens


def current_claims(
    creds: HTTPAuthorizationCredentials | None = Depends(bearer),
    tokens: TokenService = Depends(get_tokens),
) -> Claims:
    if creds is None:
        raise unauthorized()
    try:
        return Claims.model_validate(tokens.verify(creds.credentials))
    except jwt.PyJWKClientConnectionError as exc:           # JWKS endpoint down: not the caller's fault
        raise HTTPException(503, "Identity provider unavailable") from exc
    except (jwt.InvalidTokenError, jwt.PyJWKClientError, ValidationError) as exc:
        log.info("token rejected: %s", type(exc).__name__)    # class name only, never the token
        raise unauthorized("invalid_token") from exc


def require_scope(scope: str):
    def check(claims: Claims = Depends(current_claims)) -> Claims:
        if scope not in claims.scopes:
            challenge = f'Bearer error="insufficient_scope", scope="{scope}"'
            raise HTTPException(403, "Insufficient scope", headers={"WWW-Authenticate": challenge})
        return claims

    return check


class RefreshBody(BaseModel):
    refresh_token: str


def create_app(tokens: TokenService) -> FastAPI:
    app = FastAPI()
    app.state.tokens = tokens
    app.state.used_refresh = set()          # use Redis or a DB table in production

    @app.post("/auth/refresh")
    def refresh(body: RefreshBody, request: Request, svc: TokenService = Depends(get_tokens)):
        try:
            claims = svc.verify(body.refresh_token, token_use=REFRESH)
        except (jwt.InvalidTokenError, jwt.PyJWKClientError) as exc:
            raise unauthorized("invalid_token") from exc
        used: set[str] = request.app.state.used_refresh
        if claims["jti"] in used:                # a rotated token came back: reject (and revoke the family)
            raise unauthorized("invalid_token")
        used.add(claims["jti"])
        return svc.issue_pair(claims["sub"], ["orders:read"])

    @app.get("/orders")
    def list_orders(claims: Claims = Depends(require_scope("orders:read"))):
        return {"owner": claims.sub, "orders": []}

    @app.post("/orders", status_code=201)
    def create_order(claims: Claims = Depends(require_scope("orders:write"))):
        return {"owner": claims.sub}

    return app