In this post
What this post covers
Authentication answers “who is this?”; authorization (part 18) answers “may they do this?”. FastAPI gives you the building blocks as dependencies: OAuth2 password and bearer flows, API-key headers, and security scopes that also appear in OpenAPI. The cryptography itself comes from libraries: bcrypt or argon2 for passwords, PyJWT for tokens. Every example here runs, including the attacks that a careless implementation lets through.
A festival: ID check at the gate, then a wristband
At the gate you show ID once (your password). The festival gives you a wristband (a token) that says who you are and which areas you may enter, and it expires on Sunday night. Staff inside only check the wristband: they can see it is genuine (signed), but anyone can read what is printed on it (JWTs are not encrypted). Backstage passes are separate wristbands with narrower rights (scopes).
1. Storing passwords
hash with a slow, salted algorithm; verify in constant time
import time
import bcrypt
def hash_password(password: str) -> str:
pw = password.encode()
if len(pw) > 72: # bcrypt's hard limit (bcrypt 5.0 raises instead of truncating)
raise ValueError("password too long: max 72 bytes")
return bcrypt.hashpw(pw, bcrypt.gensalt(rounds=12)).decode()
def verify_password(password: str, stored: str) -> bool:
return bcrypt.checkpw(password.encode(), stored.encode())
start = time.perf_counter()
stored = hash_password("correct horse battery staple")
print(f"hash: {stored[:29]}… ({(time.perf_counter() - start) * 1000:.0f} ms: deliberately slow)")
print("same password, different hash (random salt):", hash_password("correct horse battery staple") != stored)
print("verify right:", verify_password("correct horse battery staple", stored),
"| verify wrong:", verify_password("Correct horse battery staple", stored))
try:
hash_password("🔐" * 30) # 30 emoji = 120 bytes of UTF-8
except ValueError as e:
print("ValueError:", e)
Output
hash: $2b$12$F6RJOMW0PR3iwNSxZfirJO… (443 ms: deliberately slow)
same password, different hash (random salt): True
verify right: True | verify wrong: False
ValueError: password too long: max 72 bytes
| Rule | Detail |
|---|---|
| Use a password-hashing function, not a fast hash | argon2id (via argon2-cffi or pwdlib) is today's recommendation; bcrypt is still fine. Never SHA-256 or MD5 for passwords. |
| Hash off the event loop | ~0.1–0.5 s of CPU per hash: in async def use run_in_threadpool, or make login a def endpoint (part 11) |
| Handle bcrypt's 72-byte limit | bcrypt 5 raises for longer inputs (older versions truncated silently). Cap password length in validation, or use argon2. |
| Avoid passlib in new code | it is unmaintained and emits warnings with bcrypt 4+. Use the bcrypt package directly, or pwdlib. |
| Same answer for "no such user" and "wrong password" | and comparable timing, so the endpoint cannot be used to discover which emails have accounts |
2. Login and bearer tokens (OAuth2 password flow + JWT)
/token issues a short-lived JWT; a dependency validates it on every request
from datetime import datetime, timedelta, timezone
from typing import Annotated
from uuid import uuid4
import bcrypt
import jwt
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from fastapi.testclient import TestClient
SECRET = "change-me-to-32+-random-bytes-from-settings" # from settings (SecretStr), never in code
ALGORITHM, ISSUER, AUDIENCE = "HS256", "https://auth.shop.example", "orders-api"
USERS = {"[email protected]": {"id": 7, "hash": bcrypt.hashpw(b"s3cret-pass", bcrypt.gensalt(10)), "scopes": ["orders:read"]}}
def create_access_token(user_id: int, scopes: list[str], minutes: int = 15) -> str:
now = datetime.now(timezone.utc)
claims = {"sub": str(user_id), "scope": " ".join(scopes), "iat": now, "exp": now + timedelta(minutes=minutes),
"iss": ISSUER, "aud": AUDIENCE, "jti": uuid4().hex}
return jwt.encode(claims, SECRET, algorithm=ALGORITHM)
oauth2 = OAuth2PasswordBearer(tokenUrl="token") # reads "Authorization: Bearer …", documents it in /docs
UNAUTHORIZED = HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid or expired token",
headers={"WWW-Authenticate": "Bearer"})
def current_user(token: Annotated[str, Depends(oauth2)]) -> dict:
try:
claims = jwt.decode(token, SECRET, algorithms=[ALGORITHM], # PIN the algorithm
audience=AUDIENCE, issuer=ISSUER, leeway=10,
options={"require": ["exp", "iat", "sub", "aud", "iss"]})
except jwt.PyJWTError:
raise UNAUTHORIZED from None
return {"id": int(claims["sub"]), "scopes": claims["scope"].split()}
app = FastAPI()
@app.post("/token")
def login(form: Annotated[OAuth2PasswordRequestForm, Depends()]): # def: bcrypt runs in the thread pool
user = USERS.get(form.username.lower())
if user is None or not bcrypt.checkpw(form.password.encode(), user["hash"]):
raise HTTPException(401, "incorrect email or password", headers={"WWW-Authenticate": "Bearer"})
return {"access_token": create_access_token(user["id"], user["scopes"]), "token_type": "bearer",
"expires_in": 900}
@app.get("/me")
def me(user: Annotated[dict, Depends(current_user)]):
return user
client = TestClient(app)
bad = client.post("/token", data={"username": "[email protected]", "password": "guess"})
print("wrong password:", bad.status_code, bad.json()["detail"])
token = client.post("/token", data={"username": "[email protected]", "password": "s3cret-pass"}).json()["access_token"]
print("token:", token[:40], "…")
print("/me:", client.get("/me", headers={"Authorization": f"Bearer {token}"}).json())
print("no token:", client.get("/me").status_code, client.get("/me").headers.get("www-authenticate"))
Output
wrong password: 401 incorrect email or password
token: eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ …
/me: {'id': 7, 'scopes': ['orders:read']}
no token: 401 Bearer
3. JWT attacks you must reject
alg=none, forged payload, expired token, wrong audience — and what a JWT reveals
import base64
import json
from datetime import datetime, timedelta, timezone
import jwt
SECRET, AUD = "server-secret-32-bytes-minimum!!", "orders-api"
def decode(token):
try:
return "accepted", jwt.decode(token, SECRET, algorithms=["HS256"], audience=AUD)
except jwt.PyJWTError as e:
return "rejected", type(e).__name__
now = datetime.now(timezone.utc)
good = jwt.encode({"sub": "7", "role": "customer", "aud": AUD, "exp": now + timedelta(minutes=5)}, SECRET)
print("valid: ", decode(good)[0])
none_alg = jwt.encode({"sub": "7", "role": "admin", "aud": AUD}, key=None, algorithm="none")
print("alg=none: ", decode(none_alg))
header, payload, signature = good.split(".")
forged_payload = base64.urlsafe_b64encode(json.dumps({"sub": "7", "role": "admin", "aud": AUD,
"exp": int((now + timedelta(minutes=5)).timestamp())}).encode()).rstrip(b"=").decode()
print("forged payload: ", decode(f"{header}.{forged_payload}.{signature}"))
expired = jwt.encode({"sub": "7", "aud": AUD, "exp": now - timedelta(minutes=1)}, SECRET)
print("expired: ", decode(expired))
other_api = jwt.encode({"sub": "7", "aud": "billing-api", "exp": now + timedelta(minutes=5)}, SECRET)
print("wrong audience: ", decode(other_api))
padded = payload + "=" * (-len(payload) % 4)
print("anyone can READ a JWT:", json.loads(base64.urlsafe_b64decode(padded)))
Output
valid: accepted
alg=none: ('rejected', 'InvalidAlgorithmError')
forged payload: ('rejected', 'InvalidSignatureError')
expired: ('rejected', 'ExpiredSignatureError')
wrong audience: ('rejected', 'InvalidAudienceError')
anyone can READ a JWT: {'sub': '7', 'role': 'customer', 'aud': 'orders-api', 'exp': 1789803584}
Always pass algorithms=[...] explicitly. Validate exp, aud and
iss. Keep access tokens short-lived (5–15 minutes). Never put secrets or personal data in claims:
signing proves integrity, it does not hide anything. With HS256, every service holding the secret can also
mint tokens. When several services verify tokens, use an identity provider with asymmetric keys
(RS256/ES256) so services only need the public key.
4. Refresh tokens, revocation and logout
| Concern | Pattern |
|---|---|
| Short access tokens without constant logins | a long-lived refresh token (days), stored server-side (or its hash), exchanged at /token/refresh for a new access token |
| Stolen refresh token | rotation: each refresh returns a new refresh token and invalidates the old; reuse of an old one revokes the whole family |
| Logout / "log out everywhere" / password change | delete refresh tokens, and bump a per-user token_version claim that access-token validation checks |
| Browser storage | refresh token in an HttpOnly; Secure; SameSite cookie, not in localStorage (readable by any injected script) |
| Immediate revocation of access tokens | a deny-list of jti values in Redis, checked until the token's natural expiry |
5. API keys for machine clients
prefixed keys, stored hashed, compared in constant time
import hashlib
import hmac
import secrets
from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException, Security
from fastapi.security import APIKeyHeader
from fastapi.testclient import TestClient
def new_api_key():
prefix, secret = secrets.token_hex(4), secrets.token_urlsafe(24)
return f"sk_live_{prefix}_{secret}", prefix, hashlib.sha256(secret.encode()).hexdigest()
full_key, prefix, digest = new_api_key() # show full_key to the customer ONCE
KEYS = {prefix: {"hash": digest, "client": "partner-logistics", "active": True}}
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
def api_client(key: Annotated[str | None, Security(api_key_header)]) -> str:
try:
_, _, key_prefix, secret = key.split("_", 3)
record = KEYS[key_prefix]
except (AttributeError, ValueError, KeyError):
raise HTTPException(401, "invalid API key") from None
candidate = hashlib.sha256(secret.encode()).hexdigest()
if not (record["active"] and hmac.compare_digest(candidate, record["hash"])): # constant-time comparison
raise HTTPException(401, "invalid API key")
return record["client"]
app = FastAPI()
@app.get("/partner/shipments")
def shipments(client_name: Annotated[str, Depends(api_client)]):
return {"client": client_name, "shipments": 3}
c = TestClient(app)
print(c.get("/partner/shipments", headers={"X-API-Key": full_key}).json())
print(c.get("/partner/shipments", headers={"X-API-Key": full_key[:-1] + "x"}).status_code,
c.get("/partner/shipments").status_code)
print("what the database stores:", {prefix: KEYS[prefix]["hash"][:16] + "…"})
Output
{'client': 'partner-logistics', 'shipments': 3}
401 401
what the database stores: {'a8b476fe': 'b40f9eeef6d6219b…'}
6. Security scopes
Security(...) declares required scopes per endpoint, and they appear in OpenAPI
from typing import Annotated
from fastapi import FastAPI, HTTPException, Security
from fastapi.security import OAuth2PasswordBearer, SecurityScopes
from fastapi.testclient import TestClient
oauth2 = OAuth2PasswordBearer(tokenUrl="token", scopes={"orders:read": "Read orders",
"orders:write": "Create and change orders"})
TOKENS = {"reader": ["orders:read"], "writer": ["orders:read", "orders:write"]} # stand-in for decoded JWTs
def current_user(security_scopes: SecurityScopes, token: Annotated[str, Security(oauth2)]):
granted = TOKENS.get(token)
if granted is None:
raise HTTPException(401, "invalid token", headers={"WWW-Authenticate": "Bearer"})
missing = [s for s in security_scopes.scopes if s not in granted]
if missing:
raise HTTPException(403, f"missing scopes: {missing}",
headers={"WWW-Authenticate": f'Bearer scope="{security_scopes.scope_str}"'})
return {"token": token, "scopes": granted}
app = FastAPI()
@app.get("/orders")
def list_orders(user: Annotated[dict, Security(current_user, scopes=["orders:read"])]):
return {"orders": [], "by": user["token"]}
@app.post("/orders")
def create_order(user: Annotated[dict, Security(current_user, scopes=["orders:write"])]):
return {"created": True}
client = TestClient(app)
auth = lambda t: {"Authorization": f"Bearer {t}"}
print(client.get("/orders", headers=auth("reader")).status_code, client.post("/orders", headers=auth("reader")).json())
print(client.post("/orders", headers=auth("writer")).json())
print(client.get("/openapi.json").json()["paths"]["/orders"]["post"]["security"])
Output
200 {'detail': "missing scopes: ['orders:write']"}
{'created': True}
[{'OAuth2PasswordBearer': ['orders:write']}]
7. Choosing an approach
| Clients | Recommended |
|---|---|
| your own browser front end, same site | server-side sessions in an HttpOnly cookie, or short JWT + refresh cookie. Add CSRF protection when using cookies. |
| mobile apps and SPAs across services | an OpenID Connect provider (Keycloak, Auth0, Entra ID, Cognito): your API only verifies RS256 tokens using the provider's JWKS |
| partner servers and scripts | API keys (hashed, prefixed, revocable, scoped) or OAuth2 client credentials |
| service to service inside your platform | mTLS or short-lived tokens from your identity platform |
verifying tokens from an OIDC provider (not run: needs the provider's JWKS URL)
import jwt
jwks = jwt.PyJWKClient("https://login.example.com/.well-known/jwks.json", cache_keys=True)
def verify(token: str) -> dict:
signing_key = jwks.get_signing_key_from_jwt(token) # picks the key by the token's "kid"
return jwt.decode(token, signing_key.key, algorithms=["RS256"],
audience="orders-api", issuer="https://login.example.com/")
Recap
- Passwords: argon2id or bcrypt, hashed off the event loop, 72-byte bcrypt limit handled, identical errors for unknown user and wrong password.
- JWTs: pinned algorithms, validated exp/aud/iss, short lifetimes, no secrets in claims; refresh tokens with rotation for sessions.
- API keys: prefixed, shown once, stored hashed, compared with
hmac.compare_digest, revocable. - Security scopes declare per-endpoint requirements and document them in OpenAPI. Prefer an OIDC provider as systems grow.
Get the free RAG production checklist
25 checks to run before real users see your RAG feature: exact-match misses, invented citations, prompt injection through documents, quality that slips after a prompt change. Confirm your email and the PDF comes with the welcome email, plus each new experiment.