Chapter 3 · Production FastAPI
Authentication and Authorisation
- Page 11 of 22
- 30 min read
Every AI product eventually needs to know who is calling. Your prompt library must not let strangers read private chats, your model endpoint costs money per call, and an admin screen must stay admin-only. Most real API breaches are not clever cryptography attacks but a missing check, a leaked key or a token that never expires. This page builds a small login system from the ground up with FastAPI, so you see every moving part: password hashing, JWT access and refresh tokens, revocation, roles, per-object permissions, API keys for services, cookies and login rate limiting, and then you test all of it.
The page assumes HTTP in Depth: Idempotency, Caching and Auth (401 against 403, API keys against bearer tokens, the OAuth flows) and FastAPI Architecture: Routers, Schemas and Error Handling. The code keeps its own in-memory store so it runs by itself; Databases with SQLAlchemy and Alembic shows the real database that would sit underneath.
What you will learn
- Authentication against authorisation, and the four failures an auth system must tell apart.
- Hashing passwords with
pwdliband Argon2, and a login that does not leak which usernames exist. - JWTs with PyJWT: structure, claims, signing, the attacks to block (tampering,
alg: none, expiry), and access plus refresh tokens with rotation and revocation. - A bearer-token dependency, protected routes, roles (RBAC) and per-object checks that stop BOLA.
- API keys for services, cookies against bearer tokens and CSRF, OAuth2/OIDC login with an external provider, login rate limiting, and tests for all of it.
Two questions, four outcomes
| Question | Name | Failure looks like | Where it lives in this page |
|---|---|---|---|
| Who are you? | Authentication | 401 with WWW-Authenticate: Bearer | login, tokens, current_user |
| May you do this action? | Authorisation (role) | 403 | require_role("admin") |
| May you touch this object? | Authorisation (resource) | 404 (hide existence) or 403 | owned_chat |
| Are you sending too many guesses? | Abuse control | 429 with Retry-After | the login limiter |
A request passes through them in that order. Skipping the third row is the classic bug: the user is logged in, so the endpoint trusts them with any id they type.
Set up the project
The app signs tokens with a secret that must never be in the code. It comes from the environment (the page Dependencies, Lifespan, Middleware and Settings shows the settings class that normally reads it). HS256 needs at least 32 bytes; generate a real one with python -c "import secrets; print(secrets.token_urlsafe(48))". The value below is for this tutorial only.
mkdir -p ~/projects/auth-api
cd ~/projects/auth-api
uv init --bare -q
uv add -q "fastapi[standard]" pyjwt "pwdlib[argon2]"
uv add -q --dev pytest httpx2
# with pip: python -m venv .venv && source .venv/bin/activate && pip install -q "fastapi[standard]" pyjwt "pwdlib[argon2]" pytest httpx2
mkdir -p app tests
touch app/__init__.py
cat >> pyproject.toml <<'EOF'
[tool.fastapi]
entrypoint = "app.main:app"
EOF
export JWT_SECRET="tutorial-only-secret-change-me-0123456789"
export BASE=http://localhost:8000Passwords: store a slow, salted hash
Never store passwords, and never "encrypt" them: if the database leaks, every user's password leaks with it, and users reuse passwords. Store a hash from a deliberately slow, memory-hungry algorithm, with a random salt per password, so that guessing millions of passwords per second is no longer possible. A fast hash like SHA-256 or MD5 is the wrong tool here. In 2026 the default choice is Argon2id, and in FastAPI's ecosystem the library is pwdlib (it replaces the unmaintained passlib):
from pwdlib import PasswordHash
passwords = PasswordHash.recommended() # Argon2id with sensible parameters
hashed = passwords.hash("correct-horse-1")
print(hashed.split("$")[1:3]) # algorithm and version, then salt and hash
print(hashed.startswith("$argon2id$v=19$"))
print(passwords.verify("correct-horse-1", hashed), passwords.verify("wrong-guess", hashed))
print(hashed == passwords.hash("correct-horse-1"))['argon2id', 'v=19']
True
True False
False- The result is one self-describing string: algorithm, version, cost parameters, salt and hash.
verifyreads the parameters from the string, so you can raise the cost later and old hashes still work (verify_and_updatealso returns a fresh hash to store when the parameters are outdated). - Hashing the same password twice gives different strings, because of the random salt. Never compare hashes yourself; always call
verify. - Set a maximum password length (128 is plenty) as well as a minimum: hashing a 10 MB "password" is a cheap way to burn your CPU.
Never print, log or return a hash, and never put one in a response model. Section "Run it" shows that the
UserOutmodel leaves it out even though the stored row contains it.
Tokens: what a JWT is
After login the client needs proof of who it is on every later request, without sending the password again. A JSON Web Token (JWT) is a signed, self-contained proof: three Base64URL parts joined by dots, header.payload.signature. The server signs with a secret; later it can check the signature and trust the claims inside without a database lookup. Write the security module (no FastAPI in it, in the spirit of the services from FastAPI Architecture: Routers, Schemas and Error Handling):
# file: app/security.py
"""Passwords, tokens, API keys and the login limiter. No FastAPI in here."""
import hashlib
import math
import os
import secrets
import time
import uuid
from collections.abc import Callable
from datetime import datetime, timedelta, timezone
import jwt
from pwdlib import PasswordHash
ALGORITHM = "HS256"
ISSUER = "bitbyte-auth-demo"
ACCESS_TTL = timedelta(minutes=15)
REFRESH_TTL = timedelta(days=7)
passwords = PasswordHash.recommended()
DUMMY_HASH = passwords.hash("not-a-real-password") # for unknown usernames, see login
def secret_key() -> str:
return os.environ["JWT_SECRET"]
def make_token(sub: str, kind: str, *, family: str | None = None, now: datetime | None = None,
ttl: timedelta | None = None, jti: str | None = None) -> str:
"""Sign a token. `now`, `ttl` and `jti` are parameters so examples and tests can fix them."""
now = now or datetime.now(timezone.utc)
ttl = ttl or (ACCESS_TTL if kind == "access" else REFRESH_TTL)
claims = {
"iss": ISSUER, # who issued it
"sub": sub, # whose it is (the user id, as a string)
"typ": kind, # "access" or "refresh": never accept one as the other
"fam": family or uuid.uuid4().hex, # the login session this token belongs to
"jti": jti or uuid.uuid4().hex, # unique id of this token
"iat": now, # issued at
"exp": now + ttl, # expires at
}
return jwt.encode(claims, secret_key(), algorithm=ALGORITHM)
def decode_token(token: str, kind: str) -> dict:
"""Verify signature, algorithm, issuer, expiry and token type. Raises jwt.InvalidTokenError."""
claims = jwt.decode(token, secret_key(), algorithms=[ALGORITHM], issuer=ISSUER,
options={"require": ["exp", "iat", "sub", "jti"]})
if claims.get("typ") != kind:
raise jwt.InvalidTokenError(f"expected a {kind} token")
return claims
def new_api_key() -> str:
return "bb_" + secrets.token_urlsafe(24)
def hash_api_key(key: str) -> str:
"""API keys are long and random, so a fast hash is fine (unlike passwords)."""
return hashlib.sha256(key.encode()).hexdigest()
class LoginLimiter:
"""Block a key after `max_failures` failed logins within `window` seconds."""
def __init__(self, max_failures: int = 5, window: float = 60.0,
clock: Callable[[], float] = time.monotonic) -> None:
self.max_failures, self.window, self.clock = max_failures, window, clock
self._failures: dict[str, list[float]] = {}
def retry_after(self, key: str) -> int:
"""Seconds the caller must wait, or 0 if it may try."""
now = self.clock()
recent = [t for t in self._failures.get(key, []) if now - t < self.window]
self._failures[key] = recent
if len(recent) >= self.max_failures:
return math.ceil(self.window - (now - recent[0]))
return 0
def record_failure(self, key: str) -> None:
self._failures.setdefault(key, []).append(self.clock())
def reset(self, key: str) -> None:
self._failures.pop(key, None)Now make a token at a fixed moment, so that the output is the same on every run, and look inside it:
import base64
import json
from datetime import datetime, timezone
from app.security import make_token
def part(segment: str) -> dict:
return json.loads(base64.urlsafe_b64decode(segment + "=" * (-len(segment) % 4)))
FIXED = datetime(2026, 1, 15, 4, 0, tzinfo=timezone.utc)
token = make_token("2", "access", family="demo-session", jti="demo-token", now=FIXED)
head, payload, signature = token.split(".")
print(part(head))
print(part(payload))
print(len(signature), "characters of signature"){'alg': 'HS256', 'typ': 'JWT'}
{'iss': 'bitbyte-auth-demo', 'sub': '2', 'typ': 'access', 'fam': 'demo-session', 'jti': 'demo-token', 'iat': 1768449600, 'exp': 1768450500}
43 characters of signature- The header names the signing algorithm. The payload holds claims: registered ones (
ississuer,subsubject,expexpiry,iatissued-at,jtitoken id, alsoaudaudience andnbfnot-before) and your own (typ,fam). Times are Unix seconds: this token lives 900 of them. - The payload is only encoded, not encrypted: anyone holding the token can read it. Never put passwords, emails, API keys or anything private in it. We also left the user's role out on purpose: the role is looked up on every request, so demoting an admin takes effect at once instead of when their token expires.
- The signature is an HMAC over header and payload with your secret (HS256). Change one character of the payload and it no longer matches. HS256 suits a single backend; when other services must verify tokens without being able to create them, use RS256 or ES256 and publish the public key.
What verification must reject
decode_token pins algorithms=["HS256"], checks the issuer, requires the key claims and rejects a refresh token used as an access token. Try the classic attacks against it. Times here come from the real clock, but the output prints only the error types, which do not change:
import jwt
from datetime import timedelta
from app.security import decode_token
forged_payload = base64.urlsafe_b64encode(json.dumps({**part(payload), "sub": "1"}).encode()).rstrip(b"=").decode()
now = datetime.now(timezone.utc)
cases = {
"valid": make_token("2", "access"),
"expired": make_token("2", "access", now=now - timedelta(hours=1)),
"payload edited": f"{head}.{forged_payload}.{signature}",
"alg none": jwt.encode({"sub": "1", "typ": "access", "iss": "bitbyte-auth-demo", "jti": "x",
"iat": now, "exp": now + timedelta(hours=1)}, key=None, algorithm="none"),
"other secret": jwt.encode({"sub": "1"}, "an-attacker-secret-that-is-long-enough", algorithm="HS256"),
"refresh as access": make_token("2", "refresh"),
}
for name, candidate in cases.items():
try:
decode_token(candidate, "access")
print(f"{name:18} accepted")
except jwt.InvalidTokenError as exc:
print(f"{name:18} rejected: {type(exc).__name__}")valid accepted
expired rejected: ExpiredSignatureError
payload edited rejected: InvalidSignatureError
alg none rejected: InvalidAlgorithmError
other secret rejected: InvalidSignatureError
refresh as access rejected: InvalidTokenErrorOnly the valid token passes. The two to remember are the edited payload (a user promoting themselves to user 1) and alg: none: an "unsigned" token. Old libraries that trusted the header's alg accepted it; PyJWT refuses because you told it which algorithms are allowed. Always pass algorithms=[...] explicitly.
The application
A store holds users, chats, API keys and the revocation sets. It is an object created by create_app() and kept in app.state, so every test gets a fresh one (compare the shared-singleton problem at the end of FastAPI Architecture: Routers, Schemas and Error Handling). In a real service this is your SQLAlchemy session and a Redis set.
# file: app/store.py
"""In-memory data for the demo. Swap for a database and Redis in a real service."""
from app.security import LoginLimiter, passwords
class Store:
def __init__(self) -> None:
self.users: dict[int, dict] = {}
self.chats: dict[int, dict] = {}
self.api_keys: dict[str, str] = {} # sha256(key) -> service name
self.used_refresh: set[str] = set() # refresh jti values that were already exchanged
self.revoked_families: set[str] = set() # login sessions that have been ended
self.limiter = LoginLimiter()
def add_user(self, username: str, password: str, role: str = "user") -> dict:
row = {"id": len(self.users) + 1, "username": username,
"password_hash": passwords.hash(password), "role": role}
self.users[row["id"]] = row
return row
def by_name(self, username: str) -> dict | None:
return next((u for u in self.users.values() if u["username"] == username), None)Now the app. Read it top to bottom: models, the dependency chain (oauth2_scheme → current_claims → current_user → require_role), then the routes.
# file: app/main.py
import os
import uuid
from typing import Annotated
import jwt
from fastapi import Depends, FastAPI, Header, HTTPException, Request
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel, ConfigDict, Field
from app.security import DUMMY_HASH, decode_token, hash_api_key, make_token, passwords
from app.store import Store
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")
class RegisterIn(BaseModel):
model_config = ConfigDict(extra="forbid")
username: str = Field(min_length=3, max_length=30, pattern=r"^[a-z0-9_]+$")
password: str = Field(min_length=8, max_length=128)
class UserOut(BaseModel): # no password_hash here, on purpose
id: int
username: str
role: str
class TokenPair(BaseModel):
access_token: str
refresh_token: str
token_type: str = "bearer"
class RefreshIn(BaseModel):
refresh_token: str
class ChatIn(BaseModel):
title: str = Field(min_length=1, max_length=80)
class ChatOut(BaseModel):
id: int
title: str
owner_id: int
def unauthorized(detail: str = "Could not validate credentials") -> HTTPException:
return HTTPException(401, detail, headers={"WWW-Authenticate": "Bearer"})
def get_store(request: Request) -> Store:
return request.app.state.store
StoreDep = Annotated[Store, Depends(get_store)]
def issue_pair(user: dict, family: str | None = None) -> TokenPair:
family = family or uuid.uuid4().hex
return TokenPair(access_token=make_token(str(user["id"]), "access", family=family),
refresh_token=make_token(str(user["id"]), "refresh", family=family))
def authenticate(form: OAuth2PasswordRequestForm, request: Request, store: Store) -> dict:
"""Check a username and password, with a limit on failed attempts."""
key = f"{form.username}|{request.client.host if request.client else '?'}"
wait = store.limiter.retry_after(key)
if wait:
raise HTTPException(429, "Too many failed logins, try again later", headers={"Retry-After": str(wait)})
user = store.by_name(form.username)
# Always run one hash check, even for an unknown user, so response time does not reveal which names exist.
password_ok = passwords.verify(form.password, user["password_hash"] if user else DUMMY_HASH)
if user is None or not password_ok:
store.limiter.record_failure(key)
raise unauthorized("Incorrect username or password")
store.limiter.reset(key)
return user
def current_claims(token: Annotated[str, Depends(oauth2_scheme)], store: StoreDep) -> dict:
try:
claims = decode_token(token, "access")
except jwt.ExpiredSignatureError:
raise unauthorized("Token expired") from None
except jwt.InvalidTokenError:
raise unauthorized() from None
if claims["fam"] in store.revoked_families:
raise unauthorized("Session ended")
return claims
def current_user(claims: Annotated[dict, Depends(current_claims)], store: StoreDep) -> dict:
user = store.users.get(int(claims["sub"]))
if user is None:
raise unauthorized()
return user
CurrentUser = Annotated[dict, Depends(current_user)]
def require_role(*roles: str):
def checker(user: CurrentUser) -> dict:
if user["role"] not in roles:
raise HTTPException(403, "You do not have permission to do this")
return user
return checker
def owned_chat(chat_id: int, user: CurrentUser, store: StoreDep) -> dict:
"""Resource-level check: the chat must exist AND belong to the caller (or the caller is an admin)."""
chat = store.chats.get(chat_id)
if chat is None or (chat["owner_id"] != user["id"] and user["role"] != "admin"):
raise HTTPException(404, "Chat not found") # same answer for "missing" and "not yours"
return chat
def service_name(store: StoreDep, x_api_key: Annotated[str | None, Header()] = None) -> str:
name = store.api_keys.get(hash_api_key(x_api_key)) if x_api_key else None
if name is None:
raise HTTPException(401, "Invalid or missing API key")
return name
def create_app() -> FastAPI:
app = FastAPI(title="BitByte Auth Demo")
store = Store()
store.add_user("root", "root-pass-1234", role="admin") # demo only: create admins with a CLI command
if key := os.environ.get("INGEST_API_KEY"):
store.api_keys[hash_api_key(key)] = "ingest-service"
app.state.store = store
@app.post("/auth/register", status_code=201)
def register(data: RegisterIn, store: StoreDep) -> UserOut:
if store.by_name(data.username):
raise HTTPException(409, "Username already taken")
return store.add_user(data.username, data.password)
@app.post("/auth/login")
def login(form: Annotated[OAuth2PasswordRequestForm, Depends()], request: Request, store: StoreDep) -> TokenPair:
return issue_pair(authenticate(form, request, store))
@app.post("/auth/refresh")
def refresh(data: RefreshIn, store: StoreDep) -> TokenPair:
try:
claims = decode_token(data.refresh_token, "refresh")
except jwt.InvalidTokenError:
raise unauthorized("Invalid refresh token") from None
if claims["fam"] in store.revoked_families:
raise unauthorized("Session ended")
if claims["jti"] in store.used_refresh:
store.revoked_families.add(claims["fam"]) # replay: someone holds a stolen copy, end the session
raise unauthorized("Refresh token reuse detected, log in again")
user = store.users.get(int(claims["sub"]))
if user is None:
raise unauthorized("Invalid refresh token")
store.used_refresh.add(claims["jti"]) # each refresh token works exactly once
return issue_pair(user, family=claims["fam"])
@app.post("/auth/logout", status_code=204)
def logout(claims: Annotated[dict, Depends(current_claims)], store: StoreDep) -> None:
store.revoked_families.add(claims["fam"])
@app.get("/me")
def me(user: CurrentUser) -> UserOut:
return user
@app.get("/admin/users", dependencies=[Depends(require_role("admin"))])
def list_users(store: StoreDep) -> list[UserOut]:
return list(store.users.values())
@app.post("/chats", status_code=201)
def create_chat(data: ChatIn, user: CurrentUser, store: StoreDep) -> ChatOut:
chat = {"id": len(store.chats) + 1, "title": data.title, "owner_id": user["id"]}
store.chats[chat["id"]] = chat
return chat
@app.get("/chats/{chat_id}")
def get_chat(chat: Annotated[dict, Depends(owned_chat)]) -> ChatOut:
return chat
@app.post("/ingest")
def ingest(service: Annotated[str, Depends(service_name)]) -> dict[str, str]:
return {"accepted": "yes", "service": service}
return app
app = create_app()Run it
Create a service API key (printing only its safe parts) and start the server. The key is shown to its owner once; the server keeps only its SHA-256 hash.
export INGEST_API_KEY=$(uv run python -c "from app.security import new_api_key; print(new_api_key())")
echo "${INGEST_API_KEY:0:3}... (${#INGEST_API_KEY} characters)"
uv run fastapi run --port 8000 > server.log 2>&1 &
echo $! > server.pid
curl -s --retry 20 --retry-connrefused --retry-delay 0 $BASE/docs -o /dev/null -w '%{http_code}\n'bb_... (35 characters)
200Register and log in
curl -s -w ' %{http_code}\n' -X POST $BASE/auth/register -H 'Content-Type: application/json' \
-d '{"username": "nadia", "password": "correct-horse-1"}'
curl -s -w ' %{http_code}\n' -X POST $BASE/auth/register -H 'Content-Type: application/json' \
-d '{"username": "nadia", "password": "another-pass-9"}'
curl -s -o /dev/null -w '%{http_code}\n' -X POST $BASE/auth/register -H 'Content-Type: application/json' \
-d '{"username": "x", "password": "short", "is_admin": true}'{"id":2,"username":"nadia","role":"user"} 201
{"detail":"Username already taken"} 409
422The new user is id 2 (the seeded admin is 1) and the response has no password_hash, though the stored row has one: the output model filters it. A duplicate name is 409, and bad or unexpected input is 422 (extra="forbid" also rejects a client that tries to send is_admin: never let a registration form choose its own role).
Login follows the OAuth2 "password flow" shape that FastAPI's /docs understands: form fields username and password, not JSON. It is fine for your own first-party login; do not offer it to third-party apps (RFC 9700).
curl -s -w ' %{http_code}\n' -X POST $BASE/auth/login -d 'username=nadia' -d 'password=wrong-password'
curl -s -w ' %{http_code}\n' -X POST $BASE/auth/login -d 'username=nobody' -d 'password=wrong-password'
curl -s -X POST $BASE/auth/login -d 'username=nadia' -d 'password=correct-horse-1' > login.json
export ACCESS=$(python -c "import json; print(json.load(open('login.json'))['access_token'])")
export REFRESH=$(python -c "import json; print(json.load(open('login.json'))['refresh_token'])")
python -c "import json; d = json.load(open('login.json')); print(sorted(d), d['token_type'], d['access_token'].count('.'))"{"detail":"Incorrect username or password"} 401
{"detail":"Incorrect username or password"} 401
['access_token', 'refresh_token', 'token_type'] bearer 2A wrong password and an unknown user give the identical answer, so an attacker cannot use the login form to find out which usernames exist (the code even hashes against a dummy value for unknown users so the timing matches). The tokens are random per login, so the page prints only their shape: two tokens, each with two dots.
Use the access token
curl -s -w ' %{http_code}\n' $BASE/me -H "Authorization: Bearer $ACCESS"
curl -s -i $BASE/me | tr -d '\r' | grep -iE '^(HTTP|www-authenticate)|detail'
curl -s -w ' %{http_code}\n' $BASE/me -H "Authorization: Bearer ${ACCESS}x"{"id":2,"username":"nadia","role":"user"} 200
HTTP/1.1 401 Unauthorized
www-authenticate: Bearer
{"detail":"Not authenticated"}
{"detail":"Could not validate credentials"} 401No token: 401 and WWW-Authenticate: Bearer, which tells clients how to authenticate. A damaged token: 401 too, with the same vague message (do not explain to an attacker what exactly was wrong).
Access and refresh tokens, expiry and revocation
A JWT cannot be "taken back": the server checks the signature, not a database, so a stolen token works until it expires. The standard answer is two tokens:
| Access token | Refresh token | |
|---|---|---|
| Sent to | every API endpoint | only /auth/refresh |
| Lifetime | short: 5 to 15 minutes | long: days |
| If stolen | useful for minutes | dangerous, so it is single-use (rotation) and revocable |
| Checked against | signature and expiry (plus a cheap revocation check) | signature, expiry and server-side state |
This app implements the two server-side controls that make revocation possible:
- Rotation: every refresh gives a new pair, and the old refresh token's
jtiis remembered as used. - Reuse detection and a denylist: if a used refresh token shows up again, either the user or a thief is replaying it, and the server cannot tell which, so it ends the whole login session (the
famfamily).logoutdoes the same on purpose. The set of ended sessions is a denylist; entries need to be kept only until the tokens they cover would have expired, which is what a Redis key with a TTL is for (Caching, Rate Limiting and Reliability with Redis).
curl -s -X POST $BASE/auth/refresh -H 'Content-Type: application/json' -d "{\"refresh_token\": \"$REFRESH\"}" > rotated.json
export NEW_REFRESH=$(python -c "import json; print(json.load(open('rotated.json'))['refresh_token'])")
python -c "import os; print('new refresh token differs:', os.environ['NEW_REFRESH'] != os.environ['REFRESH'])"
echo "--- replay the OLD refresh token"
curl -s -w ' %{http_code}\n' -X POST $BASE/auth/refresh -H 'Content-Type: application/json' -d "{\"refresh_token\": \"$REFRESH\"}"
echo "--- the NEW one, and the access token, are dead too"
curl -s -w ' %{http_code}\n' -X POST $BASE/auth/refresh -H 'Content-Type: application/json' -d "{\"refresh_token\": \"$NEW_REFRESH\"}"
curl -s -w ' %{http_code}\n' $BASE/me -H "Authorization: Bearer $ACCESS"new refresh token differs: True
--- replay the OLD refresh token
{"detail":"Refresh token reuse detected, log in again"} 401
--- the NEW one, and the access token, are dead too
{"detail":"Session ended"} 401
{"detail":"Session ended"} 401One replayed refresh token ended the session for everyone holding it, thief and victim alike; the victim logs in again. That is the intended trade. Normal use stays smooth: a client keeps the newest pair and refreshes shortly before the access token expires, or after a 401 Token expired.
Real clients serialise their refresh calls (two tabs refreshing at once look like reuse), and should always discard their tokens on logout as well.
Roles (RBAC) and per-object permissions (BOLA)
Role-based access control maps users to roles and roles to actions: here user and admin, enforced by require_role("admin"). Roles answer "may this kind of user call this endpoint?". They cannot answer "may this user open this chat?". That second question is BOLA (broken object level authorisation, number one in the OWASP API Security Top 10): an endpoint that checks that you are logged in, then trusts any id you type into the URL. In AI products the ids are chats, uploaded documents, embeddings, fine-tuned models and usage records, and a missing owner check leaks other customers' prompts.
The fix is in owned_chat: load the object, compare its owner_id to the caller and treat "not yours" like "does not exist". Try it with three users. The helper script logs in and prints a token:
# file: login.sh
# usage: sh login.sh USERNAME PASSWORD prints an access token
curl -s -X POST "$BASE/auth/login" -d "username=$1" -d "password=$2" | python -c "import sys, json; print(json.load(sys.stdin)['access_token'])"export NADIA=$(sh login.sh nadia correct-horse-1)
export ROOT=$(sh login.sh root root-pass-1234)
curl -s -o /dev/null -X POST $BASE/auth/register -H 'Content-Type: application/json' -d '{"username": "tanvir", "password": "tanvir-pass-77"}'
export TANVIR=$(sh login.sh tanvir tanvir-pass-77)
curl -s -w ' %{http_code}\n' -X POST $BASE/chats -H "Authorization: Bearer $NADIA" -H 'Content-Type: application/json' -d '{"title": "Refund policy questions"}'
for who in NADIA TANVIR ROOT; do
echo -n "$who reads chat 1: "
curl -s -w ' %{http_code}\n' $BASE/chats/1 -H "Authorization: Bearer ${!who}"
done
echo -n "TANVIR reads chat 99: "
curl -s -w ' %{http_code}\n' $BASE/chats/99 -H "Authorization: Bearer $TANVIR"
echo "--- roles"
curl -s -w ' %{http_code}\n' $BASE/admin/users -H "Authorization: Bearer $NADIA"
curl -s -w ' %{http_code}\n' $BASE/admin/users -H "Authorization: Bearer $ROOT"{"id":1,"title":"Refund policy questions","owner_id":2} 201
NADIA reads chat 1: {"id":1,"title":"Refund policy questions","owner_id":2} 200
TANVIR reads chat 1: {"detail":"Chat not found"} 404
ROOT reads chat 1: {"id":1,"title":"Refund policy questions","owner_id":2} 200
TANVIR reads chat 99: {"detail":"Chat not found"} 404
--- roles
{"detail":"You do not have permission to do this"} 403
[{"id":1,"username":"root","role":"admin"},{"id":2,"username":"nadia","role":"user"},{"id":3,"username":"tanvir","role":"user"}] 200Tanvir gets the same 404 for Nadia's chat and for a chat that never existed, so he learns nothing about other people's ids. Using 403 for "not yours" is also acceptable, but it confirms that the object exists. The role check is different: Nadia is a known user calling an admin-only endpoint, so 403 is the honest answer.
Keep the ownership check in one reusable dependency like owned_chat, filter in the database query (WHERE id = :id AND owner_id = :user), and write a "wrong user" test for every route with an id in it. When "admin" is too blunt, roles grow into scopes (chats:read, models:write, as in HTTP in Depth: Idempotency, Caching and Auth).
API keys for services
A user logs in; a nightly ingestion job or another team's service does not. Machines get API keys (or OAuth client credentials). A few rules, all visible in the code above:
- Generate keys with
secrets, long and random, with a recognisable prefix (bb_) so secret scanners and your support team can spot them. - Store only the hash. A fast SHA-256 is fine here because the key is 192 random bits (an attacker cannot guess it, unlike a password). Show the plaintext to its owner once.
- Send it in a header (
X-API-Key), never in the URL. Give each service its own key, so a leak is revoked without breaking the others, and log which service called.
curl -s -w ' %{http_code}\n' -X POST $BASE/ingest
curl -s -w ' %{http_code}\n' -X POST $BASE/ingest -H "X-API-Key: $INGEST_API_KEY"
curl -s -w ' %{http_code}\n' -X POST $BASE/ingest -H "X-API-Key: $NADIA"{"detail":"Invalid or missing API key"} 401
{"accepted":"yes","service":"ingest-service"} 200
{"detail":"Invalid or missing API key"} 401A user's access token is not an API key and vice versa: they are different credentials for different callers, checked by different dependencies.
Cookies or bearer tokens, and CSRF
Where does the browser keep the token?
| Authorization header (bearer) | Cookie | |
|---|---|---|
| Who attaches it | your JavaScript, deliberately | the browser, automatically, to every request to your domain |
| Stolen by XSS | yes, if it lives in localStorage | not with HttpOnly (scripts cannot read it, though they can still make requests) |
| CSRF risk | none: a malicious site cannot make the browser add the header | yes: a malicious site can make the browser send the cookie |
| Best for | mobile apps, scripts, other services, LLM clients, public APIs | a web app on your own domain (the usual backend-for-frontend setup) |
CSRF (cross-site request forgery): while Nadia is logged in, another site makes her browser send POST /chats/1/delete to your domain, and the browser helpfully attaches her cookie. Defences, used together: SameSite=Lax or Strict on the cookie (the browser does not send it on cross-site POSTs), a CSRF token that the page must echo in a header for every unsafe method, and checking the Origin header. Bearer-header APIs need none of this, which is a good reason to prefer them for APIs. Setting the cookie with the attributes that matter:
from fastapi import Response
response = Response()
response.set_cookie("access_token", "<token>", max_age=900, httponly=True, secure=True, samesite="lax", path="/")
print(response.headers["set-cookie"])access_token="<token>"; HttpOnly; Max-Age=900; Path=/; SameSite=lax; SecureHttpOnly hides the cookie from JavaScript, Secure sends it only over HTTPS, SameSite=lax blocks the cross-site POST, and Max-Age matches the token's life. A cookie-based API would read request.cookies in its dependency and also verify a CSRF token; the token logic is the one you already have.
OAuth2 and OIDC: letting a provider log users in
Many products skip passwords entirely: "Sign in with Google/GitHub/Microsoft". HTTP in Depth explained the authorization code flow with PKCE. What your backend adds is the last step. After the redirect it holds an ID token, a JWT signed by the provider with an asymmetric key, and must verify it like any other token before trusting it:
- Fetch the provider's public keys from its JWKS URL (
PyJWKClientcaches them) and pick the key named in the token header. - Verify the signature with the expected algorithm (RS256), the
iss(the provider), theaud(your client id: a token issued to another app is not for you), expiry and thenonceyou sent at the start. - Map the stable
sub(not the email, which can change) to a row in your ownuserstable, creating it on first login, and then issue your own session or JWT pair as above. Roles and object permissions stay in your database either way.
import jwt
from jwt import PyJWKClient
jwks = PyJWKClient("https://accounts.example.com/.well-known/jwks.json") # caches keys
def verify_id_token(id_token: str, *, nonce: str) -> dict:
signing_key = jwks.get_signing_key_from_jwt(id_token)
claims = jwt.decode(id_token, signing_key, algorithms=["RS256"],
audience="my-client-id", issuer="https://accounts.example.com")
if claims.get("nonce") != nonce:
raise jwt.InvalidTokenError("nonce mismatch")
return claimsThis block is not run here: it needs a real provider and PyJWT's cryptography extra (pip install "pyjwt[crypto]"). In production most teams use a maintained library for the redirect dance (Authlib, or an identity service such as Auth0, Clerk, Keycloak or Supabase Auth) and keep the verification rules above.
Login rate limiting
Passwords can be guessed, and leaked password lists from other sites are tried against yours (credential stuffing). Argon2 makes each guess expensive for an attacker, but only a limit makes guessing at scale impossible. The app's LoginLimiter blocks a username-plus-IP pair after 5 failures in 60 seconds, answers 429 with Retry-After, and resets on success. Mallory does not exist as a user, and still gets limited, which is correct:
for i in 1 2 3 4 5 6 7; do
curl -s -o /dev/null -w '%{http_code} ' -X POST $BASE/auth/login -d 'username=mallory' -d "password=guess$i"
done; echo
curl -s -i -X POST $BASE/auth/login -d 'username=mallory' -d 'password=guess8' | tr -d '\r' | grep -icE '^retry-after'
curl -s -o /dev/null -w 'nadia can still log in: %{http_code}\n' -X POST $BASE/auth/login -d 'username=nadia' -d 'password=correct-horse-1'401 401 401 401 401 429 429
1
nadia can still log in: 200Limiting on the username alone would let an attacker lock any user out on purpose, so combine it with the IP (behind a proxy, read the real client address from a trusted header, not from anything the client can fake). In memory the counter works for one process; with several workers it must live in Redis, with the fixed and sliding window techniques of Caching, Rate Limiting and Reliability with Redis.
Stop the server:
kill "$(cat server.pid)"Testing auth
Auth code deserves more tests than most, because its bugs are silent: everything works for the happy user. The rules from Testing Professional APIs apply: build a fresh app per test, test every route with nobody, the wrong user and the right user, and keep a regression test for every vulnerability you fix. Tokens in tests come from make_token with explicit times, never from sleep.
# file: tests/test_auth.py
import os
from datetime import datetime, timedelta, timezone
import jwt
import pytest
from fastapi.testclient import TestClient
os.environ.setdefault("JWT_SECRET", "tutorial-only-secret-change-me-0123456789")
from app.main import create_app # noqa: E402
from app.security import make_token # noqa: E402
@pytest.fixture
def client(monkeypatch):
monkeypatch.setenv("INGEST_API_KEY", "bb_test-key")
with TestClient(create_app()) as tc:
yield tc
def login(client, username, password="correct-horse-1", register=True):
if register:
client.post("/auth/register", json={"username": username, "password": password})
return client.post("/auth/login", data={"username": username, "password": password})
def bearer(token: str) -> dict:
return {"Authorization": f"Bearer {token}"}
def test_register_login_and_me(client):
assert login(client, "nadia").status_code == 200
token = login(client, "nadia", register=False).json()["access_token"]
me = client.get("/me", headers=bearer(token))
assert me.status_code == 200
assert me.json() == {"id": 2, "username": "nadia", "role": "user"} # no password_hash
def test_wrong_password_and_unknown_user_look_the_same(client):
login(client, "nadia")
wrong = login(client, "nadia", password="not-the-password", register=False)
unknown = login(client, "nobody", register=False)
assert wrong.status_code == unknown.status_code == 401
assert wrong.json() == unknown.json()
def bad_token(kind: str) -> str:
now = datetime.now(timezone.utc)
claims = {"iss": "bitbyte-auth-demo", "sub": "1", "typ": "access", "fam": "f", "jti": "j",
"iat": now, "exp": now + timedelta(hours=1)}
return {
"expired": make_token("1", "access", now=now - timedelta(hours=1)),
"other_secret": jwt.encode(claims, "an-attacker-secret-that-is-long-enough", algorithm="HS256"),
"alg_none": jwt.encode(claims, key=None, algorithm="none"),
"refresh_as_access": make_token("1", "refresh"),
"garbage": "not.a.token",
}[kind]
@pytest.mark.parametrize("kind", ["expired", "other_secret", "alg_none", "refresh_as_access", "garbage"])
def test_protected_route_rejects_bad_tokens(client, kind):
resp = client.get("/me", headers=bearer(bad_token(kind)))
assert resp.status_code == 401
assert resp.headers["WWW-Authenticate"] == "Bearer"
def test_refresh_rotation_and_reuse_detection(client):
first = login(client, "nadia").json()
second = client.post("/auth/refresh", json={"refresh_token": first["refresh_token"]})
assert second.status_code == 200
replay = client.post("/auth/refresh", json={"refresh_token": first["refresh_token"]})
assert replay.status_code == 401
# the replay ended the whole session: the newer tokens are dead as well
assert client.post("/auth/refresh", json={"refresh_token": second.json()["refresh_token"]}).status_code == 401
assert client.get("/me", headers=bearer(second.json()["access_token"])).status_code == 401
def test_roles_and_bola(client):
nadia = login(client, "nadia").json()["access_token"]
tanvir = login(client, "tanvir").json()["access_token"]
root = login(client, "root", password="root-pass-1234", register=False).json()["access_token"]
chat = client.post("/chats", json={"title": "Private"}, headers=bearer(nadia)).json()
url = f"/chats/{chat['id']}"
assert client.get(url, headers=bearer(nadia)).status_code == 200
assert client.get(url, headers=bearer(tanvir)).status_code == 404 # BOLA regression test
assert client.get(url, headers=bearer(root)).status_code == 200
assert client.get("/admin/users", headers=bearer(nadia)).status_code == 403
assert client.get("/admin/users", headers=bearer(root)).status_code == 200
def test_login_is_rate_limited(client):
codes = [login(client, "mallory", password=f"guess-{i}-xx", register=False).status_code for i in range(7)]
assert codes == [401] * 5 + [429] * 2
def test_api_key(client):
assert client.post("/ingest").status_code == 401
assert client.post("/ingest", headers={"X-API-Key": "bb_wrong"}).status_code == 401
ok = client.post("/ingest", headers={"X-API-Key": "bb_test-key"})
assert ok.status_code == 200 and ok.json()["service"] == "ingest-service"uv run python -m pytest -q tests........... [100%]
11 passed in 5.60sThe tests never wait for a token to expire (a token issued an hour ago is created directly), and no state leaks between them because every test calls create_app(). In other tests you can skip login with app.dependency_overrides[current_user]; keep these auth tests as the ones that do not.
Where this shows up in AI work
- LLM gateways issue a key per team or service, rate-limit and meter by key, and log the key's name next to token usage, so cost can be traced to an owner.
- Chat products store conversations, uploaded files and embeddings per user. Every retrieval query (RAG) must filter by owner or tenant, or one customer's documents are answered to another: BOLA wearing an AI costume.
- Agents and MCP servers call your API on behalf of a user. Give them short-lived, narrowly scoped tokens (OAuth) rather than a master key, so a prompt-injected agent can do only what that user could do.
Common mistakes
- Fast hashes or plain text for passwords. One leaked table and every account is open. Use Argon2id through
pwdlib, verify withverify, and never log passwords (including in request-body logs). - Not pinning the algorithm, or skipping claim checks.
jwt.decode(token, key)withoutalgorithms, or withverify_signature=False"just to read it", is howalg: noneand forged tokens get in. Always passalgorithms,issuerand required claims. - Long-lived access tokens and no way to revoke. A 30-day access JWT is a 30-day breach. Keep access tokens to minutes, refresh with rotation, and keep a denylist for logout and reuse.
- Checking login but not ownership.
GET /chats/{id}that only requires a valid token is BOLA. Add an owner check to every route that takes an id, and a test with a second user. - Secrets and tokens in the wrong place. A
JWT_SECRETcommitted to Git (Git for AI Projects: Data, Models, LFS, DVC and Leaked Secrets shows the cleanup, and why you rotate the secret first), tokens in URLs or logs, or private data inside the payload. Secrets come from the environment; tokens travel in headers.
Try it yourself
- Easy: make a 5-minute access token for user
"7"at the fixed time2026-01-15 04:00 UTCand print itssuband the number of seconds betweeniatandexp. Then show thatdecode_tokenrejects an access token when a refresh token is expected. - Medium: use
LoginLimiterwith a fake clock (no sleeping) to show that three failures block a key for 60 seconds, and that after 61 seconds it may try again. - Hard: with
TestClient, write the story of a stolen refresh token: log in, refresh once, replay the first refresh token, then use the refresh token from the second step. Print the status codes of the three refresh calls and the error messages.
Answers
# 1. Easy
import os
from datetime import datetime, timedelta, timezone
import jwt
from app.security import decode_token, make_token
fixed = datetime(2026, 1, 15, 4, 0, tzinfo=timezone.utc)
short = make_token("7", "access", now=fixed, ttl=timedelta(minutes=5))
claims = jwt.decode(short, os.environ["JWT_SECRET"], algorithms=["HS256"], options={"verify_exp": False})
print(claims["sub"], claims["exp"] - claims["iat"])
try:
decode_token(make_token("7", "access"), "refresh")
except jwt.InvalidTokenError as exc:
print("rejected:", exc)
# 2. Medium
from app.security import LoginLimiter
clock = [0.0]
limiter = LoginLimiter(max_failures=3, window=60, clock=lambda: clock[0])
for _ in range(3):
limiter.record_failure("nadia|10.0.0.1")
print("blocked for", limiter.retry_after("nadia|10.0.0.1"), "seconds")
clock[0] = 61
print("after 61 s:", limiter.retry_after("nadia|10.0.0.1"))7 300
rejected: expected a refresh token
blocked for 60 seconds
after 61 s: 0# 3. Hard
from fastapi.testclient import TestClient
from app.main import create_app
with TestClient(create_app()) as tc:
tc.post("/auth/register", json={"username": "nadia", "password": "correct-horse-1"})
first = tc.post("/auth/login", data={"username": "nadia", "password": "correct-horse-1"}).json()
second = tc.post("/auth/refresh", json={"refresh_token": first["refresh_token"]})
replay = tc.post("/auth/refresh", json={"refresh_token": first["refresh_token"]})
newest = tc.post("/auth/refresh", json={"refresh_token": second.json()["refresh_token"]})
print(second.status_code, replay.status_code, newest.status_code)
print(replay.json()["detail"], "|", newest.json()["detail"])200 401 401
Refresh token reuse detected, log in again | Session endedSummary
- Authentication answers "who", authorisation answers "may you"; return 401 with
WWW-Authenticatefor the first, 403 or 404 for the second, and 429 withRetry-Afterfor too many guesses. - Hash passwords with Argon2id (
pwdlib), make unknown users look like wrong passwords, and never return or log hashes. - A JWT is signed, not encrypted: keep claims minimal, pin
algorithms, checkiss,expand token type, and keep the secret in the environment. - Use short access tokens plus single-use rotating refresh tokens, with a denylist of ended sessions for logout and replay detection.
- Combine roles (RBAC) with a per-object ownership check on every route with an id to stop BOLA; give services their own hashed API keys; prefer bearer headers for APIs, and add SameSite and CSRF defences if you use cookies.
- Delegate login to an OIDC provider by verifying its ID token (signature,
iss,aud,nonce) and mappingsubto your own user; rate-limit logins; test every route as nobody, the wrong user and the right user.
Next: Caching, Rate Limiting and Reliability with Redis moves the denylist and the login counters into Redis, and adds caching, proper rate limits, timeouts, circuit breakers and request IDs.