← Security & Authentication

Select a question from the sidebar

18 questions in this topic

Question 1 / 18

How does OAuth2PasswordBearer work in FastAPI?

OAuth2PasswordBearer is a FastAPI security scheme class that declares a dependency expecting a Bearer token in the Authorization header.

How it works:

You create an instance with tokenUrl pointing to your login endpoint
It tells FastAPI's auto-docs (Swagger UI) where to send username/password to get a token
When used as a dependency, it extracts the token from the Authorization: Bearer <token> header
If no token is present, it returns a 401 Unauthorized automatically
It does not validate the token itself -- that is your responsibility

What `tokenUrl` does:

It is only used by the Swagger UI "Authorize" button to know which endpoint to POST credentials to
It does not create the endpoint -- you must define it yourself
It should be a relative URL path (e.g., "token" or "/auth/token")

Integration with OpenAPI:

FastAPI uses this to generate the correct OpenAPI security scheme, enabling the lock icon and "Authorize" dialog in Swagger UI.

Pseudo Code

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# When used as dependency:
# 1. Reads Authorization header
# 2. Strips "Bearer " prefix
# 3. Returns raw token string
# 4. Raises 401 if missing

@app.get("/protected")
def protected(token: str = Depends(oauth2_scheme)):
    user = validate_and_decode(token)
    return user

Code

from fastapi import FastAPI, Depends
from fastapi.security import OAuth2PasswordBearer
from fastapi.testclient import TestClient

app = FastAPI()

# Declare the OAuth2 scheme -- tokenUrl is for Swagger UI
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.get("/items")
def read_items(token: str = Depends(oauth2_scheme)):
    """Token is the raw Bearer token string."""
    return {"token": token}

# --- Test it ---
client = TestClient(app)

# Without token -> 401
response = client.get("/items")
print(response.status_code)
print(response.json())

# With token -> 200
response = client.get("/items", headers={"Authorization": "Bearer my-secret-token"})
print(response.status_code)
print(response.json())

401
{'detail': 'Not authenticated'}
200
{'token': 'my-secret-token'}

Key Points

•OAuth2PasswordBearer extracts the Bearer token string from the Authorization header
•tokenUrl tells Swagger UI where to send credentials -- it does not create the endpoint
•Returns 401 Unauthorized automatically if no token is provided
•The token is returned as a raw string -- you must validate/decode it yourself

Gotchas

•tokenUrl is relative to the app root, not an absolute URL -- use 'token' not 'http://localhost/token'
•OAuth2PasswordBearer does NOT validate the token -- it only extracts it from the header
•If you use auto_error=False, it returns None instead of raising 401 when the token is missing

OAuth2PasswordBearer(tokenUrl="token") creates a dependency that extracts the Bearer token from the Authorization header. It returns the raw token string and raises 401 if missing. tokenUrl is only used by Swagger UI to know where to POST credentials.

•Extracts Bearer token from Authorization header, returns raw string
•Raises 401 automatically if token is missing

from fastapi import FastAPI, Depends
from fastapi.security import OAuth2PasswordBearer

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

@app.get("/protected")
def protected(token: str = Depends(oauth2_scheme)):
    return {"token": token}  # raw Bearer token string

It does NOT validate the token -- only extracts it

Question 2 / 18

How do you implement JWT token creation and validation?

JWT (JSON Web Tokens) are the standard way to implement stateless authentication in FastAPI. You encode user data into a signed token, and validate it on each request.

Token creation:

Build a payload with user data and expiration time (exp claim)
Sign it with a secret key using an algorithm like HS256
Return the encoded token string

Token validation:

Decode the token using the same secret and algorithm
Check if the token has expired (the exp claim is checked automatically)
Extract user info from the payload
Raise 401 if decoding fails for any reason

Libraries:

python-jose: pip install python-jose[cryptography] -- most commonly used with FastAPI
PyJWT: pip install PyJWT -- simpler alternative

Security considerations:

Use a strong, random SECRET_KEY (at least 32 bytes)
Keep expiration times short (15-30 minutes for access tokens)
Never store sensitive data in the payload (it is base64-encoded, not encrypted)
Use HTTPS in production to prevent token interception

Pseudo Code

# Creation:
payload = {"sub": user_id, "exp": now + timedelta(minutes=30)}
token = jwt.encode(payload, SECRET_KEY, algorithm="HS256")

# Validation:
try:
    payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
    user_id = payload["sub"]
except JWTError:
    raise 401 Unauthorized

Code

from datetime import datetime, timedelta, timezone
from jose import JWTError, jwt

SECRET_KEY = "your-secret-key-keep-it-secret"  # Use a real secret in production
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
    """Create a JWT access token."""
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + (
        expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def verify_token(token: str) -> dict:
    """Decode and validate a JWT token."""
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError as e:
        raise ValueError(f"Invalid token: {e}")

# --- Usage ---
token = create_access_token({"sub": "alice", "role": "admin"})
print(f"Token: {token[:50]}...")

payload = verify_token(token)
print(f"User: {payload['sub']}")
print(f"Role: {payload['role']}")

# Expired token example
expired_token = create_access_token(
    {"sub": "bob"}, expires_delta=timedelta(seconds=-1)
)
try:
    verify_token(expired_token)
except ValueError as e:
    print(f"Error: {e}")

Token: eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJz...
User: alice
Role: admin
Error: Invalid token: Signature has expired.

Key Points

•Use python-jose or PyJWT to encode/decode JWT tokens
•Always include an 'exp' claim for token expiration
•The 'sub' (subject) claim typically holds the username or user ID
•jwt.decode() automatically rejects expired tokens

Gotchas

•algorithms parameter in jwt.decode() must be a list, not a string
•JWT payloads are base64-encoded, NOT encrypted -- never store secrets in them
•Use timezone-aware datetimes (timezone.utc) to avoid subtle expiration bugs
•python-jose and PyJWT have slightly different APIs -- do not mix them

Use python-jose to create JWTs with jwt.encode({"sub": user, "exp": expiry}, SECRET, algorithm="HS256") and validate with jwt.decode(token, SECRET, algorithms=["HS256"]). The exp claim auto-rejects expired tokens.

•jwt.encode() creates tokens, jwt.decode() validates and extracts data
•The 'exp' claim handles expiration automatically

from datetime import datetime, timedelta, timezone
from jose import jwt, JWTError

SECRET = "my-secret"

def create_token(username: str) -> str:
    expire = datetime.now(timezone.utc) + timedelta(minutes=30)
    return jwt.encode({"sub": username, "exp": expire}, SECRET, algorithm="HS256")

def decode_token(token: str) -> str:
    payload = jwt.decode(token, SECRET, algorithms=["HS256"])
    return payload["sub"]

JWT payloads are NOT encrypted -- anyone can read them with base64 decoding

Question 3 / 18

How do you hash and verify passwords with passlib/bcrypt?

Never store plaintext passwords. Use passlib with the bcrypt scheme to hash passwords before storing them and verify them during login.

How bcrypt works:

Hashing: Takes a password and produces a fixed-length hash with a random salt baked in
Verification: Hashes the input password with the same salt and compares results
Cost factor: Controls how slow hashing is (higher = more secure but slower)

Setup:

pip install passlib[bcrypt]

passlib's CryptContext:

Manages hashing schemes and migration between them
schemes=["bcrypt"] specifies the hashing algorithm
deprecated="auto" auto-migrates old hashes when users log in
Handles salt generation automatically

Why bcrypt:

Intentionally slow (prevents brute-force attacks)
Built-in salt (no salt management needed)
Configurable work factor (adapts to faster hardware)
Industry standard for password hashing

Pseudo Code

pwd_context = CryptContext(schemes=["bcrypt"])

# Storing a password:
hashed = pwd_context.hash(plain_password)
store_in_db(user, hashed)

# Verifying at login:
if pwd_context.verify(input_password, stored_hash):
    grant_access()
else:
    deny_access()

Code

from passlib.context import CryptContext

# Create a password context with bcrypt
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def hash_password(password: str) -> str:
    """Hash a plaintext password."""
    return pwd_context.hash(password)

def verify_password(plain_password: str, hashed_password: str) -> bool:
    """Verify a password against its hash."""
    return pwd_context.verify(plain_password, hashed_password)

# --- Usage ---
password = "my-secret-password"
hashed = hash_password(password)

print(f"Hash: {hashed[:30]}...")
print(f"Hash length: {len(hashed)}")

# Correct password
print(f"Correct: {verify_password('my-secret-password', hashed)}")

# Wrong password
print(f"Wrong: {verify_password('wrong-password', hashed)}")

# Same password produces different hashes (different salt each time)
hashed2 = hash_password(password)
print(f"Same hash? {hashed == hashed2}")
print(f"Both verify? {verify_password(password, hashed) and verify_password(password, hashed2)}")

Hash: $2b$12$...
Hash length: 60
Correct: True
Wrong: False
Same hash? False
Both verify? True

Key Points

•CryptContext(schemes=['bcrypt']) creates a reusable password hasher
•pwd_context.hash() generates a salted bcrypt hash -- different each time
•pwd_context.verify() compares a plaintext password against a stored hash
•deprecated='auto' enables transparent migration to newer hash schemes

Gotchas

•Never compare hashes directly -- use verify() which handles salt extraction
•bcrypt has a 72-byte password limit -- longer passwords are silently truncated
•Install passlib with bcrypt extra: pip install passlib[bcrypt], not just passlib
•Hash operations are intentionally slow (~200ms) -- do not run in tight loops

Use passlib.context.CryptContext(schemes=["bcrypt"]) to hash with .hash(password) and verify with .verify(plain, hashed). Bcrypt automatically handles salting and is intentionally slow to resist brute-force attacks.

•hash() creates a salted hash, verify() checks against it

from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

hashed = pwd_context.hash("my-password")
print(pwd_context.verify("my-password", hashed))   # True
print(pwd_context.verify("wrong", hashed))          # False

True
False

bcrypt silently truncates passwords longer than 72 bytes

Question 4 / 18

How do you protect routes with authentication dependencies?

FastAPI's dependency injection system is the primary way to protect routes. You create a dependency that validates the token and returns the current user, then inject it into any endpoint that requires authentication.

Approaches:

Per-route dependency: Add Depends(get_current_user) to individual endpoints
Router-level dependency: Apply to all routes in an APIRouter
App-level dependency: Apply to every route in the application

How it works:

Request arrives with Authorization: Bearer <token> header
OAuth2PasswordBearer extracts the token
Your get_current_user dependency decodes and validates the token
If valid, the user object is injected into the endpoint
If invalid, a 401 is raised before the endpoint runs

Using Annotated (modern style):

Pydantic v2 and modern FastAPI prefer Annotated[User, Depends(get_current_user)] over the default parameter style for cleaner, reusable type hints.

Pseudo Code

# Define the dependency chain:
oauth2_scheme -> extracts token string
get_current_user(token) -> decodes JWT, returns User

# Protect a route:
@app.get("/me")
def me(user: Annotated[User, Depends(get_current_user)]):
    return user

# Protect all routes in a router:
router = APIRouter(dependencies=[Depends(get_current_user)])

Code

from typing import Annotated
from fastapi import FastAPI, Depends, HTTPException, status, APIRouter
from fastapi.security import OAuth2PasswordBearer
from fastapi.testclient import TestClient
from pydantic import BaseModel
from jose import JWTError, jwt
from datetime import datetime, timedelta, timezone

app = FastAPI()

SECRET_KEY = "test-secret"
ALGORITHM = "HS256"
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

class User(BaseModel):
    username: str
    role: str = "user"

def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]) -> User:
    """Dependency that validates JWT and returns current user."""
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        return User(username=username, role=payload.get("role", "user"))
    except JWTError:
        raise credentials_exception

# Reusable type alias (modern Annotated style)
CurrentUser = Annotated[User, Depends(get_current_user)]

# --- Per-route protection ---
@app.get("/me")
def read_me(current_user: CurrentUser):
    return {"username": current_user.username, "role": current_user.role}

# --- Router-level protection (all routes require auth) ---
admin_router = APIRouter(
    prefix="/admin",
    dependencies=[Depends(get_current_user)],
)

@admin_router.get("/dashboard")
def admin_dashboard():
    return {"message": "Admin area"}

app.include_router(admin_router)

# --- Test it ---
def make_token(sub: str, role: str = "user") -> str:
    exp = datetime.now(timezone.utc) + timedelta(hours=1)
    return jwt.encode({"sub": sub, "role": role, "exp": exp}, SECRET_KEY, algorithm=ALGORITHM)

client = TestClient(app)

# No token
resp = client.get("/me")
print(resp.status_code)

# Valid token
token = make_token("alice", "admin")
resp = client.get("/me", headers={"Authorization": f"Bearer {token}"})
print(resp.json())

# Router-level protection
resp = client.get("/admin/dashboard", headers={"Authorization": f"Bearer {token}"})
print(resp.json())

401
{'username': 'alice', 'role': 'admin'}
{'message': 'Admin area'}

Key Points

•Create a get_current_user dependency that decodes JWT and returns a User
•Use Annotated[User, Depends(get_current_user)] for reusable type aliases
•Router-level dependencies protect all routes in that router
•Always include WWW-Authenticate header in 401 responses per HTTP spec

Gotchas

•Router-level dependencies run but their return values are not injected into endpoints -- use per-route Depends if you need the user object
•HTTPException in a dependency stops the entire request chain
•Annotated style requires Python 3.9+ (or typing_extensions for 3.8)

Create a get_current_user dependency that uses OAuth2PasswordBearer to extract the token, decodes the JWT, and returns a user object. Inject it with Depends() on any route that needs protection. Use Annotated[User, Depends(get_current_user)] for clean reuse.

•get_current_user dependency decodes JWT and returns user data
•Annotated[type, Depends()] is the modern, reusable style

from typing import Annotated
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import jwt, JWTError

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

def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
    try:
        payload = jwt.decode(token, "secret", algorithms=["HS256"])
        return {"username": payload["sub"]}
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

@app.get("/me")
def me(user: Annotated[dict, Depends(get_current_user)]):
    return user

Router-level dependencies run but do NOT inject their return value into endpoints

Question 5 / 18

How do you implement role-based access control (RBAC)?

Role-Based Access Control (RBAC) restricts endpoint access based on the user's role (e.g., admin, editor, viewer). In FastAPI, this is implemented by chaining dependencies.

Approach:

Store the user's role in the JWT token (or fetch from DB)
Create a dependency factory function that accepts allowed roles
The factory returns a dependency that checks if the current user has the required role
If not, raise a 403 Forbidden

Dependency factory pattern:

A function that returns a function -- the outer function configures which roles are allowed, the inner function does the actual check at request time.

RBAC vs ABAC:

RBAC: Access based on roles (admin, user, editor)
ABAC: Access based on attributes (owns resource, department, time of day)
FastAPI supports both through flexible dependency injection

Pseudo Code

def require_role(*roles):
    def checker(user = Depends(get_current_user)):
        if user.role not in roles:
            raise 403 Forbidden
        return user
    return checker

@app.get("/admin", dependencies=[Depends(require_role("admin"))])
def admin_only(): ...

Code

from typing import Annotated
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from fastapi.testclient import TestClient
from pydantic import BaseModel
from jose import jwt
from datetime import datetime, timedelta, timezone

app = FastAPI()

SECRET = "test-secret"
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

class User(BaseModel):
    username: str
    role: str

def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]) -> User:
    payload = jwt.decode(token, SECRET, algorithms=["HS256"])
    return User(username=payload["sub"], role=payload.get("role", "viewer"))

CurrentUser = Annotated[User, Depends(get_current_user)]

# --- Dependency factory for role checking ---
def require_role(*allowed_roles: str):
    """Returns a dependency that enforces role-based access."""
    def role_checker(current_user: CurrentUser) -> User:
        if current_user.role not in allowed_roles:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"Role '{current_user.role}' not authorized. "
                       f"Required: {', '.join(allowed_roles)}",
            )
        return current_user
    return role_checker

# Reusable role dependencies
AdminUser = Annotated[User, Depends(require_role("admin"))]
EditorUser = Annotated[User, Depends(require_role("admin", "editor"))]

@app.get("/admin/settings")
def admin_settings(user: AdminUser):
    return {"message": f"Welcome admin {user.username}"}

@app.get("/posts/create")
def create_post(user: EditorUser):
    return {"message": f"{user.username} ({user.role}) can create posts"}

@app.get("/profile")
def profile(user: CurrentUser):
    return {"message": f"Hello {user.username}", "role": user.role}

# --- Test ---
def make_token(username: str, role: str) -> str:
    exp = datetime.now(timezone.utc) + timedelta(hours=1)
    return jwt.encode({"sub": username, "role": role, "exp": exp}, SECRET, algorithm="HS256")

client = TestClient(app)

admin_token = make_token("alice", "admin")
editor_token = make_token("bob", "editor")
viewer_token = make_token("charlie", "viewer")

# Admin can access admin settings
resp = client.get("/admin/settings", headers={"Authorization": f"Bearer {admin_token}"})
print(resp.json())

# Editor cannot access admin settings
resp = client.get("/admin/settings", headers={"Authorization": f"Bearer {editor_token}"})
print(resp.status_code, resp.json()["detail"])

# Editor can create posts
resp = client.get("/posts/create", headers={"Authorization": f"Bearer {editor_token}"})
print(resp.json())

{'message': 'Welcome admin alice'}
403 Role 'editor' not authorized. Required: admin
{'message': 'bob (editor) can create posts'}

Key Points

•Use a dependency factory (function returning a function) to parameterize role checks
•Chain dependencies: require_role depends on get_current_user automatically
•Return 403 Forbidden for authenticated users lacking the required role (not 401)
•Create reusable Annotated type aliases like AdminUser for clean endpoint signatures

Gotchas

•Use 403 (Forbidden) not 401 (Unauthorized) for role failures -- the user IS authenticated, just not authorized
•Role stored in JWT cannot be revoked until the token expires -- consider short-lived tokens
•Dependency factories must return the inner function, not call it

Create a dependency factory require_role(*roles) that returns a dependency checking if current_user.role is in the allowed roles. Raise 403 if not. Use Annotated aliases like AdminUser = Annotated[User, Depends(require_role("admin"))] for clean reuse.

•Dependency factory returns a checker function parameterized with allowed roles

from typing import Annotated
from fastapi import Depends, HTTPException, status

def require_role(*allowed_roles):
    def checker(user: CurrentUser):  # CurrentUser = Annotated[User, Depends(get_current_user)]
        if user.role not in allowed_roles:
            raise HTTPException(status_code=403, detail="Not authorized")
        return user
    return checker

AdminUser = Annotated[User, Depends(require_role("admin"))]

Use 403 (Forbidden) for role failures, not 401 (Unauthorized)

Question 6 / 18

How do you implement API key authentication?

API key authentication is simpler than OAuth2/JWT and commonly used for service-to-service communication or public APIs with rate limiting.

FastAPI provides three API key security classes:

APIKeyHeader: Key sent in a custom header (e.g., X-API-Key)
APIKeyQuery: Key sent as a query parameter
APIKeyCookie: Key sent in a cookie

How it works:

Define the security scheme (where to look for the key)
Create a dependency that extracts and validates the key
Inject the dependency into protected routes

Best practices:

Use header-based keys (not query params) to avoid keys leaking in URL logs
Store valid API keys hashed in the database
Implement rate limiting per API key
Use secrets.compare_digest() for timing-safe comparison to prevent timing attacks

Pseudo Code

api_key_header = APIKeyHeader(name="X-API-Key")

def validate_api_key(key = Depends(api_key_header)):
    if key not in VALID_KEYS:
        raise 403
    return key

@app.get("/data", dependencies=[Depends(validate_api_key)])
def get_data(): ...

Code

import secrets
from typing import Annotated
from fastapi import FastAPI, Depends, HTTPException, Security, status
from fastapi.security import APIKeyHeader, APIKeyQuery
from fastapi.testclient import TestClient

app = FastAPI()

# Valid API keys (in production, store hashed in a database)
API_KEYS = {
    "sk-live-abc123": {"name": "Service A", "tier": "premium"},
    "sk-live-def456": {"name": "Service B", "tier": "basic"},
}

# Look for key in X-API-Key header
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
# Fallback: look in query parameter
api_key_query = APIKeyQuery(name="api_key", auto_error=False)

def get_api_key(
    header_key: Annotated[str | None, Security(api_key_header)],
    query_key: Annotated[str | None, Security(api_key_query)],
) -> str:
    """Accept API key from header or query param."""
    key = header_key or query_key
    if not key:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="API key required",
        )
    # Timing-safe comparison against all valid keys
    for valid_key in API_KEYS:
        if secrets.compare_digest(key, valid_key):
            return key
    raise HTTPException(
        status_code=status.HTTP_403_FORBIDDEN,
        detail="Invalid API key",
    )

ApiKey = Annotated[str, Depends(get_api_key)]

@app.get("/data")
def get_data(api_key: ApiKey):
    client_info = API_KEYS[api_key]
    return {"client": client_info["name"], "tier": client_info["tier"]}

# --- Test ---
client = TestClient(app)

# No key
print(client.get("/data").status_code)

# Header key
resp = client.get("/data", headers={"X-API-Key": "sk-live-abc123"})
print(resp.json())

# Query param key
resp = client.get("/data?api_key=sk-live-def456")
print(resp.json())

# Invalid key
print(client.get("/data", headers={"X-API-Key": "bad-key"}).status_code)

401
{'client': 'Service A', 'tier': 'premium'}
{'client': 'Service B', 'tier': 'basic'}
403

Key Points

•APIKeyHeader, APIKeyQuery, APIKeyCookie extract keys from different locations
•Use auto_error=False to check multiple sources with fallback logic
•Use secrets.compare_digest() for timing-safe string comparison
•Security() works like Depends() but adds the scheme to OpenAPI docs

Gotchas

•Query parameter API keys leak in server logs and browser history -- prefer headers
•Without auto_error=False, a missing key raises 403 immediately, blocking fallback checks
•Simple == comparison is vulnerable to timing attacks -- always use secrets.compare_digest()

Use APIKeyHeader(name="X-API-Key") to extract a key from a custom header, then validate it in a dependency. Use secrets.compare_digest() for safe comparison. Security() works like Depends() but adds the scheme to OpenAPI docs.

•APIKeyHeader extracts from a named header, validates in a dependency

from fastapi import FastAPI, Depends, HTTPException, Security
from fastapi.security import APIKeyHeader
import secrets

app = FastAPI()
api_key_header = APIKeyHeader(name="X-API-Key")

def validate_key(key: str = Security(api_key_header)):
    if not secrets.compare_digest(key, "sk-live-abc123"):
        raise HTTPException(status_code=403, detail="Invalid key")
    return key

@app.get("/data")
def data(key: str = Depends(validate_key)):
    return {"status": "authorized"}

Use secrets.compare_digest() not == to prevent timing attacks

Question 7 / 18

How does HTTPBearer work for Bearer token auth?

HTTPBearer is an alternative to OAuth2PasswordBearer that uses the standard HTTP Bearer authentication scheme. It returns an HTTPAuthorizationCredentials object instead of a raw string.

Key differences from OAuth2PasswordBearer:

Feature	OAuth2PasswordBearer	HTTPBearer
Returns	Raw token string	HTTPAuthorizationCredentials(scheme, credentials)
OpenAPI	OAuth2 flow with tokenUrl	Generic HTTP Bearer scheme
Swagger UI	Shows "Authorize" with username/password	Shows "Authorize" with token input only
Use case	Full OAuth2 password flow	Simple Bearer token auth

When to use HTTPBearer:

You do not have a login endpoint in the same API (external auth provider)
You want a simpler Swagger UI experience (just paste a token)
You are implementing non-OAuth2 Bearer authentication

HTTPAuthorizationCredentials:

scheme: The authentication scheme (always "Bearer")
credentials: The actual token string

Pseudo Code

security = HTTPBearer()

def get_user(creds: HTTPAuthorizationCredentials = Depends(security)):
    token = creds.credentials  # The actual token
    scheme = creds.scheme      # "Bearer"
    return decode(token)

@app.get("/me")
def me(user = Depends(get_user)): ...

Code

from typing import Annotated
from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi.testclient import TestClient

app = FastAPI()

# HTTPBearer security scheme
security = HTTPBearer()

@app.get("/profile")
def profile(
    credentials: Annotated[HTTPAuthorizationCredentials, Depends(security)],
):
    """credentials has .scheme and .credentials attributes."""
    return {
        "scheme": credentials.scheme,
        "token": credentials.credentials,
        "token_length": len(credentials.credentials),
    }

# --- Optional: HTTPBearer with auto_error=False for custom handling ---
optional_security = HTTPBearer(auto_error=False)

@app.get("/public-or-private")
def flexible(
    credentials: Annotated[
        HTTPAuthorizationCredentials | None,
        Depends(optional_security),
    ],
):
    if credentials:
        return {"authenticated": True, "token": credentials.credentials}
    return {"authenticated": False, "message": "Public access"}

# --- Test ---
client = TestClient(app)

# Valid Bearer token
resp = client.get("/profile", headers={"Authorization": "Bearer my-jwt-token"})
print(resp.json())

# No token -> 403
resp = client.get("/profile")
print(resp.status_code)

# Optional endpoint with token
resp = client.get("/public-or-private", headers={"Authorization": "Bearer abc"})
print(resp.json())

# Optional endpoint without token
resp = client.get("/public-or-private")
print(resp.json())

{'scheme': 'Bearer', 'token': 'my-jwt-token', 'token_length': 12}
403
{'authenticated': True, 'token': 'abc'}
{'authenticated': False, 'message': 'Public access'}

Key Points

•HTTPBearer returns HTTPAuthorizationCredentials with .scheme and .credentials
•Use it when you do not have a login endpoint in the same API
•auto_error=False makes authentication optional (returns None instead of 403)
•Swagger UI shows a simple token input instead of username/password fields

Gotchas

•HTTPBearer returns 403 by default when no token is provided, not 401 like OAuth2PasswordBearer
•You get an object (HTTPAuthorizationCredentials), not a raw string -- use .credentials to get the token
•HTTPBearer does not support tokenUrl -- use OAuth2PasswordBearer if you need the full OAuth2 password flow

HTTPBearer() is a simpler alternative to OAuth2PasswordBearer. It returns HTTPAuthorizationCredentials with .scheme ("Bearer") and .credentials (the token). Swagger UI shows a token-only input instead of username/password fields.

•Returns HTTPAuthorizationCredentials object, not a raw string

from typing import Annotated
from fastapi import FastAPI, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

app = FastAPI()
security = HTTPBearer()

@app.get("/me")
def me(creds: Annotated[HTTPAuthorizationCredentials, Depends(security)]):
    token = creds.credentials  # The actual token
    return {"token": token}

HTTPBearer raises 403 by default (not 401) when the token is missing

Question 8 / 18

What are OAuth2 security scopes and how do you use them?

OAuth2 scopes provide fine-grained permission control beyond simple role checks. Instead of "admin" or "user" roles, scopes define specific permissions like items:read, items:write, users:manage.

How FastAPI implements scopes:

Define available scopes when creating OAuth2PasswordBearer
Include granted scopes in the JWT token
Declare required scopes in Security() dependencies
FastAPI checks scopes automatically and returns 403 if insufficient

Key components:

SecurityScopes: Injected into your dependency, contains the list of required scopes
Security(dependency, scopes=[...]): Declares which scopes a route needs
The scopes are hierarchical -- you can require multiple scopes on one endpoint

When to use scopes vs roles:

Roles: Simple access levels (admin, editor, viewer)
Scopes: Granular permissions, often used with third-party OAuth2 integrations (like GitHub, Google)

Pseudo Code

oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl="token",
    scopes={"items:read": "Read items", "items:write": "Write items"}
)

def get_current_user(security_scopes: SecurityScopes, token = Depends(oauth2_scheme)):
    payload = decode(token)
    token_scopes = payload.get("scopes", [])
    for scope in security_scopes.scopes:
        if scope not in token_scopes:
            raise 403
    return user

@app.get("/items")
def read_items(user = Security(get_current_user, scopes=["items:read"])): ...

Code

from typing import Annotated
from fastapi import FastAPI, Depends, HTTPException, Security, status
from fastapi.security import OAuth2PasswordBearer, SecurityScopes
from fastapi.testclient import TestClient
from jose import jwt, JWTError
from pydantic import BaseModel
from datetime import datetime, timedelta, timezone

app = FastAPI()
SECRET = "test-secret"

# Define available scopes
oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl="token",
    scopes={
        "items:read": "Read items",
        "items:write": "Create and update items",
        "users:manage": "Manage user accounts",
    },
)

class User(BaseModel):
    username: str
    scopes: list[str] = []

def get_current_user(
    security_scopes: SecurityScopes,
    token: Annotated[str, Depends(oauth2_scheme)],
) -> User:
    """Validate token AND check required scopes."""
    # Build WWW-Authenticate header with required scopes
    authenticate_value = (
        f'Bearer scope="{security_scopes.scope_str}"'
        if security_scopes.scopes
        else "Bearer"
    )
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": authenticate_value},
    )
    try:
        payload = jwt.decode(token, SECRET, algorithms=["HS256"])
        username = payload.get("sub")
        token_scopes = payload.get("scopes", [])
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception

    user = User(username=username, scopes=token_scopes)

    # Check each required scope
    for scope in security_scopes.scopes:
        if scope not in user.scopes:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"Missing scope: {scope}",
                headers={"WWW-Authenticate": authenticate_value},
            )
    return user

# Routes with scope requirements
@app.get("/items")
def read_items(
    user: Annotated[User, Security(get_current_user, scopes=["items:read"])],
):
    return {"user": user.username, "action": "reading items"}

@app.post("/items")
def create_item(
    user: Annotated[User, Security(get_current_user, scopes=["items:write"])],
):
    return {"user": user.username, "action": "creating item"}

@app.get("/users")
def manage_users(
    user: Annotated[
        User, Security(get_current_user, scopes=["users:manage", "items:read"])
    ],
):
    return {"user": user.username, "action": "managing users"}

# --- Test ---
def make_token(username: str, scopes: list[str]) -> str:
    exp = datetime.now(timezone.utc) + timedelta(hours=1)
    return jwt.encode(
        {"sub": username, "scopes": scopes, "exp": exp}, SECRET, algorithm="HS256"
    )

client = TestClient(app)

# User with read-only scope
read_token = make_token("alice", ["items:read"])
resp = client.get("/items", headers={"Authorization": f"Bearer {read_token}"})
print(resp.json())

# Cannot write
resp = client.post("/items", headers={"Authorization": f"Bearer {read_token}"})
print(resp.status_code, resp.json()["detail"])

# Admin with all scopes
admin_token = make_token("admin", ["items:read", "items:write", "users:manage"])
resp = client.get("/users", headers={"Authorization": f"Bearer {admin_token}"})
print(resp.json())

{'user': 'alice', 'action': 'reading items'}
403 Missing scope: items:write
{'user': 'admin', 'action': 'managing users'}

Key Points

•Define scopes dict in OAuth2PasswordBearer to document available permissions
•SecurityScopes is injected into the dependency and contains required scope list
•Use Security(dep, scopes=['scope']) instead of Depends() to declare required scopes
•Scopes are cumulative -- requiring ['a', 'b'] means the user needs BOTH

Gotchas

•SecurityScopes must be the first parameter of the dependency function
•Scopes in JWT are just a list of strings -- there is no built-in hierarchy
•The scopes dict in OAuth2PasswordBearer is for documentation only -- enforcement is your responsibility

Define scopes in OAuth2PasswordBearer(scopes={...}), include granted scopes in the JWT, and use Security(get_current_user, scopes=["items:read"]) on endpoints. Your dependency receives SecurityScopes to check required vs granted permissions.

•Security() declares required scopes, SecurityScopes receives them in the dependency

from fastapi import Security
from fastapi.security import OAuth2PasswordBearer, SecurityScopes

oauth2 = OAuth2PasswordBearer(tokenUrl="token", scopes={"read": "Read", "write": "Write"})

def get_user(security_scopes: SecurityScopes, token: str = Depends(oauth2)):
    user_scopes = decode_token(token)["scopes"]
    for scope in security_scopes.scopes:
        if scope not in user_scopes:
            raise HTTPException(status_code=403, detail=f"Missing: {scope}")
    return user

@app.get("/items")
def read(user = Security(get_user, scopes=["read"])): ...

Scope enforcement is your responsibility -- FastAPI only provides the plumbing

Question 9 / 18

How do you implement token refresh functionality?

Token refresh uses two tokens to balance security and usability:

Two-token pattern:

Access token: Short-lived (15-30 minutes), used for API requests
Refresh token: Long-lived (days/weeks), used only to get new access tokens

Why two tokens:

Access tokens are sent with every request (higher exposure risk)
Short-lived access tokens limit the window of compromise
Refresh tokens are sent rarely (only to the refresh endpoint)
If an access token is stolen, it expires quickly
Refresh tokens can be revoked server-side (stored in DB)

Flow:

User logs in -> receives both access and refresh tokens
Client uses access token for API calls
When access token expires, client sends refresh token to /refresh
Server validates refresh token and issues new access token
Optionally issue a new refresh token too (token rotation)

Refresh token rotation:

Issuing a new refresh token on each refresh and invalidating the old one prevents replay attacks. If a stolen refresh token is used, the real user's next refresh fails, alerting them.

Pseudo Code

# Login: issue both tokens
access_token = create_token(user, expires=30min, type="access")
refresh_token = create_token(user, expires=7days, type="refresh")
store_refresh_token_in_db(refresh_token)
return {access_token, refresh_token}

# Refresh: validate and rotate
def refresh(token):
    payload = decode(token)
    assert payload["type"] == "refresh"
    assert token exists in DB
    revoke_old_refresh_token(token)
    return new access_token + new refresh_token

Code

from datetime import datetime, timedelta, timezone
from typing import Annotated
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from fastapi.testclient import TestClient
from pydantic import BaseModel
from jose import jwt, JWTError

app = FastAPI()

SECRET = "test-secret"
ALGORITHM = "HS256"
ACCESS_EXPIRE = timedelta(minutes=30)
REFRESH_EXPIRE = timedelta(days=7)

# In production, store valid refresh tokens in a database
refresh_token_store: set[str] = set()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

class TokenPair(BaseModel):
    access_token: str
    refresh_token: str
    token_type: str = "bearer"

def create_token(data: dict, expires_delta: timedelta, token_type: str) -> str:
    payload = data.copy()
    payload.update({
        "exp": datetime.now(timezone.utc) + expires_delta,
        "type": token_type,
    })
    return jwt.encode(payload, SECRET, algorithm=ALGORITHM)

def create_token_pair(username: str) -> TokenPair:
    access = create_token({"sub": username}, ACCESS_EXPIRE, "access")
    refresh = create_token({"sub": username}, REFRESH_EXPIRE, "refresh")
    refresh_token_store.add(refresh)  # Store valid refresh token
    return TokenPair(access_token=access, refresh_token=refresh)

@app.post("/token")
def login():
    """Simplified login -- returns both tokens."""
    return create_token_pair("alice")

@app.post("/refresh")
def refresh_token(token: Annotated[str, Depends(oauth2_scheme)]):
    """Exchange a refresh token for a new token pair."""
    try:
        payload = jwt.decode(token, SECRET, algorithms=[ALGORITHM])
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid refresh token")

    # Verify it is a refresh token
    if payload.get("type") != "refresh":
        raise HTTPException(status_code=401, detail="Not a refresh token")

    # Verify it has not been revoked
    if token not in refresh_token_store:
        raise HTTPException(status_code=401, detail="Refresh token revoked")

    # Rotate: revoke old, issue new pair
    refresh_token_store.discard(token)
    username = payload["sub"]
    return create_token_pair(username)

@app.post("/logout")
def logout(token: Annotated[str, Depends(oauth2_scheme)]):
    """Revoke the refresh token."""
    refresh_token_store.discard(token)
    return {"message": "Logged out"}

# --- Test ---
client = TestClient(app)

# Login
resp = client.post("/token")
tokens = resp.json()
print(f"Got access + refresh tokens: {bool(tokens['access_token'])}")

# Refresh
resp = client.post("/refresh", headers={"Authorization": f"Bearer {tokens['refresh_token']}"})
new_tokens = resp.json()
print(f"Refreshed: {bool(new_tokens['access_token'])}")

# Old refresh token is now invalid (rotation)
resp = client.post("/refresh", headers={"Authorization": f"Bearer {tokens['refresh_token']}"})
print(f"Old token rejected: {resp.status_code}")

Got access + refresh tokens: True
Refreshed: True
Old token rejected: 401

Key Points

•Access tokens are short-lived (minutes), refresh tokens are long-lived (days)
•The 'type' claim distinguishes access tokens from refresh tokens
•Refresh token rotation: issue a new pair and revoke the old refresh token
•Store valid refresh tokens server-side so they can be revoked

Gotchas

•Without server-side storage, refresh tokens cannot be revoked before expiration
•Always check the 'type' claim -- do not accept a refresh token as an access token
•Token rotation requires atomic revoke-and-issue to prevent race conditions
•Refresh endpoint should be rate-limited to prevent abuse

Issue short-lived access tokens and long-lived refresh tokens. The refresh endpoint validates the refresh token, revokes it, and returns a new token pair. Store valid refresh tokens server-side for revocation.

•Two tokens: short-lived access, long-lived refresh
•Always verify the 'type' claim to prevent token misuse

from datetime import datetime, timedelta
def create_token_pair(username: str):
    access = jwt.encode(
        {"sub": username, "type": "access", "exp": now + timedelta(minutes=30)},
        SECRET, algorithm="HS256"
    )
    refresh = jwt.encode(
        {"sub": username, "type": "refresh", "exp": now + timedelta(days=7)},
        SECRET, algorithm="HS256"
    )
    return {"access_token": access, "refresh_token": refresh}

@app.post("/refresh")
def refresh(token: str = Depends(oauth2_scheme)):
    payload = jwt.decode(token, SECRET, algorithms=["HS256"])
    if payload["type"] != "refresh":
        raise HTTPException(status_code=401)
    return create_token_pair(payload["sub"])

Without server-side storage, refresh tokens cannot be revoked early

Question 10 / 18

How do you implement the complete login flow with OAuth2PasswordRequestForm?

OAuth2PasswordRequestForm is a FastAPI dependency class that parses the standard OAuth2 password grant form data. The token endpoint receives username and password as form fields (not JSON).

Why form data instead of JSON:

The OAuth2 spec requires application/x-www-form-urlencoded content type
This is what Swagger UI's "Authorize" dialog sends
Libraries and tools expect this format for the token endpoint

Complete flow:

Client sends POST /token with form data: username + password
Server looks up user in the database
Server verifies password hash
Server creates and returns JWT token
Client sends token in Authorization: Bearer <token> header on subsequent requests

OAuth2PasswordRequestForm fields:

username: Required
password: Required
scope: Optional space-separated scopes string
grant_type: Optional, should be "password"
client_id: Optional
client_secret: Optional

Pseudo Code

@app.post("/token")
def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate(form_data.username, form_data.password)
    if not user:
        raise 401
    token = create_jwt(user)
    return {"access_token": token, "token_type": "bearer"}

Code

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

app = FastAPI()

SECRET = "test-secret"
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# Simulated user database
fake_users_db = {
    "alice": {
        "username": "alice",
        "hashed_password": pwd_context.hash("secret123"),
        "disabled": False,
    },
}

class Token(BaseModel):
    access_token: str
    token_type: str

def authenticate_user(username: str, password: str) -> dict | None:
    """Look up user and verify password."""
    user = fake_users_db.get(username)
    if not user:
        return None
    if not pwd_context.verify(password, user["hashed_password"]):
        return None
    return user

@app.post("/token", response_model=Token)
def login(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
):
    """OAuth2 token endpoint -- receives form data, not JSON."""
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    if user.get("disabled"):
        raise HTTPException(status_code=400, detail="Inactive user")

    access_token = jwt.encode(
        {
            "sub": user["username"],
            "exp": datetime.now(timezone.utc) + timedelta(minutes=30),
        },
        SECRET,
        algorithm="HS256",
    )
    return Token(access_token=access_token, token_type="bearer")

@app.get("/me")
def read_me(token: Annotated[str, Depends(oauth2_scheme)]):
    payload = jwt.decode(token, SECRET, algorithms=["HS256"])
    return {"username": payload["sub"]}

# --- Test ---
client = TestClient(app)

# Login with form data (not JSON!)
resp = client.post("/token", data={"username": "alice", "password": "secret123"})
print(f"Login status: {resp.status_code}")
token = resp.json()["access_token"]

# Use the token
resp = client.get("/me", headers={"Authorization": f"Bearer {token}"})
print(resp.json())

# Wrong password
resp = client.post("/token", data={"username": "alice", "password": "wrong"})
print(f"Bad login: {resp.status_code}")

Login status: 200
{'username': 'alice'}
Bad login: 401

Key Points

•OAuth2PasswordRequestForm receives form data (not JSON) per OAuth2 spec
•The login endpoint must return {'access_token': ..., 'token_type': 'bearer'}
•Use Annotated[OAuth2PasswordRequestForm, Depends()] for modern style
•Always verify both that the user exists AND the password matches

Gotchas

•Send form data (data={...}) in tests, not JSON (json={...}) -- this is the most common mistake
•Do not reveal whether the username or password was wrong -- use a generic error message
•OAuth2PasswordRequestForm is a Depends class, not a Pydantic model -- use Depends() not Body()
•The response MUST include 'token_type' field alongside 'access_token'

OAuth2PasswordRequestForm parses username and password from form data (not JSON). The token endpoint verifies credentials and returns {"access_token": ..., "token_type": "bearer"}. Use data= not json= in test clients.

•OAuth2PasswordRequestForm parses form fields, not JSON body
•Response must include both access_token and token_type

@app.post("/token")
def login(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]):
    user = authenticate(form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=401, detail="Bad credentials")
    token = jwt.encode({"sub": user["username"], "exp": ...}, SECRET)
    return {"access_token": token, "token_type": "bearer"}

# Test: use data=, not json=
client.post("/token", data={"username": "alice", "password": "secret"})

Use data={} not json={} when testing -- it is form data, not JSON

Question 11 / 18

How do you get the current user from a JWT token?

The get_current_user pattern is the central authentication dependency in FastAPI applications. It combines token extraction, JWT decoding, and user lookup into a single reusable dependency.

Typical dependency chain:

OAuth2PasswordBearer extracts raw token from Authorization header
get_current_user decodes JWT, looks up user in DB, returns User model
get_current_active_user (optional) checks if the user is not disabled/banned

Best practices:

Return a Pydantic model or ORM model, not a raw dict
Create a reusable CurrentUser type alias with Annotated
Separate "user exists" from "user is active" for clear error messages
Use sub (subject) claim for the user identifier per JWT standard

Database lookup:

In production, you typically decode the JWT to get the user ID, then fetch the full user from the database. This ensures you get the latest user data (role changes, bans) even during a token's lifetime.

Pseudo Code

# Dependency chain:
# token (str) -> get_current_user -> get_current_active_user

def get_current_user(token):
    payload = decode_jwt(token)
    user = db.get_user(payload["sub"])
    if not user:
        raise 401
    return user

def get_current_active_user(user = Depends(get_current_user)):
    if user.disabled:
        raise 400 "Inactive user"
    return user

Code

from typing import Annotated
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from fastapi.testclient import TestClient
from pydantic import BaseModel
from jose import jwt, JWTError
from datetime import datetime, timedelta, timezone

app = FastAPI()

SECRET = "test-secret"
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# Simulated database
class UserInDB(BaseModel):
    username: str
    email: str
    role: str = "user"
    disabled: bool = False

fake_db = {
    "alice": UserInDB(username="alice", email="alice@example.com", role="admin"),
    "bob": UserInDB(username="bob", email="bob@example.com", disabled=True),
}

def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
) -> UserInDB:
    """Decode JWT and fetch user from database."""
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET, algorithms=["HS256"])
        username: str | None = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception

    # Look up user in database
    user = fake_db.get(username)
    if user is None:
        raise credentials_exception
    return user

def get_current_active_user(
    current_user: Annotated[UserInDB, Depends(get_current_user)],
) -> UserInDB:
    """Ensure the user account is not disabled."""
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

# Reusable type aliases
CurrentUser = Annotated[UserInDB, Depends(get_current_user)]
ActiveUser = Annotated[UserInDB, Depends(get_current_active_user)]

@app.get("/me")
def read_me(user: ActiveUser):
    return {"username": user.username, "email": user.email, "role": user.role}

# --- Test ---
def make_token(username: str) -> str:
    exp = datetime.now(timezone.utc) + timedelta(hours=1)
    return jwt.encode({"sub": username, "exp": exp}, SECRET, algorithm="HS256")

client = TestClient(app)

# Active user
token = make_token("alice")
resp = client.get("/me", headers={"Authorization": f"Bearer {token}"})
print(resp.json())

# Disabled user
token = make_token("bob")
resp = client.get("/me", headers={"Authorization": f"Bearer {token}"})
print(resp.status_code, resp.json()["detail"])

# Non-existent user
token = make_token("nobody")
resp = client.get("/me", headers={"Authorization": f"Bearer {token}"})
print(resp.status_code)

{'username': 'alice', 'email': 'alice@example.com', 'role': 'admin'}
400 Inactive user
401

Key Points

•Chain dependencies: token extraction -> user lookup -> active check
•Always look up the user in the database to get fresh data (role, disabled status)
•Create Annotated aliases (CurrentUser, ActiveUser) for clean endpoint signatures
•Separate get_current_user from get_current_active_user for clear error messages

Gotchas

•A deleted user with a valid (non-expired) token can still pass JWT decoding -- always check the DB
•Do not expose whether the user does not exist vs bad token -- use a generic 401 message
•The get_current_active_user check runs on every request -- consider caching for high-traffic apps

Create get_current_user that decodes the JWT, looks up the user in the DB, and returns a User model. Chain get_current_active_user to check if the account is disabled. Use Annotated aliases for clean reuse.

•Chain: token -> get_current_user -> get_current_active_user

CurrentUser = Annotated[User, Depends(get_current_user)]
ActiveUser = Annotated[User, Depends(get_current_active_user)]

def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]) -> User:
    payload = jwt.decode(token, SECRET, algorithms=["HS256"])
    user = db.get(payload["sub"])
    if not user:
        raise HTTPException(status_code=401)
    return user

def get_current_active_user(user: CurrentUser) -> User:
    if user.disabled:
        raise HTTPException(status_code=400, detail="Inactive")
    return user

Always check the database -- a deleted user may still have a valid JWT

Question 12 / 18

How do you handle token expiration?

Token expiration is handled through the exp (expiration) claim in JWT tokens. The JWT library automatically rejects expired tokens during decoding.

How it works:

When creating the token, set exp to current_time + timedelta
When decoding, the library checks if exp is in the past
If expired, it raises ExpiredSignatureError (a subclass of JWTError)

Best practices:

Access tokens: 15-30 minutes
Refresh tokens: 7-30 days
Use timezone.utc for all datetime operations to avoid timezone bugs
Return a clear error message so the client knows to refresh
Include iat (issued at) claim for audit/debugging

Client-side handling:

The client should catch 401 responses, check if the error indicates expiration, and automatically use the refresh token to get a new access token. If the refresh also fails, redirect to login.

Pseudo Code

# Creation with expiry:
payload = {"sub": user, "exp": now_utc + timedelta(minutes=30), "iat": now_utc}
token = jwt.encode(payload, secret)

# Decoding catches expiration:
try:
    payload = jwt.decode(token, secret, algorithms=["HS256"])
except ExpiredSignatureError:
    raise 401 "Token expired, please refresh"
except JWTError:
    raise 401 "Invalid token"

Code

from datetime import datetime, timedelta, timezone
from jose import jwt, JWTError, ExpiredSignatureError
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from fastapi.testclient import TestClient
from typing import Annotated

app = FastAPI()

SECRET = "test-secret"
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def create_access_token(username: str, expires_delta: timedelta) -> str:
    now = datetime.now(timezone.utc)
    payload = {
        "sub": username,
        "exp": now + expires_delta,
        "iat": now,  # Issued at -- useful for debugging
    }
    return jwt.encode(payload, SECRET, algorithm="HS256")

def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]) -> dict:
    try:
        payload = jwt.decode(token, SECRET, algorithms=["HS256"])
        return {"username": payload["sub"]}
    except ExpiredSignatureError:
        # Specific error for expired tokens
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token has expired",
            headers={"WWW-Authenticate": "Bearer error=\"invalid_token\""},
        )
    except JWTError:
        # Generic error for all other JWT issues
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Could not validate credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )

@app.get("/me")
def me(user: Annotated[dict, Depends(get_current_user)]):
    return user

# --- Test ---
client = TestClient(app)

# Valid token (1 hour)
token = create_access_token("alice", timedelta(hours=1))
resp = client.get("/me", headers={"Authorization": f"Bearer {token}"})
print(resp.json())

# Expired token (negative delta = already expired)
expired = create_access_token("alice", timedelta(seconds=-1))
resp = client.get("/me", headers={"Authorization": f"Bearer {expired}"})
print(resp.status_code, resp.json()["detail"])

# Tampered token
resp = client.get("/me", headers={"Authorization": "Bearer invalid.token.here"})
print(resp.status_code, resp.json()["detail"])

{'username': 'alice'}
401 Token has expired
401 Could not validate credentials

Key Points

•The 'exp' claim is checked automatically by jwt.decode()
•Catch ExpiredSignatureError separately to give specific expiry feedback
•Include 'iat' (issued at) claim for debugging token lifetime issues
•Use timezone-aware UTC datetimes to avoid timezone-related bugs

Gotchas

•ExpiredSignatureError is a subclass of JWTError -- catch it BEFORE the generic JWTError
•Naive datetimes (without timezone) can cause tokens to expire at wrong times across timezones
•There is no way to invalidate a JWT before its exp without server-side blacklisting
•Clock skew between servers can cause unexpected expiration -- consider a small leeway

Set exp claim when creating tokens. jwt.decode() automatically rejects expired tokens with ExpiredSignatureError. Catch it separately from JWTError to give a specific "token expired" message so the client knows to refresh.

•exp claim is checked automatically by jwt.decode()
•Catch ExpiredSignatureError before JWTError for specific error messages

from jose import jwt, ExpiredSignatureError, JWTError

try:
    payload = jwt.decode(token, SECRET, algorithms=["HS256"])
except ExpiredSignatureError:
    raise HTTPException(status_code=401, detail="Token expired")
except JWTError:
    raise HTTPException(status_code=401, detail="Invalid token")

ExpiredSignatureError must be caught BEFORE JWTError since it is a subclass

Question 13 / 18

What is the difference between OAuth2PasswordBearer and HTTPBearer?

Both extract Bearer tokens from the Authorization header, but they differ in their OpenAPI representation and return types.

OAuth2PasswordBearer:

OpenAPI type: oauth2 with password flow
Swagger UI: Shows "Authorize" with username/password fields + sends to tokenUrl
Returns: Raw token string (str)
Default error: 401 Unauthorized
Use when: Your API has its own login endpoint and you want the full OAuth2 password flow in docs

HTTPBearer:

OpenAPI type: http with bearer scheme
Swagger UI: Shows "Authorize" with a single token input field
Returns: HTTPAuthorizationCredentials object with .scheme and .credentials
Default error: 403 Forbidden
Use when: Token is obtained externally (third-party auth, API gateway) or you want a simpler auth dialog

Decision guide:

Scenario	Use
Full login flow in same API	OAuth2PasswordBearer
External auth provider (Auth0, Firebase)	HTTPBearer
Service-to-service with pre-shared tokens	HTTPBearer
Need Swagger username/password dialog	OAuth2PasswordBearer
Microservice behind API gateway	HTTPBearer

Pseudo Code

# OAuth2PasswordBearer:
oauth2 = OAuth2PasswordBearer(tokenUrl="token")
def dep(token: str = Depends(oauth2)):  # returns str
    ...

# HTTPBearer:
bearer = HTTPBearer()
def dep(creds: HTTPAuthorizationCredentials = Depends(bearer)):  # returns object
    token = creds.credentials  # extract str

Code

from fastapi import FastAPI, Depends
from fastapi.security import OAuth2PasswordBearer, HTTPBearer, HTTPAuthorizationCredentials
from fastapi.testclient import TestClient
from typing import Annotated

app = FastAPI()

# OAuth2PasswordBearer -- returns raw string
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# HTTPBearer -- returns HTTPAuthorizationCredentials
http_bearer = HTTPBearer()

@app.get("/oauth2-style")
def oauth2_endpoint(token: Annotated[str, Depends(oauth2_scheme)]):
    return {"type": "OAuth2PasswordBearer", "value": token, "is_str": isinstance(token, str)}

@app.get("/bearer-style")
def bearer_endpoint(
    creds: Annotated[HTTPAuthorizationCredentials, Depends(http_bearer)],
):
    return {
        "type": "HTTPBearer",
        "scheme": creds.scheme,
        "credentials": creds.credentials,
        "is_str": isinstance(creds, str),
    }

# --- Test ---
client = TestClient(app)
headers = {"Authorization": "Bearer my-token-123"}

resp = client.get("/oauth2-style", headers=headers)
print(resp.json())

resp = client.get("/bearer-style", headers=headers)
print(resp.json())

# Missing token behavior
print("OAuth2 missing:", client.get("/oauth2-style").status_code)  # 401
print("HTTPBearer missing:", client.get("/bearer-style").status_code)  # 403

{'type': 'OAuth2PasswordBearer', 'value': 'my-token-123', 'is_str': True}
{'type': 'HTTPBearer', 'scheme': 'Bearer', 'credentials': 'my-token-123', 'is_str': False}
OAuth2 missing: 401
HTTPBearer missing: 403

Key Points

•OAuth2PasswordBearer returns a raw string; HTTPBearer returns HTTPAuthorizationCredentials
•OAuth2PasswordBearer shows username/password in Swagger; HTTPBearer shows a token field
•OAuth2PasswordBearer defaults to 401; HTTPBearer defaults to 403 for missing tokens
•Use OAuth2PasswordBearer when your API has its own login flow, HTTPBearer for external auth

Gotchas

•Mixing them up changes your Swagger UI experience significantly
•HTTPBearer returning 403 instead of 401 can confuse clients expecting standard OAuth2 behavior
•Both read from the same Authorization header -- they differ only in OpenAPI representation and return type

OAuth2PasswordBearer returns a raw token string, shows username/password in Swagger UI, and defaults to 401. HTTPBearer returns HTTPAuthorizationCredentials (with .credentials), shows a token-only input, and defaults to 403. Use OAuth2PasswordBearer for login flows, HTTPBearer for external auth.

•OAuth2PasswordBearer -> str + 401 + Swagger login flow
•HTTPBearer -> HTTPAuthorizationCredentials + 403 + token-only input

# OAuth2PasswordBearer: returns str, 401 on missing, username/password in Swagger
oauth2 = OAuth2PasswordBearer(tokenUrl="token")
def dep(token: str = Depends(oauth2)): ...  # token is a string

# HTTPBearer: returns object, 403 on missing, token-only in Swagger
bearer = HTTPBearer()
def dep(creds = Depends(bearer)): token = creds.credentials  # extract string

HTTPBearer returns 403, not 401 -- this can break client-side refresh logic

Question 14 / 18

How do you implement multi-factor authentication concepts?

Multi-Factor Authentication (MFA) adds a second verification step beyond username/password. The most common approach is TOTP (Time-based One-Time Password) using apps like Google Authenticator or Authy.

TOTP flow:

Setup: Generate a secret key, show QR code to user, user scans with authenticator app
Login step 1: User provides username + password, gets a partial/temporary token
Login step 2: User provides TOTP code from their authenticator app
Validation: Server generates the expected code from the stored secret and compares

Implementation approach:

Use pyotp library for TOTP generation and verification
Store the TOTP secret per user in the database (encrypted)
Issue a limited-scope "mfa-pending" token after password verification
Issue a full access token only after TOTP verification

Two-step login with JWT:

Password verified -> issue token with {"mfa_verified": false}
TOTP verified -> issue token with {"mfa_verified": true}
Protected routes check for mfa_verified: true

Pseudo Code

# Setup MFA:
secret = pyotp.random_base32()
totp = pyotp.TOTP(secret)
qr_url = totp.provisioning_uri(email, issuer_name="MyApp")
store_secret(user, secret)

# Login step 1 (password):
if verify_password(password, user.hash):
    return partial_token(user, mfa_verified=False)

# Login step 2 (TOTP):
if totp.verify(user_code):
    return full_token(user, mfa_verified=True)

# Protected route:
if not token.mfa_verified:
    raise 403 "MFA required"

Code

import pyotp
from datetime import datetime, timedelta, timezone
from typing import Annotated
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from fastapi.testclient import TestClient
from pydantic import BaseModel
from jose import jwt

app = FastAPI()

SECRET = "jwt-secret"
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# Simulated user DB with MFA secret
users_db = {
    "alice": {
        "username": "alice",
        "password": "secret123",
        "mfa_secret": pyotp.random_base32(),  # Store encrypted in real DB
        "mfa_enabled": True,
    },
}

class MFASetupResponse(BaseModel):
    secret: str
    qr_uri: str

def create_token(username: str, mfa_verified: bool) -> str:
    return jwt.encode(
        {
            "sub": username,
            "mfa_verified": mfa_verified,
            "exp": datetime.now(timezone.utc) + timedelta(hours=1),
        },
        SECRET,
        algorithm="HS256",
    )

@app.post("/mfa/setup")
def setup_mfa(username: str):
    """Generate MFA secret and QR code URI."""
    secret = pyotp.random_base32()
    totp = pyotp.TOTP(secret)
    qr_uri = totp.provisioning_uri(name=username, issuer_name="MyApp")
    # In production: store secret in DB
    return MFASetupResponse(secret=secret, qr_uri=qr_uri)

@app.post("/login")
def login(username: str, password: str):
    """Step 1: Password verification."""
    user = users_db.get(username)
    if not user or user["password"] != password:
        raise HTTPException(status_code=401, detail="Invalid credentials")

    if user["mfa_enabled"]:
        # Issue partial token -- MFA still required
        token = create_token(username, mfa_verified=False)
        return {"access_token": token, "mfa_required": True}

    # No MFA -- issue full token
    token = create_token(username, mfa_verified=True)
    return {"access_token": token, "mfa_required": False}

@app.post("/mfa/verify")
def verify_mfa(
    code: str,
    token: Annotated[str, Depends(oauth2_scheme)],
):
    """Step 2: TOTP verification."""
    payload = jwt.decode(token, SECRET, algorithms=["HS256"])
    username = payload["sub"]
    user = users_db[username]

    totp = pyotp.TOTP(user["mfa_secret"])
    if not totp.verify(code, valid_window=1):  # Allow 30s window
        raise HTTPException(status_code=401, detail="Invalid MFA code")

    # Issue full access token
    full_token = create_token(username, mfa_verified=True)
    return {"access_token": full_token, "mfa_verified": True}

# --- Test ---
client = TestClient(app)

# Step 1: Login
resp = client.post("/login?username=alice&password=secret123")
print(f"MFA required: {resp.json()['mfa_required']}")
partial_token = resp.json()["access_token"]

# Step 2: Verify TOTP
user = users_db["alice"]
totp = pyotp.TOTP(user["mfa_secret"])
current_code = totp.now()
resp = client.post(
    f"/mfa/verify?code={current_code}",
    headers={"Authorization": f"Bearer {partial_token}"},
)
print(f"MFA verified: {resp.json()['mfa_verified']}")

MFA required: True
MFA verified: True

Key Points

•Use pyotp for TOTP generation and verification
•Issue a partial token (mfa_verified=false) after password, full token after TOTP
•Store the TOTP secret encrypted in the database per user
•valid_window parameter allows a time window to account for clock drift

Gotchas

•TOTP codes change every 30 seconds -- use valid_window to allow slight drift
•The MFA secret must be stored securely (encrypted) -- if leaked, MFA is bypassed
•The partial token should have limited scope or short expiry to prevent misuse
•Always verify the partial token belongs to the same user requesting MFA verification

Use pyotp for TOTP-based MFA. After password login, issue a partial token (mfa_verified: false). After the user submits the TOTP code from their authenticator app, verify with totp.verify(code) and issue a full token (mfa_verified: true).

•pyotp handles TOTP generation and verification
•Issue partial token after password, full token after TOTP

import pyotp

# Setup: generate secret and QR URI
secret = pyotp.random_base32()
totp = pyotp.TOTP(secret)
uri = totp.provisioning_uri(name="user@example.com", issuer_name="MyApp")

# Verify: check the code from the authenticator app
is_valid = totp.verify("123456", valid_window=1)  # 30s tolerance

TOTP codes expire every 30 seconds -- always use valid_window for tolerance

Question 15 / 18

How do you handle CORS with authentication?

CORS (Cross-Origin Resource Sharing) with authentication requires special configuration because browsers enforce strict rules when requests include credentials (cookies, Authorization headers).

Key rules when using credentials:

allow_origins cannot be ["*"] -- you must list specific origins
allow_credentials must be True
The Access-Control-Allow-Origin response header must echo the exact requesting origin, not *
Preflight (OPTIONS) requests must also return the correct CORS headers

Common setup:

Frontend at http://localhost:3000 sends requests with Authorization: Bearer ... header
Backend at http://localhost:8000 must allow that origin with credentials
The browser sends a preflight OPTIONS request first for non-simple requests

What breaks without proper CORS:

Browser blocks the response (the request is actually sent but the response is hidden)
JavaScript cannot read the response body or headers
Error in browser console: "has been blocked by CORS policy"

Pseudo Code

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],  # NOT "*" with credentials
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],  # Includes Authorization header
)

Code

from fastapi import FastAPI, Depends
from fastapi.middleware.cors import CORSMiddleware
from fastapi.security import OAuth2PasswordBearer
from fastapi.testclient import TestClient
from typing import Annotated

app = FastAPI()

# CORS configuration for authenticated requests
app.add_middleware(
    CORSMiddleware,
    # Must list specific origins when allow_credentials=True
    allow_origins=[
        "http://localhost:3000",       # React dev server
        "http://localhost:5173",       # Vite dev server
        "https://myapp.example.com",   # Production frontend
    ],
    allow_credentials=True,  # Allow cookies and Authorization headers
    allow_methods=["GET", "POST", "PUT", "DELETE", "PATCH"],
    allow_headers=[
        "Authorization",   # For Bearer tokens
        "Content-Type",    # For JSON requests
        "X-API-Key",       # Custom headers if needed
    ],
    expose_headers=["X-Total-Count"],  # Headers the client can read
    max_age=600,  # Cache preflight responses for 10 minutes
)

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.get("/api/data")
def get_data(token: Annotated[str, Depends(oauth2_scheme)]):
    return {"data": "secret", "token_received": True}

# --- Dynamic origins (for multi-tenant apps) ---
from starlette.middleware.cors import CORSMiddleware as StarletteCORS

def get_allowed_origins() -> list[str]:
    """Fetch allowed origins from config or database."""
    return ["http://localhost:3000", "https://app.example.com"]

# For dynamic origins, subclass the middleware or use a custom middleware

# --- Test ---
client = TestClient(app)

# Simulate cross-origin request with credentials
resp = client.get(
    "/api/data",
    headers={
        "Authorization": "Bearer test-token",
        "Origin": "http://localhost:3000",
    },
)
print(f"Status: {resp.status_code}")
print(f"CORS header: {resp.headers.get('access-control-allow-origin')}")
print(f"Credentials: {resp.headers.get('access-control-allow-credentials')}")

# Preflight request
resp = client.options(
    "/api/data",
    headers={
        "Origin": "http://localhost:3000",
        "Access-Control-Request-Method": "GET",
        "Access-Control-Request-Headers": "Authorization",
    },
)
print(f"Preflight status: {resp.status_code}")
print(f"Allowed headers: {resp.headers.get('access-control-allow-headers')}")

Status: 200
CORS header: http://localhost:3000
Credentials: true
Preflight status: 200
Allowed headers: Authorization

Key Points

•allow_origins cannot be ['*'] when allow_credentials is True
•allow_credentials=True is required for Authorization headers and cookies
•Browsers send preflight OPTIONS requests for non-simple cross-origin requests
•expose_headers controls which response headers JavaScript can read

Gotchas

•Setting allow_origins=['*'] with allow_credentials=True silently fails -- the browser rejects the response
•Preflight requests do not include the Authorization header -- only the actual request does
•max_age caches preflight responses -- set it to avoid repeated OPTIONS requests
•CORS is enforced by the browser, not the server -- curl and Postman always work

When using credentials (Bearer tokens, cookies), allow_origins must list specific origins (not "*"), and allow_credentials must be True. Browsers block responses if CORS headers do not match.

•allow_origins=['*'] does NOT work with allow_credentials=True

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],  # NOT "*"
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

CORS is browser-enforced -- server returns headers but cannot block curl/Postman

Question 16 / 18

How do you implement API rate limiting per user?

Rate limiting prevents API abuse by restricting how many requests a user can make within a time window. In FastAPI, implement this as a dependency using in-memory storage or Redis.

Approaches:

In-memory: Simple dict with timestamps -- works for single-server deployments
Redis: Production-grade, works across multiple servers/workers
slowapi: Third-party library wrapping limits for FastAPI

Sliding window algorithm:

Track request timestamps per user
On each request, remove timestamps older than the window
If remaining count exceeds limit, reject with 429 Too Many Requests
Return Retry-After header telling the client when to retry

Rate limit headers (standard):

X-RateLimit-Limit: Maximum requests per window
X-RateLimit-Remaining: Requests remaining
X-RateLimit-Reset: Unix timestamp when the window resets
Retry-After: Seconds until the client should retry (on 429)

Pseudo Code

rate_limits = {}  # {user_id: [timestamp1, timestamp2, ...]}

def check_rate_limit(user, max_requests, window_seconds):
    now = time.time()
    timestamps = rate_limits.get(user, [])
    # Remove old timestamps
    timestamps = [t for t in timestamps if now - t < window_seconds]
    if len(timestamps) >= max_requests:
        raise 429 Too Many Requests
    timestamps.append(now)
    rate_limits[user] = timestamps

Code

import time
from collections import defaultdict
from typing import Annotated
from fastapi import FastAPI, Depends, HTTPException, status, Request, Response
from fastapi.security import OAuth2PasswordBearer
from fastapi.testclient import TestClient
from jose import jwt
from datetime import datetime, timedelta, timezone

app = FastAPI()

SECRET = "test-secret"
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# In-memory rate limit storage (use Redis in production)
rate_limit_store: dict[str, list[float]] = defaultdict(list)

def rate_limit(max_requests: int = 10, window_seconds: int = 60):
    """Dependency factory for per-user rate limiting."""
    def limiter(
        request: Request,
        response: Response,
        token: Annotated[str, Depends(oauth2_scheme)],
    ):
        # Identify user from token
        payload = jwt.decode(token, SECRET, algorithms=["HS256"])
        user_id = payload["sub"]

        now = time.time()
        # Clean old entries
        timestamps = [
            t for t in rate_limit_store[user_id]
            if now - t < window_seconds
        ]
        rate_limit_store[user_id] = timestamps

        # Set rate limit headers
        remaining = max(0, max_requests - len(timestamps))
        response.headers["X-RateLimit-Limit"] = str(max_requests)
        response.headers["X-RateLimit-Remaining"] = str(remaining)
        response.headers["X-RateLimit-Reset"] = str(
            int(now + window_seconds)
        )

        if len(timestamps) >= max_requests:
            retry_after = int(window_seconds - (now - timestamps[0]))
            raise HTTPException(
                status_code=status.HTTP_429_TOO_MANY_REQUESTS,
                detail="Rate limit exceeded",
                headers={"Retry-After": str(retry_after)},
            )

        timestamps.append(now)
        rate_limit_store[user_id] = timestamps
        return user_id
    return limiter

@app.get("/api/data")
def get_data(
    user_id: Annotated[str, Depends(rate_limit(max_requests=3, window_seconds=60))],
):
    return {"user": user_id, "data": "here"}

# --- Test ---
def make_token(username: str) -> str:
    exp = datetime.now(timezone.utc) + timedelta(hours=1)
    return jwt.encode({"sub": username, "exp": exp}, SECRET, algorithm="HS256")

client = TestClient(app)
headers = {"Authorization": f"Bearer {make_token('alice')}"}

# Make 4 requests (limit is 3)
for i in range(4):
    resp = client.get("/api/data", headers=headers)
    print(f"Request {i+1}: {resp.status_code} - Remaining: {resp.headers.get('X-RateLimit-Remaining', 'N/A')}")

Request 1: 200 - Remaining: 2
Request 2: 200 - Remaining: 1
Request 3: 200 - Remaining: 0
Request 4: 429 - Remaining: N/A

Key Points

•Use a dependency factory to make rate limits configurable per endpoint
•Sliding window: track timestamps, remove old ones, count remaining
•Return standard rate limit headers (X-RateLimit-Limit, Remaining, Reset)
•Use 429 Too Many Requests with Retry-After header

Gotchas

•In-memory storage does not work with multiple workers/servers -- use Redis for production
•HTTPException in the rate limiter prevents response.headers from reaching the client -- set headers on the exception instead
•Rate limiting by IP instead of user catches unauthenticated abuse but penalizes shared IPs (offices, VPNs)
•time.time() returns float -- be careful with comparison precision

Create a dependency that tracks request timestamps per user in a sliding window. If the count exceeds the limit, raise 429 with a Retry-After header. Use Redis for multi-server deployments.

•Sliding window: track timestamps, remove expired, count remaining

rate_limits = defaultdict(list)

def rate_limit(max_req: int = 10, window: int = 60):
    def limiter(token: str = Depends(oauth2_scheme)):
        user = jwt.decode(token, SECRET)["sub"]
        now = time.time()
        timestamps = [t for t in rate_limits[user] if now - t < window]
        if len(timestamps) >= max_req:
            raise HTTPException(status_code=429, detail="Rate limit exceeded")
        timestamps.append(now)
        rate_limits[user] = timestamps
    return limiter

In-memory storage does not work across multiple workers -- use Redis in production

Question 17 / 18

How do you secure WebSocket connections?

WebSocket connections cannot use standard HTTP Authorization headers after the initial handshake, so authentication requires different approaches.

Authentication methods for WebSockets:

Query parameter token: Pass token in the WebSocket URL (ws://host/ws?token=xxx)
- Simple but token appears in server logs and browser history
First message authentication: Connect first, send token as the first message
- Cleaner but requires protocol agreement between client and server
Cookie-based: If using session cookies, they are sent automatically with the WebSocket handshake
- Works well with browser-based clients

Why not Authorization header:

The browser's WebSocket API (new WebSocket(url)) does not support custom headers
The Authorization header cannot be set on WebSocket connections from browsers
Server-side WebSocket clients (Python, Node.js) can set headers, but browser clients cannot

Best practice:

Validate the token during the WebSocket handshake (before accept())
Close the connection immediately if authentication fails
Periodically re-validate if using long-lived connections

Pseudo Code

# Query parameter approach:
@app.websocket("/ws")
async def ws(websocket, token: str = Query(...)):
    user = validate_token(token)
    if not user:
        await websocket.close(code=1008)  # Policy Violation
        return
    await websocket.accept()
    # ... handle messages

# First message approach:
@app.websocket("/ws")
async def ws(websocket):
    await websocket.accept()
    auth_msg = await websocket.receive_text()
    user = validate_token(auth_msg)
    if not user:
        await websocket.close(code=1008)
        return

Code

from typing import Annotated
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Query, status
from fastapi.testclient import TestClient
from jose import jwt, JWTError
from datetime import datetime, timedelta, timezone

app = FastAPI()
SECRET = "test-secret"

def validate_ws_token(token: str) -> dict | None:
    """Validate JWT token for WebSocket connections."""
    try:
        payload = jwt.decode(token, SECRET, algorithms=["HS256"])
        return {"username": payload["sub"]}
    except JWTError:
        return None

# --- Method 1: Query parameter token ---
@app.websocket("/ws/query")
async def ws_query_auth(
    websocket: WebSocket,
    token: Annotated[str | None, Query()] = None,
):
    if not token:
        await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
        return
    user = validate_ws_token(token)
    if not user:
        await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
        return

    await websocket.accept()
    await websocket.send_json({"message": f"Hello {user['username']}"})
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_json({"echo": data, "user": user["username"]})
    except WebSocketDisconnect:
        pass

# --- Method 2: First message authentication ---
@app.websocket("/ws/first-msg")
async def ws_first_msg_auth(websocket: WebSocket):
    await websocket.accept()

    # First message must be the token
    try:
        token = await websocket.receive_text()
    except WebSocketDisconnect:
        return

    user = validate_ws_token(token)
    if not user:
        await websocket.send_json({"error": "Authentication failed"})
        await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
        return

    await websocket.send_json({"authenticated": True, "user": user["username"]})
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_json({"echo": data})
    except WebSocketDisconnect:
        pass

# --- Test ---
def make_token(username: str) -> str:
    exp = datetime.now(timezone.utc) + timedelta(hours=1)
    return jwt.encode({"sub": username, "exp": exp}, SECRET, algorithm="HS256")

client = TestClient(app)
token = make_token("alice")

# Test query param auth
with client.websocket_connect(f"/ws/query?token={token}") as ws:
    greeting = ws.receive_json()
    print(greeting)
    ws.send_text("hello")
    echo = ws.receive_json()
    print(echo)

# Test first message auth
with client.websocket_connect("/ws/first-msg") as ws:
    ws.send_text(token)
    auth_resp = ws.receive_json()
    print(auth_resp)

{'message': 'Hello alice'}
{'echo': 'hello', 'user': 'alice'}
{'authenticated': True, 'user': 'alice'}

Key Points

•Browser WebSocket API cannot set Authorization headers -- use query params or first message
•Validate token before calling websocket.accept() when possible (query param method)
•Use WebSocket close code 1008 (Policy Violation) for authentication failures
•Re-validate tokens periodically for long-lived WebSocket connections

Gotchas

•Query parameter tokens appear in server logs, proxy logs, and browser history
•Calling websocket.close() before accept() requires no status code in some frameworks
•WebSocket dependencies (Query, Header) only work during the handshake, not after accept()
•Unlike HTTP, a closed WebSocket cannot return a JSON error body -- send the error before closing

WebSocket connections from browsers cannot set Authorization headers. Use query parameter tokens (ws://host/ws?token=xxx) or send the token as the first WebSocket message. Validate before accept() when possible, and close with code 1008 on failure.

•Use query param or first message for WebSocket auth -- browsers cannot set custom headers

@app.websocket("/ws")
async def websocket_endpoint(
    websocket: WebSocket,
    token: str = Query(...),
):
    user = validate_token(token)
    if not user:
        await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
        return
    await websocket.accept()
    await websocket.send_json({"hello": user["username"]})

Tokens in query params appear in server logs -- use TLS and short-lived tokens

Question 18 / 18

How do you implement permission-based authorization?

Permission-based authorization is more granular than role-based. Instead of checking if a user has a role, you check if they have specific permissions (e.g., create_post, delete_user, view_reports).

RBAC vs Permission-based:

RBAC: "Is this user an admin?" -> admin can do everything
Permission-based: "Does this user have the delete_user permission?" -> granular control
Combined: Roles grant sets of permissions; check permissions at the endpoint level

Implementation:

Define permissions as an Enum or set of strings
Map roles to permission sets
Create a dependency factory require_permissions(*perms) that checks the user's permissions
Permissions can come from JWT claims, database, or role-to-permission mapping

Advantages over simple RBAC:

Adding a new role does not require changing endpoint code
Permissions can be assigned individually (beyond role defaults)
Easier to audit ("who can delete_user?" vs "what can admins do?")

Pseudo Code

# Define permissions per role:
ROLE_PERMISSIONS = {
    "admin": {"create", "read", "update", "delete", "manage_users"},
    "editor": {"create", "read", "update"},
    "viewer": {"read"},
}

def require_permissions(*required):
    def checker(user = Depends(get_current_user)):
        user_perms = ROLE_PERMISSIONS[user.role]
        missing = set(required) - user_perms
        if missing:
            raise 403 f"Missing permissions: {missing}"
        return user
    return checker

@app.delete("/users/{id}", dependencies=[Depends(require_permissions("delete", "manage_users"))])
def delete_user(id): ...

Code

from enum import Enum
from typing import Annotated
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from fastapi.testclient import TestClient
from pydantic import BaseModel
from jose import jwt
from datetime import datetime, timedelta, timezone

app = FastAPI()
SECRET = "test-secret"
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# --- Permission definitions ---
class Permission(str, Enum):
    READ_POSTS = "read_posts"
    CREATE_POSTS = "create_posts"
    DELETE_POSTS = "delete_posts"
    MANAGE_USERS = "manage_users"
    VIEW_REPORTS = "view_reports"

# Role-to-permission mapping
ROLE_PERMISSIONS: dict[str, set[str]] = {
    "admin": {
        Permission.READ_POSTS, Permission.CREATE_POSTS,
        Permission.DELETE_POSTS, Permission.MANAGE_USERS,
        Permission.VIEW_REPORTS,
    },
    "editor": {
        Permission.READ_POSTS, Permission.CREATE_POSTS,
        Permission.DELETE_POSTS,
    },
    "viewer": {
        Permission.READ_POSTS,
    },
}

class User(BaseModel):
    username: str
    role: str
    permissions: set[str] = set()

def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]) -> User:
    payload = jwt.decode(token, SECRET, algorithms=["HS256"])
    role = payload.get("role", "viewer")
    # Get permissions from role + any extra permissions
    perms = ROLE_PERMISSIONS.get(role, set())
    extra = set(payload.get("extra_permissions", []))
    return User(
        username=payload["sub"],
        role=role,
        permissions=perms | extra,
    )

CurrentUser = Annotated[User, Depends(get_current_user)]

def require_permissions(*required: str):
    """Dependency factory that checks for specific permissions."""
    def checker(current_user: CurrentUser) -> User:
        missing = set(required) - current_user.permissions
        if missing:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"Missing permissions: {', '.join(sorted(missing))}",
            )
        return current_user
    return checker

@app.get("/posts")
def list_posts(
    user: Annotated[User, Depends(require_permissions(Permission.READ_POSTS))],
):
    return {"user": user.username, "action": "listing posts"}

@app.post("/posts")
def create_post(
    user: Annotated[User, Depends(require_permissions(Permission.CREATE_POSTS))],
):
    return {"user": user.username, "action": "creating post"}

@app.delete("/posts/{post_id}")
def delete_post(
    post_id: int,
    user: Annotated[
        User,
        Depends(require_permissions(Permission.DELETE_POSTS)),
    ],
):
    return {"user": user.username, "action": f"deleting post {post_id}"}

@app.get("/admin/users")
def manage_users(
    user: Annotated[User, Depends(require_permissions(Permission.MANAGE_USERS))],
):
    return {"user": user.username, "action": "managing users"}

# --- Test ---
def make_token(username: str, role: str, extra_perms: list[str] | None = None) -> str:
    exp = datetime.now(timezone.utc) + timedelta(hours=1)
    payload = {"sub": username, "role": role, "exp": exp}
    if extra_perms:
        payload["extra_permissions"] = extra_perms
    return jwt.encode(payload, SECRET, algorithm="HS256")

client = TestClient(app)

# Viewer can read but not create
viewer_token = make_token("charlie", "viewer")
resp = client.get("/posts", headers={"Authorization": f"Bearer {viewer_token}"})
print(resp.json())

resp = client.post("/posts", headers={"Authorization": f"Bearer {viewer_token}"})
print(f"{resp.status_code}: {resp.json()['detail']}")

# Editor can create and delete
editor_token = make_token("bob", "editor")
resp = client.delete("/posts/42", headers={"Authorization": f"Bearer {editor_token}"})
print(resp.json())

# Viewer with extra permission
super_viewer = make_token("charlie", "viewer", [Permission.VIEW_REPORTS])
print(f"Extra perms work: viewer + view_reports")

{'user': 'charlie', 'action': 'listing posts'}
403: Missing permissions: create_posts
{'user': 'bob', 'action': 'deleting post 42'}
Extra perms work: viewer + view_reports

Key Points

•Define permissions as enums or strings, map roles to permission sets
•require_permissions() factory checks if user has all required permissions
•Combine role-based permissions with per-user extra permissions for flexibility
•Use set operations (difference) to find missing permissions efficiently

Gotchas

•Role-to-permission mapping in code requires redeployment to change -- consider storing in DB for dynamic updates
•Permissions in JWT tokens cannot be revoked until the token expires
•Checking permissions on every request can be expensive if fetched from DB -- consider caching
•Permission names should be consistent -- use enums to avoid typos

Map roles to permission sets, then create a require_permissions() dependency factory that checks if the user has all required permissions. This is more granular than RBAC -- new roles just need a permission mapping, no endpoint changes.

•Map roles to permission sets, check with set operations

ROLE_PERMS = {
    "admin": {"read", "write", "delete", "manage_users"},
    "editor": {"read", "write"},
    "viewer": {"read"},
}

def require_permissions(*required):
    def checker(user: CurrentUser):
        user_perms = ROLE_PERMS.get(user.role, set())
        if not set(required).issubset(user_perms):
            raise HTTPException(status_code=403, detail="Insufficient permissions")
        return user
    return checker

Permissions stored in JWT cannot be revoked until the token expires

Built by onllm.dev

Home GitHub Support awesomePrep

How does OAuth2PasswordBearer work in FastAPI?

How it works:

What tokenUrl does:

Integration with OpenAPI:

How do you implement JWT token creation and validation?

Token creation:

Token validation:

Libraries:

Security considerations:

How do you hash and verify passwords with passlib/bcrypt?

How bcrypt works:

Setup:

passlib's CryptContext:

Why bcrypt:

How do you protect routes with authentication dependencies?

Approaches:

How it works:

Using Annotated (modern style):

How do you implement role-based access control (RBAC)?

Approach:

Dependency factory pattern:

RBAC vs ABAC:

How do you implement API key authentication?

FastAPI provides three API key security classes:

How it works:

Best practices:

How does HTTPBearer work for Bearer token auth?

Key differences from OAuth2PasswordBearer:

When to use HTTPBearer:

HTTPAuthorizationCredentials:

What are OAuth2 security scopes and how do you use them?

How FastAPI implements scopes:

Key components:

When to use scopes vs roles:

How do you implement token refresh functionality?

Two-token pattern:

Why two tokens:

Flow:

Refresh token rotation:

How do you implement the complete login flow with OAuth2PasswordRequestForm?

Why form data instead of JSON:

Complete flow:

OAuth2PasswordRequestForm fields:

How do you get the current user from a JWT token?

Typical dependency chain:

Best practices:

Database lookup:

How do you handle token expiration?

How it works:

Best practices:

Client-side handling:

What is the difference between OAuth2PasswordBearer and HTTPBearer?

OAuth2PasswordBearer:

HTTPBearer:

Decision guide:

How do you implement multi-factor authentication concepts?

TOTP flow:

Implementation approach:

Two-step login with JWT:

How do you handle CORS with authentication?

Key rules when using credentials:

Common setup:

What breaks without proper CORS:

How do you implement API rate limiting per user?

Approaches:

Sliding window algorithm:

Rate limit headers (standard):

How do you secure WebSocket connections?

Authentication methods for WebSockets:

Why not Authorization header:

Best practice:

How do you implement permission-based authorization?

RBAC vs Permission-based:

Implementation:

Advantages over simple RBAC:

Security & Authentication

What `tokenUrl` does: