API Authentication

Cognito JWT validation, OAuth2 scopes, and rate limiting.

All /v1/ endpoints require a valid Cognito Bearer token unless explicitly public. In FastAPI, authentication is expressed as a dependency — never validate tokens inside route bodies. This is the Python counterpart to the Express auth middleware; the Cognito user pool, issuer, and scope model are identical.

Request flow

Cognito JWT verification

bash

uv add "python-jose[cryptography]" httpx

python

# app/core/security.py
from functools import lru_cache
 
import httpx
from jose import jwt
from jose.exceptions import JWTError
 
from app.core.config import settings
 
ISSUER = (
    f"https://cognito-idp.{settings.cognito_region}.amazonaws.com/"
    f"{settings.cognito_user_pool_id}"
)
 
 
@lru_cache(maxsize=1)
def _jwks() -> dict:
    resp = httpx.get(f"{ISSUER}/.well-known/jwks.json", timeout=5)
    resp.raise_for_status()
    return resp.json()
 
 
def verify_cognito_token(token: str) -> dict:
    header = jwt.get_unverified_header(token)
    key = next((k for k in _jwks()["keys"] if k["kid"] == header["kid"]), None)
    if key is None:
        raise JWTError("Signing key not found")
 
    return jwt.decode(
        token,
        key,
        algorithms=["RS256"],
        issuer=ISSUER,
        options={"verify_aud": False},  # access tokens have no aud claim
    )

Auth dependency

python

# app/api/deps.py
from typing import Annotated
 
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from jose.exceptions import JWTError
from pydantic import BaseModel
 
from app.core.security import verify_cognito_token
 
bearer = HTTPBearer(auto_error=False)
 
 
class AuthUser(BaseModel):
    sub: str
    email: str | None = None
    scopes: list[str] = []
 
 
async def get_current_user(
    creds: Annotated[HTTPAuthorizationCredentials | None, Depends(bearer)],
) -> AuthUser:
    if creds is None:
        raise HTTPException(
            status.HTTP_401_UNAUTHORIZED,
            detail={"code": "unauthorized", "message": "Missing bearer token"},
        )
    try:
        claims = verify_cognito_token(creds.credentials)
    except JWTError:
        raise HTTPException(
            status.HTTP_401_UNAUTHORIZED,
            detail={"code": "unauthorized", "message": "Invalid or expired token"},
        )
 
    return AuthUser(
        sub=claims["sub"],
        email=claims.get("email"),
        scopes=claims.get("custom:scopes", "").split(),
    )
 
 
CurrentUser = Annotated[AuthUser, Depends(get_current_user)]

CurrentUser is a reusable typed dependency — add it to any route signature to require authentication and inject the verified user.

Scope dependency

python

# app/api/deps.py (continued)
from fastapi import Depends
 
 
def require_scope(*required: str):
    async def checker(user: CurrentUser) -> None:
        if not any(s in user.scopes or "admin:*" in user.scopes for s in required):
            raise HTTPException(
                status.HTTP_403_FORBIDDEN,
                detail={"code": "forbidden", "message": "Insufficient scope"},
            )
 
    return checker

Use it alongside the auth dependency:

python

@router.post("", dependencies=[Depends(require_scope("write:projects"))])
async def create_project(payload: ProjectCreate, user: CurrentUser):
    ...

Scopes reference

Scope	Access
`read:projects`	List and view projects
`write:projects`	Create and update projects
`read:users`	View user profiles
`write:users`	Update user profiles
`admin:*`	Full admin access — bypasses scope checks

Scopes are stored in the Cognito custom attribute custom:scopes as a space-separated string — the same source of truth used by the Node services.

Rate limiting

bash

uv add slowapi

python

# app/middleware/rate_limit.py
from slowapi import Limiter
from slowapi.util import get_remote_address
 
limiter = Limiter(key_func=get_remote_address, default_limits=["100/minute"])

python

# app/main.py (wiring)
from slowapi import _rate_limit_exceeded_handler
from slowapi.errors import RateLimitExceeded
 
from app.middleware.rate_limit import limiter
 
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

Tier	Limit	Window	Key
Unauthenticated	100 req	1 min	IP address
Authenticated	1,000 req	1 min	Cognito `sub`
Service account	10,000 req	1 min	API key (internal)

Error responses

FastAPI serializes the HTTPException detail dict directly, keeping the same JSON shape as the Node services:

json

{
  "detail": {
    "code": "unauthorized",
    "message": "Invalid or expired token."
  }
}

Status	Code	When
401	`unauthorized`	Missing or invalid token
403	`forbidden`	Valid token, insufficient scope
429	`rate_limit_exceeded`	Rate limit hit

Testing with curl

bash

export TOKEN="eyJhbG..."
 
curl http://localhost:4000/v1/projects \
  -H "Authorization: Bearer $TOKEN"