FastAPI Authentication with JWT + Refresh Tokens (Hands-On)

Pictalingo

·

FastAPI Authentication with JWT + Refresh Tokens (Hands-On)

If you’re building an API that will be used by a browser SPA or a mobile app, you usually need two things:

  • A short-lived access token (JWT) for calling protected endpoints.

  • A longer-lived refresh token to get a new access token without forcing the user to log in again.

This article walks through a practical FastAPI setup with:

  • Password hashing

  • JWT access tokens

  • Rotating refresh tokens (recommended)

  • Logout / revoke refresh tokens

Project setup

Install dependencies:

pip install fastapi uvicorn python-jose[cryptography] passlib[bcrypt] pydantic

Run the app:

uvicorn main:app --reload

We’ll keep everything in one file for clarity. In a real app, you’d split into routers, services, and models.

Core concepts (what we’re building)

  • Access token: JWT, short TTL (e.g., 15 minutes). Not stored server-side.

  • Refresh token: long TTL (e.g., 7–30 days). Stored server-side (or in DB) so you can revoke it.

  • Rotation: every refresh request invalidates the old refresh token and issues a new one. This reduces damage if a refresh token is stolen.

Working example: JWT + rotating refresh tokens

Create main.py:

from datetime import datetime, timedelta, timezone from typing import Optional, Dict import secrets import hashlib from fastapi import FastAPI, Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer from pydantic import BaseModel from passlib.context import CryptContext from jose import jwt, JWTError app = FastAPI() # ---- Security settings ---- JWT_ALG = "HS256" JWT_SECRET = "CHANGE_ME_TO_A_LONG_RANDOM_SECRET" # put in env var in real apps ACCESS_TTL_MIN = 15 REFRESH_TTL_DAYS = 14 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login") # ---- In-memory stores (replace with DB in real apps) ---- # Users: username -> dict USERS: Dict[str, dict] = { "alice": { "username": "alice", "password_hash": pwd_context.hash("password123"), "role": "user", } } # Refresh tokens store: refresh_id -> record # record: { username, expires_at, revoked, token_hash } REFRESH_STORE: Dict[str, dict] = {} # ---- Pydantic models ---- class LoginRequest(BaseModel): username: str password: str class TokenPair(BaseModel): access_token: str refresh_token: str token_type: str = "bearer" class RefreshRequest(BaseModel): refresh_token: str class LogoutRequest(BaseModel): refresh_token: str # ---- Helper functions ---- def now_utc() -> datetime: return datetime.now(timezone.utc) def hash_refresh_token(token: str) -> str: # Never store raw refresh tokens. Store a hash, similar to passwords. return hashlib.sha256(token.encode("utf-8")).hexdigest() def verify_password(plain: str, password_hash: str) -> bool: return pwd_context.verify(plain, password_hash) def create_access_token(username: str, role: str) -> str: exp = now_utc() + timedelta(minutes=ACCESS_TTL_MIN) payload = { "sub": username, "role": role, "exp": exp, "iat": now_utc(), "type": "access", } return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALG) def create_refresh_token() -> str: # High-entropy random string. You could also make this a JWT, but opaque tokens are common. return secrets.token_urlsafe(48) def issue_refresh_token(username: str) -> str: raw = create_refresh_token() refresh_id = secrets.token_urlsafe(16) # server-side key REFRESH_STORE[refresh_id] = { "username": username, "expires_at": now_utc() + timedelta(days=REFRESH_TTL_DAYS), "revoked": False, "token_hash": hash_refresh_token(raw), } # We return refresh_id + raw so we can lookup + verify hash. return f"{refresh_id}.{raw}" def parse_refresh_token(refresh_token: str) -> tuple[str, str]: try: refresh_id, raw = refresh_token.split(".", 1) return refresh_id, raw except ValueError: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid refresh token format", ) def validate_refresh_token(refresh_token: str) -> str: refresh_id, raw = parse_refresh_token(refresh_token) rec = REFRESH_STORE.get(refresh_id) if not rec: raise HTTPException(status_code=401, detail="Refresh token not found") if rec["revoked"]: raise HTTPException(status_code=401, detail="Refresh token revoked") if rec["expires_at"] < now_utc(): raise HTTPException(status_code=401, detail="Refresh token expired") if hash_refresh_token(raw) != rec["token_hash"]: # token mismatch (someone forged raw part or copied wrong token) raise HTTPException(status_code=401, detail="Refresh token invalid") return rec["username"] def revoke_refresh_token(refresh_token: str) -> None: refresh_id, _ = parse_refresh_token(refresh_token) rec = REFRESH_STORE.get(refresh_id) if rec: rec["revoked"] = True def rotate_refresh_token(old_refresh_token: str) -> str: username = validate_refresh_token(old_refresh_token) # revoke old token and issue a new one revoke_refresh_token(old_refresh_token) return issue_refresh_token(username) def get_current_user(token: str = Depends(oauth2_scheme)) -> dict: try: payload = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALG]) if payload.get("type") != "access": raise HTTPException(status_code=401, detail="Wrong token type") username = payload.get("sub") if not username: raise HTTPException(status_code=401, detail="Missing subject") except JWTError: raise HTTPException(status_code=401, detail="Invalid access token") user = USERS.get(username) if not user: raise HTTPException(status_code=401, detail="User not found") return user # ---- Routes ---- @app.post("/auth/login", response_model=TokenPair) def login(body: LoginRequest): user = USERS.get(body.username) if not user or not verify_password(body.password, user["password_hash"]): raise HTTPException(status_code=401, detail="Invalid credentials") access = create_access_token(user["username"], user["role"]) refresh = issue_refresh_token(user["username"]) return TokenPair(access_token=access, refresh_token=refresh) @app.post("/auth/refresh", response_model=TokenPair) def refresh(body: RefreshRequest): # Validate old refresh token and rotate username = validate_refresh_token(body.refresh_token) user = USERS.get(username) if not user: raise HTTPException(status_code=401, detail="User not found") new_refresh = rotate_refresh_token(body.refresh_token) new_access = create_access_token(user["username"], user["role"]) return TokenPair(access_token=new_access, refresh_token=new_refresh) @app.post("/auth/logout") def logout(body: LogoutRequest): revoke_refresh_token(body.refresh_token) return {"ok": True} @app.get("/me") def me(user: dict = Depends(get_current_user)): return {"username": user["username"], "role": user["role"]} @app.get("/admin/metrics") def admin_metrics(user: dict = Depends(get_current_user)): if user.get("role") != "admin": raise HTTPException(status_code=403, detail="Admins only") return {"requests_today": 123, "uptime": "48h"}

Try it with curl

Login to get tokens:

curl -s -X POST http://127.0.0.1:8000/auth/login \ -H "Content-Type: application/json" \ -d '{"username":"alice","password":"password123"}'

Copy the access_token and call a protected endpoint:

curl -s http://127.0.0.1:8000/me \ -H "Authorization: Bearer <ACCESS_TOKEN>"

When the access token expires, refresh it:

curl -s -X POST http://127.0.0.1:8000/auth/refresh \ -H "Content-Type: application/json" \ -d '{"refresh_token":"<REFRESH_TOKEN>"}'

Notice that refresh returns a new refresh token. The old one is revoked (rotation). If you try the old refresh token again, you should get 401.

Logout (revokes refresh token):

curl -s -X POST http://127.0.0.1:8000/auth/logout \ -H "Content-Type: application/json" \ -d '{"refresh_token":"<REFRESH_TOKEN>"}'

Practical storage guidance (so this works in real apps)

The demo uses in-memory dicts. In production, use a database table like refresh_tokens:

  • id (refresh_id), primary key

  • user_id

  • token_hash

  • expires_at

  • revoked boolean

  • created_at, rotated_at, maybe revoked_at

  • Optional: user_agent, ip_prefix, device_name for session management UX

On refresh:

  • Check token exists, not revoked, not expired

  • Compare hashes

  • Revoke old, insert new (ideally in a transaction)

Browser/mobile token handling (common pitfalls)

  • Don’t store access tokens in localStorage if you can avoid it for SPAs. A common approach is:

    • Access token kept in memory (lost on refresh, that’s fine)

    • Refresh token in an HttpOnly, Secure cookie

    • Use /auth/refresh on app start or 401 responses

  • Use short access TTLs (10–20 minutes is typical) so token leakage is less damaging.

  • Rotate refresh tokens (like we did) and store refresh tokens server-side so you can revoke sessions.

  • Scope your JWT claims: keep them small—sub, role, maybe aud. Don’t put sensitive data in JWT payloads.

Hardening checklist

  • Move secrets to env vars: JWT_SECRET should never be hardcoded.

  • Add rate limiting to /auth/login and /auth/refresh (e.g., per IP + per username).

  • Use HTTPS everywhere. JWT + refresh tokens over HTTP is asking for trouble.

  • Consider token “family” reuse detection: if a revoked refresh token is used again, you can revoke all tokens for that user/session (signals theft).

  • Set CORS correctly if you’re using cookies for refresh tokens.

  • Test expiry paths: unit test time-based behavior (expired refresh, expired access, revoked refresh).

Where to take this next

Once you’re comfortable with the flow, the next upgrades that junior/mid developers often ship are:

  • Multiple sessions per user (“devices”), with a /sessions endpoint to list and revoke them.

  • Database-backed users + refresh tokens via SQLAlchemy or SQLModel.

  • Role/permission middleware with route-level dependencies.

  • Structured logging around auth events (login success/fail, refresh, revoke).

You now have a practical JWT auth setup that’s realistic enough to use in production once you swap the in-memory stores for a database and apply the hardening checklist.

Leave a Reply

Your email address will not be published. Required fields are marked *