🟩 FastAPI

Auth: OAuth2 & JWT

📖 Lesson 40 ⏱ 50 min 🧪 5 questions 💻 3 exercises

🎯 Learning Objectives

Understand the difference between authentication and authorisation
Explain the OAuth2 Password Flow and when to use it
Understand JWT structure (header, payload, signature) and how claims work
Issue JWTs with expiry using python-jose or PyJWT
Hash and verify passwords with passlib and bcrypt
Build a complete login endpoint that issues access tokens
Protect routes with a get_current_user dependency that validates JWTs

Authentication vs Authorisation

Authentication (AuthN): who are you? — verifying identity (login, token validation).

Authorisation (AuthZ): what can you do? — checking permissions/roles after identity is confirmed.

Strategy	How it works	Best for
Session cookies	Server stores session; browser sends cookie	Traditional web apps
API keys	Static secret in header	Server-to-server, simple APIs
OAuth2 + JWT	Short-lived signed tokens, stateless	SPAs, mobile apps, microservices
OAuth2 + refresh tokens	Access token (short) + refresh token (long)	Production apps needing token rotation

JWT-based auth is stateless — the server stores no session. The token itself contains the claims; any server can verify it with the secret key. This is what makes it ideal for microservices and horizontal scaling.

JWT Structure

A JWT consists of three parts separated by dots: header.payload.signature

Header (base64url-encoded JSON)

{ "alg": "HS256", "typ": "JWT" }

header.json

Payload (base64url-encoded JSON) — standard claims

{
  "sub": "42",
  "email": "alice@example.com",
  "role": "admin",
  "exp": 1719878400,
  "iat": 1719792000,
  "jti": "unique-token-id"
}

payload.json

Standard claims explained:

sub — subject (user id)
exp — expiry (Unix timestamp)
iat — issued-at
jti — JWT ID (for revocation)
iss — issuer
aud — audience

Signature

HMACSHA256(base64url(header) + "." + base64url(payload), secret)

Decoding (NOT verifying) a token

import base64, json

token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiI0MiIsInJvbGUiOiJhZG1pbiIsImV4cCI6MTcxOTg3ODQwMH0.SIGNATURE"
header_b64, payload_b64, sig = token.split(".")
# Add padding
payload_b64 += "=" * (-len(payload_b64) % 4)
print(json.loads(base64.urlsafe_b64decode(payload_b64)))
# {'sub': '42', 'role': 'admin', 'exp': 1719878400}

decode_demo.py

The payload is base64-encoded, not encrypted — anyone can read it. Never put passwords, credit card numbers, or secrets in JWT claims. The signature only proves the token wasn't tampered with.

Password Hashing with passlib

pip install passlib[bcrypt]

terminal

from passlib.context import CryptContext

# bcrypt is the recommended algorithm — slow by design (resists brute force)
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def hash_password(plain: str) -> str:
    return pwd_context.hash(plain)

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

# Usage
hashed = hash_password("MySecret123")
print(hashed)                                  # $2b$12$...  (bcrypt hash)
print(verify_password("MySecret123", hashed))  # True
print(verify_password("WrongPass",   hashed))  # False

hashing.py

Never store plain-text passwords. Never use MD5 or SHA-256 for passwords — they are fast, making brute-force trivial. bcrypt, argon2, or scrypt are purposely slow. The deprecated="auto" flag automatically rehashes old passwords on next login if you upgrade the algorithm.

Issuing JWTs with PyJWT

pip install PyJWT

terminal

import jwt
from datetime import datetime, timedelta, timezone
from typing import Any

SECRET_KEY = "your-256-bit-secret-change-in-production"  # use secrets.token_hex(32)
ALGORITHM  = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

def create_access_token(
    subject: str | int,
    extra_claims: dict[str, Any] | None = None,
    expires_delta: timedelta | None = None,
) -> str:
    expire = datetime.now(timezone.utc) + (
        expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    payload = {
        "sub": str(subject),
        "exp": expire,
        "iat": datetime.now(timezone.utc),
    }
    if extra_claims:
        payload.update(extra_claims)
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

def decode_access_token(token: str) -> dict:
    try:
        return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    except jwt.ExpiredSignatureError:
        raise ValueError("Token has expired")
    except jwt.InvalidTokenError as e:
        raise ValueError(f"Invalid token: {e}")

# Issue a token
token = create_access_token(subject=42, extra_claims={"role": "admin"})
print(token)

# Verify and decode
claims = decode_access_token(token)
print(claims["sub"])   # "42"
print(claims["role"])  # "admin"

token_utils.py

Generate a secure secret key:

import secrets
print(secrets.token_hex(32))  # 64-char hex — use this as SECRET_KEY

generate_key.py

Store SECRET_KEY in an environment variable (via pydantic-settings), never in source code. Rotate keys periodically by supporting multiple keys during a transition period.

The OAuth2 Password Flow

Client                    FastAPI
  │                          │
  │  POST /auth/token         │
  │  {username, password}    │
  │ ─────────────────────── ▶│
  │                          │  1. Lookup user by username
  │                          │  2. verify_password(plain, hashed)
  │                          │  3. create_access_token(user.id)
  │  {access_token, type}    │
  │ ◀─────────────────────── │
  │                          │
  │  GET /me                  │
  │  Authorization: Bearer …  │
  │ ─────────────────────── ▶│
  │                          │  4. Extract token from header
  │                          │  5. decode_access_token(token)
  │                          │  6. Lookup user by sub claim
  │  {user data}             │
  │ ◀─────────────────────── │

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel
from passlib.context import CryptContext
import jwt
from datetime import datetime, timedelta, timezone

app = FastAPI()
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")

SECRET_KEY = "change-me-in-production"
ALGORITHM  = "HS256"
EXPIRE_MIN = 30

# ── Fake user DB ──
class UserInDB(BaseModel):
    id: int
    username: str
    email: str
    hashed_password: str
    role: str = "user"

USERS_DB: dict[str, UserInDB] = {
    "alice": UserInDB(
        id=1, username="alice", email="alice@example.com",
        hashed_password=pwd_context.hash("secret123"),
        role="admin",
    ),
}

# ── Helpers ──
def get_user(username: str) -> UserInDB | None:
    return USERS_DB.get(username)

def authenticate_user(username: str, password: str) -> UserInDB | None:
    user = get_user(username)
    if not user or not pwd_context.verify(password, user.hashed_password):
        return None
    return user

def create_token(subject: str, role: str) -> str:
    expire = datetime.now(timezone.utc) + timedelta(minutes=EXPIRE_MIN)
    return jwt.encode({"sub": subject, "role": role, "exp": expire}, SECRET_KEY, ALGORITHM)

# ── Token endpoint ──
class Token(BaseModel):
    access_token: str
    token_type: str = "bearer"

@app.post("/auth/token", response_model=Token)
def login(form: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(form.username, form.password)
    if not user:
        raise HTTPException(
            status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    token = create_token(str(user.id), user.role)
    return Token(access_token=token)

main.py

Protecting Routes with `get_current_user`

class UserOut(BaseModel):
    id: int
    username: str
    email: str
    role: str

def get_current_user(token: str = Depends(oauth2_scheme)) -> UserInDB:
    credentials_exception = HTTPException(
        status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: str = payload.get("sub")
        if user_id is None:
            raise credentials_exception
    except jwt.InvalidTokenError:
        raise credentials_exception

    # Look up user (in production: DB query by id)
    user = next((u for u in USERS_DB.values() if str(u.id) == user_id), None)
    if user is None:
        raise credentials_exception
    return user

def require_admin(user: UserInDB = Depends(get_current_user)) -> UserInDB:
    if user.role != "admin":
        raise HTTPException(status.HTTP_403_FORBIDDEN, detail="Admin only")
    return user

@app.get("/me", response_model=UserOut)
def me(user: UserInDB = Depends(get_current_user)):
    return user

@app.get("/admin/dashboard")
def dashboard(_: UserInDB = Depends(require_admin)):
    return {"message": "Welcome, admin"}

main.py

Refresh Tokens

Access tokens are short-lived (15–60 min); refresh tokens are long-lived (7–30 days) and stored securely (httpOnly cookie). When the access token expires, the client uses the refresh token to get a new one without re-logging in.

from fastapi import Response, Cookie
import secrets

REFRESH_TOKEN_EXPIRE_DAYS = 7
refresh_token_store: dict[str, str] = {}   # token → user_id (use Redis in production)

def create_refresh_token(user_id: str) -> str:
    token = secrets.token_urlsafe(32)
    refresh_token_store[token] = user_id
    return token

@app.post("/auth/refresh")
def refresh(
    response: Response,
    refresh_token: str = Cookie(default=None),
):
    if not refresh_token or refresh_token not in refresh_token_store:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail="Invalid refresh token")

    user_id = refresh_token_store[refresh_token]
    # Rotate: invalidate old, issue new refresh token
    del refresh_token_store[refresh_token]
    new_refresh = create_refresh_token(user_id)

    response.set_cookie(
        key="refresh_token",
        value=new_refresh,
        httponly=True,      # not accessible via JS
        secure=True,        # HTTPS only
        samesite="lax",
        max_age=60 * 60 * 24 * REFRESH_TOKEN_EXPIRE_DAYS,
    )

    # Also issue a new access token
    user = next((u for u in USERS_DB.values() if str(u.id) == user_id), None)
    new_access = create_token(user_id, user.role if user else "user")
    return Token(access_token=new_access)

main.py

Always rotate refresh tokens on use (issue a new one, invalidate the old one). This limits the window of exposure if a refresh token is stolen. Store refresh tokens in Redis or a DB — never in a JWT (that defeats revocability).

Token Revocation & Blocklists

JWTs are stateless — once issued, they are valid until they expire. To revoke a token before expiry (on logout, password change, or compromise), you need a blocklist.

import redis
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
import jwt
from datetime import datetime, timezone

app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")

# Redis blocklist — store revoked JTI (JWT ID) until expiry
r = redis.Redis(host="localhost", port=6379, decode_responses=True)

SECRET_KEY = "change-me"
ALGORITHM  = "HS256"

def get_current_user(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    except jwt.InvalidTokenError:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail="Invalid token")

    jti = payload.get("jti")
    if jti and r.exists(f"blocklist:{jti}"):
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail="Token has been revoked")

    return payload

@app.post("/auth/logout")
def logout(payload: dict = Depends(get_current_user)):
    jti = payload.get("jti")
    exp = payload.get("exp", 0)
    ttl = max(0, exp - int(datetime.now(timezone.utc).timestamp()))
    if jti and ttl > 0:
        r.setex(f"blocklist:{jti}", ttl, "revoked")
    return {"message": "Logged out"}

revocation.py

Always include a jti (JWT ID) claim — a uuid4() or secrets.token_urlsafe(16) — when issuing tokens you may need to revoke. Set the Redis TTL equal to the token's remaining lifetime so blocklist entries auto-expire and don't accumulate forever.

HTTPS, CORS & Security Headers

Auth tokens are only as safe as the transport layer. FastAPI makes it easy to add CORS restrictions and security headers.

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from starlette.middleware.httpsredirect import HTTPSRedirectMiddleware

app = FastAPI()

# ── HTTPS redirect (production only) ──
# app.add_middleware(HTTPSRedirectMiddleware)

# ── CORS — restrict which origins can call your API ──
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://myapp.com", "https://www.myapp.com"],
    allow_credentials=True,     # needed to send cookies (refresh tokens)
    allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE"],
    allow_headers=["Authorization", "Content-Type"],
)

# ── Security headers middleware ──
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request

class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        response = await call_next(request)
        response.headers["X-Content-Type-Options"] = "nosniff"
        response.headers["X-Frame-Options"]        = "DENY"
        response.headers["Referrer-Policy"]        = "strict-origin-when-cross-origin"
        return response

app.add_middleware(SecurityHeadersMiddleware)

middleware.py

Never use allow_origins=["*"] with allow_credentials=True — browsers will block it and it defeats the purpose of CORS. Always list explicit allowed origins in production. In development, allow_origins=["http://localhost:3000"] is fine.

Complete Auth Module Layout

A clean, production-ready auth module layout for a FastAPI project:

myapi/
├── auth/
│   ├── __init__.py
│   ├── router.py        ← /auth/token, /auth/refresh, /auth/logout
│   ├── dependencies.py  ← get_current_user, require_admin, require_role
│   ├── schemas.py       ← Token, TokenData, UserCreate, UserOut
│   ├── service.py       ← authenticate_user, create_token, hash/verify password
│   └── models.py        ← SQLAlchemy User model (Lesson 41)
├── config.py            ← Settings (SECRET_KEY, EXPIRE_MIN, etc.)
└── main.py

project layout

# auth/dependencies.py — the three reusable auth deps
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
import jwt
from .service import decode_token
from .schemas import TokenData

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")

def get_current_user(token: str = Depends(oauth2_scheme)) -> TokenData:
    try:
        payload = decode_token(token)
        return TokenData(**payload)
    except ValueError as e:
        raise HTTPException(
            status.HTTP_401_UNAUTHORIZED,
            detail=str(e),
            headers={"WWW-Authenticate": "Bearer"},
        )

def require_role(role: str):
    def _checker(user: TokenData = Depends(get_current_user)):
        if user.role != role:
            raise HTTPException(status.HTTP_403_FORBIDDEN, detail=f"{role} required")
        return user
    return _checker

require_admin = require_role("admin")

auth/dependencies.py

Best Practices

Use short-lived access tokens — 15 to 60 minutes maximum. Pair with refresh tokens for seamless UX without long-lived, hard-to-revoke tokens.
Always include exp in every JWT — tokens without an expiry claim are valid forever and a security risk.
Include a jti claim for any token that may need revocation (logout, password reset, account suspension).
Store the SECRET_KEY in an environment variable — never hardcode it. Use secrets.token_hex(32) to generate a strong key.
Use bcrypt (or argon2) for password hashing — never MD5, SHA-1, or unsalted SHA-256.
Store refresh tokens server-side (Redis or DB) so they can be revoked. An httpOnly + Secure cookie is the safest client-side storage option.
Never put sensitive data in JWT claims — the payload is only base64-encoded, not encrypted. Include only what is needed (user id, role) — never passwords, SSNs, or financial data.
Restrict CORS origins explicitly — allow_origins=["*"] with credentials is blocked by browsers and is insecure.

Exercises

Exercise 1 — Complete Login & Protected Routes

Build a working auth system from scratch:

Create an in-memory user store with two users: alice (admin) and bob (user), both with bcrypt-hashed passwords.
Implement POST /auth/token using OAuth2PasswordRequestForm — return a signed JWT with sub, role, and exp.
Implement GET /me — returns the current user's id, username, and role (no password).
Implement GET /admin/stats — returns {"users": 2} but raises 403 for non-admins.
Test all three endpoints using TestClient: successful login, wrong password (401), user accessing admin route (403), admin accessing admin route (200).

💡 Hint — test structure

from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def get_token(username, password):
    resp = client.post("/auth/token", data={"username": username, "password": password})
    assert resp.status_code == 200
    return resp.json()["access_token"]

def test_wrong_password():
    resp = client.post("/auth/token", data={"username": "alice", "password": "wrong"})
    assert resp.status_code == 401

def test_me():
    token = get_token("alice", "secret123")
    resp = client.get("/me", headers={"Authorization": f"Bearer {token}"})
    assert resp.status_code == 200
    assert resp.json()["username"] == "alice"

def test_user_cannot_access_admin():
    token = get_token("bob", "bobsecret")
    resp = client.get("/admin/stats", headers={"Authorization": f"Bearer {token}"})
    assert resp.status_code == 403

def test_admin_can_access_admin():
    token = get_token("alice", "secret123")
    resp = client.get("/admin/stats", headers={"Authorization": f"Bearer {token}"})
    assert resp.status_code == 200

Exercise 2 — Token Expiry & Refresh Flow

Extend the auth system with refresh tokens:

Modify POST /auth/token to also set an httpOnly cookie named refresh_token containing a secrets.token_urlsafe(32) value (stored server-side in a dict).
Implement POST /auth/refresh — reads the cookie, validates it, rotates it (delete old, issue new), and returns a new access token.
Implement POST /auth/logout — removes the refresh token from the store and clears the cookie.
Write tests for the refresh flow: login → get tokens → call refresh → confirm old refresh token is invalid → confirm new access token works.

💡 Hint — setting and reading cookies

from fastapi import Response, Cookie

@app.post("/auth/token")
def login(response: Response, form: OAuth2PasswordRequestForm = Depends()):
    # ... validate user ...
    refresh = secrets.token_urlsafe(32)
    refresh_store[refresh] = str(user.id)
    response.set_cookie("refresh_token", refresh, httponly=True, secure=False, samesite="lax")
    return Token(access_token=create_token(str(user.id), user.role))

@app.post("/auth/refresh")
def refresh(response: Response, refresh_token: str = Cookie(default=None)):
    if not refresh_token or refresh_token not in refresh_store:
        raise HTTPException(401, "Invalid refresh token")
    user_id = refresh_store.pop(refresh_token)
    new_refresh = secrets.token_urlsafe(32)
    refresh_store[new_refresh] = user_id
    response.set_cookie("refresh_token", new_refresh, httponly=True, secure=False, samesite="lax")
    return Token(access_token=create_token(user_id, "user"))

Exercise 3 — Password Change with Token Invalidation

Implement a secure password-change endpoint:

Add POST /me/password — accepts {"current_password": "…", "new_password": "…"}.
Verify the current password; reject with 401 if wrong.
Enforce the new password complexity rules (min 8 chars, at least one uppercase, one digit).
After changing the password, invalidate all existing refresh tokens for that user (remove them from the store).
Return 200 with {"message": "Password updated. Please log in again."}.
Write tests confirming: wrong current password → 401, weak new password → 422, successful change → old refresh token invalid.

💡 Hint

from pydantic import BaseModel, field_validator

class PasswordChange(BaseModel):
    current_password: str
    new_password: str

    @field_validator("new_password")
    @classmethod
    def strength(cls, v):
        if len(v) < 8:
            raise ValueError("min 8 characters")
        if not any(c.isupper() for c in v):
            raise ValueError("must contain uppercase")
        if not any(c.isdigit() for c in v):
            raise ValueError("must contain a digit")
        return v

@app.post("/me/password")
def change_password(
    body: PasswordChange,
    user: UserInDB = Depends(get_current_user),
):
    if not pwd_context.verify(body.current_password, user.hashed_password):
        raise HTTPException(401, "Current password is incorrect")
    # Update hash
    USERS_DB[user.username].hashed_password = pwd_context.hash(body.new_password)
    # Invalidate all refresh tokens for this user
    to_delete = [k for k, v in refresh_store.items() if v == str(user.id)]
    for k in to_delete:
        del refresh_store[k]
    return {"message": "Password updated. Please log in again."}

Auth: OAuth2 & JWT

🎯 Learning Objectives

Authentication vs Authorisation

JWT Structure

Header (base64url-encoded JSON)

Payload (base64url-encoded JSON) — standard claims

Signature

Decoding (NOT verifying) a token

Password Hashing with passlib

Issuing JWTs with PyJWT

The OAuth2 Password Flow

Protecting Routes with get_current_user

Refresh Tokens

Token Revocation & Blocklists

HTTPS, CORS & Security Headers

Complete Auth Module Layout

Best Practices

Exercises

Exercise 1 — Complete Login & Protected Routes

Exercise 2 — Token Expiry & Refresh Flow

Exercise 3 — Password Change with Token Invalidation

Protecting Routes with `get_current_user`