Skip to main content

Module 7 — Token authentication

Every route in modules 1 to 6 is reachable from anyone who can hit the port. That is fine on a laptop and never fine anywhere else. This module protects the churn service with an API key carried in a header, wires it through FastAPI's dependency injection, discusses when to graduate to JWTs, spells out key rotation, and closes with rate limiting because "authenticated" and "unlimited" are two very different guarantees.

Why the key belongs in a header, not the URL​

The very first temptation is to accept the key as a query parameter (?api_key=abc123). Do not. URLs are logged everywhere:

  • The client's browser history and referrer headers.
  • Every intermediate proxy's access log (GET /predict?api_key=abc123 200).
  • Your own server's access log, unless you actively scrub it.
  • Every error report and every screenshot support tickets attach.

An API key in the URL is a key that leaks into logs the moment it is used. The standard is to carry it in a request header (X-API-Key: <token>, or Authorization: Bearer <token> for OAuth-style clients). Headers are usually excluded from access logs by default and never end up in a browser history.

A security dependency that returns the caller​

FastAPI's dependency system is exactly the right shape for auth. Define a dependency that reads the header, resolves it to a caller, and either returns the caller or raises a 401. Every protected route then adds caller = Depends(current_caller) and receives a typed object it can log or authorize against:

# app/auth.py
import os
import hmac
from dataclasses import dataclass
from fastapi import Depends, HTTPException, status
from fastapi.security import APIKeyHeader

api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)


@dataclass(frozen=True)
class Caller:
    name: str
    scopes: tuple[str, ...]


# In production the store is a database or a KMS-backed secret manager;
# here it is a small dict keyed on the SHA-256 of the key.
CALLERS: dict[str, Caller] = {
    # sha256("dev-key-do-not-use") ->
    "3d3f1a...": Caller(name="dashboard-dev", scopes=("predict",)),
}


def _hash(token: str) -> str:
    import hashlib
    return hashlib.sha256(token.encode("utf-8")).hexdigest()


def current_caller(api_key: str | None = Depends(api_key_header)) -> Caller:
    if not api_key:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail={"code": "missing_api_key", "message": "X-API-Key header is required."},
            headers={"WWW-Authenticate": "APIKey"},
        )
    digest = _hash(api_key)
    caller = CALLERS.get(digest)
    if caller is None:
        # Constant-time is unnecessary here because the lookup is a hash
        # comparison, not the raw key. But we still avoid revealing which
        # part of the key failed.
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail={"code": "invalid_api_key", "message": "Unknown or revoked key."},
            headers={"WWW-Authenticate": "APIKey"},
        )
    return caller

The two lines that carry the security posture are APIKeyHeader(name="X-API-Key", auto_error=False) — which reads the header and returns None rather than raising, so we can shape the error ourselves — and the fact that CALLERS is keyed on _hash(token), never on the raw token. A database dump of the auth table is now useless: it contains hashes, not usable keys.

Use it on any protected route:

# app/main.py
from fastapi import FastAPI, Depends
from app.auth import Caller, current_caller
from app.schemas import Subscriber, Prediction
from app.scoring import _score_one

app = FastAPI(title="Churn scoring API", version="0.7.0")


@app.post("/predict", response_model=Prediction, tags=["prediction"])
def predict(subscriber: Subscriber, caller: Caller = Depends(current_caller)) -> Prediction:
    if "predict" not in caller.scopes:
        raise HTTPException(403, {"code": "forbidden", "message": "Missing scope: predict"})
    return _score_one(subscriber)

401 vs 403 is worth pausing on. 401 Unauthorized means "I do not know who you are"; 403 Forbidden means "I know who you are and you cannot do this". Confusing them makes support tickets ambiguous — the client cannot tell whether to fix its key or ask for permission.

JWT: when the API key stops fitting​

An API key is a shared secret: one string per caller, all requests carry the same bytes. That works up to a few dozen callers with stable identities. Beyond that, three limitations bite:

  • Revoking a leaked key means updating a database and (if you cache) invalidating the cache.
  • No structured claims: everything the server needs to know about the caller has to be looked up in the database.
  • No expiry: the key is valid until someone thinks to revoke it.

A JSON Web Token (JWT) fixes all three. It is a base64url-encoded JSON blob signed with either a shared secret (HS256) or a private key (RS256, ES256). The server verifies the signature and reads the claims — sub (subject), exp (expiry), scopes, ... — without a database call. A shape sketch:

# app/auth_jwt.py — preview only
import time, jwt  # pip install pyjwt
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer

bearer = HTTPBearer(auto_error=False)
JWT_PUBLIC_KEY = open("/etc/keys/jwt-pub.pem").read()


def caller_from_jwt(cred: HTTPAuthorizationCredentials = Depends(bearer)) -> Caller:
    if cred is None:
        raise HTTPException(401, {"code": "missing_token"})
    try:
        claims = jwt.decode(cred.credentials, JWT_PUBLIC_KEY, algorithms=["RS256"], audience="churn-api")
    except jwt.PyJWTError:
        raise HTTPException(401, {"code": "invalid_token"})
    return Caller(name=claims["sub"], scopes=tuple(claims.get("scopes", ())))

Two design choices matter. RS256 (private-key signature) means the API only needs the public key — a leaked API image cannot mint tokens. And exp is enforced by jwt.decode — a stolen token stops being valid at its expiry, without any revocation infrastructure. JWT is not a replacement for the API key pattern for a small internal service; it is what you graduate to when the caller set becomes an identity provider's problem.

Rotating keys without an outage​

A key that has never been rotated is a key that has been leaked and no one noticed. The rotation pattern that costs nothing is two active keys with an overlap window:

  1. Generate a new key. Add it to the CALLERS store alongside the old one.
  2. Distribute the new key to the client (a Kubernetes secret, an env var, a CI variable).
  3. The client restarts and starts using the new key.
  4. After a 72 h grace window, delete the old key from the store.

Both keys are valid during the overlap, so no request is ever refused because of a race. If step 3 fails, you notice within seconds because the client's error rate rises, and you can extend the overlap. If step 3 succeeds, step 4 removes the old key long before an attacker can use a snapshot from before the rotation.

Automate the schedule (e.g. quarterly), because a rotation that requires a human calendar entry does not happen.

Rate limiting: authenticated does not mean unlimited​

An authenticated caller can still bring down the service. Every real ML API has at least two limits:

  • Per-caller QPS: a hard ceiling on requests per second per key. A misconfigured cron job that hits /predict in a tight loop is common; a limit of 20 QPS per key contains the blast.
  • Per-caller monthly quota: for cost tracking and to catch runaway spend. Report the remaining quota in response headers so clients can back off preemptively.

FastAPI has no built-in rate limiter, but slowapi (based on limits) is the standard choice:

# app/rate_limit.py
from slowapi import Limiter
from slowapi.util import get_remote_address
from fastapi import Request


def key_from_caller(request: Request) -> str:
    # Prefer the authenticated identity; fall back to the client IP.
    caller = getattr(request.state, "caller", None)
    return caller.name if caller else get_remote_address(request)


limiter = Limiter(key_func=key_from_caller, default_limits=["20/second", "10000/hour"])

Attach it to the app and decorate the routes; a caller that goes over the limit gets a 429 Too Many Requests with a Retry-After header. A 429 is a first-class HTTP status code — clients should recognize it and back off, and the module 5 error format applies here too.

An API key stored in the git repo is a leaked key

The most common leak of an API key is not a sophisticated attack; it is a developer committing a .env file. Enforce it in CI (a pre-commit hook, git-secrets, truffleHog) rather than trusting reviewers. And treat a key that has ever touched git as compromised, even after a git filter-repo — someone has already cloned it.

In summary​

  • Carry the token in a header (X-API-Key or Authorization: Bearer), never in the URL. Store its hash on the server, never the raw key.
  • Wrap the auth check in a FastAPI dependency that returns a typed Caller — routes stay clean, tests can override it, and 401 vs 403 stays honest.
  • JWT graduates the pattern when you need expiry, structured claims and no per-request database lookup; RS256 so the API only holds the public key.
  • Rotate on a schedule with an overlap window; rate-limit per caller because authenticated does not mean unlimited, and return 429 with Retry-After.

Next module: logging and health probes — putting a request ID on every log line so a support ticket can be traced in seconds, and exposing the two probe endpoints the orchestrator needs.