FastAPI Authentication and Authorization Testing
Authentication bugs are the most expensive bugs you can ship. A missing authorization check or a JWT validation flaw can expose every user's data or let attackers elevate their privileges. This guide covers systematic testing of every auth pattern FastAPI supports, with enough depth to catch the edge cases that simple happy-path testing misses.
Testing OAuth2 Password Flow
The OAuth2 password flow is FastAPI's most common auth pattern. Testing it means verifying the token endpoint, token validation, and token expiry:
# app/auth.py
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
to_encode = data.copy()
expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
to_encode["exp"] = expire
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
async def get_current_user(token: str = Depends(oauth2_scheme)):
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
except JWTError:
raise credentials_exception
return {"username": username, "scopes": payload.get("scopes", [])}# tests/test_oauth2.py
import pytest
from fastapi.testclient import TestClient
from datetime import timedelta
from app.main import app
from app.auth import create_access_token, SECRET_KEY, ALGORITHM
from jose import jwt
@pytest.fixture
def client():
return TestClient(app)
class TestTokenEndpoint:
def test_valid_credentials_return_access_token(self, client):
response = client.post("/token", data={
"username": "testuser",
"password": "correctpassword",
"grant_type": "password"
})
assert response.status_code == 200
body = response.json()
assert "access_token" in body
assert body["token_type"] == "bearer"
# Token should be a valid JWT
payload = jwt.decode(body["access_token"], SECRET_KEY, algorithms=[ALGORITHM])
assert payload["sub"] == "testuser"
def test_invalid_password_returns_401(self, client):
response = client.post("/token", data={
"username": "testuser",
"password": "wrongpassword",
"grant_type": "password"
})
assert response.status_code == 401
assert "Incorrect username or password" in response.json()["detail"]
def test_nonexistent_user_returns_401(self, client):
response = client.post("/token", data={
"username": "nosuchuser@example.com",
"password": "anypassword",
"grant_type": "password"
})
assert response.status_code == 401
def test_missing_grant_type_returns_422(self, client):
response = client.post("/token", data={
"username": "testuser",
"password": "password"
# missing grant_type
})
assert response.status_code == 422JWT Token Testing
JWT tokens carry claims that your application trusts. Test every claim your code relies on, and test every invalid token your code should reject:
# tests/test_jwt.py
import pytest
import time
from datetime import datetime, timedelta
from fastapi.testclient import TestClient
from jose import jwt
from app.main import app
from app.auth import create_access_token, SECRET_KEY, ALGORITHM
@pytest.fixture
def client():
return TestClient(app)
def make_token(
username: str = "testuser",
scopes: list[str] = None,
expires_in: int = 30,
secret: str = SECRET_KEY,
algorithm: str = ALGORITHM,
extra_claims: dict = None
) -> str:
data = {"sub": username, "scopes": scopes or []}
if extra_claims:
data.update(extra_claims)
return create_access_token(data, expires_delta=timedelta(minutes=expires_in))
class TestJWTValidation:
def test_valid_token_grants_access(self, client):
token = make_token()
response = client.get(
"/users/me",
headers={"Authorization": f"Bearer {token}"}
)
assert response.status_code == 200
def test_expired_token_returns_401(self, client):
# Create token that expired 1 second ago
token = make_token(expires_in=-1)
response = client.get(
"/users/me",
headers={"Authorization": f"Bearer {token}"}
)
assert response.status_code == 401
assert "credentials" in response.json()["detail"].lower()
def test_tampered_signature_returns_401(self, client):
token = make_token()
# Modify the signature (last segment)
parts = token.split(".")
parts[2] = parts[2][:-5] + "XXXXX"
tampered = ".".join(parts)
response = client.get(
"/users/me",
headers={"Authorization": f"Bearer {tampered}"}
)
assert response.status_code == 401
def test_wrong_secret_returns_401(self, client):
token = make_token(secret="completely-different-secret")
response = client.get(
"/users/me",
headers={"Authorization": f"Bearer {token}"}
)
assert response.status_code == 401
def test_none_algorithm_attack_rejected(self, client):
"""Verify the app rejects the 'none' algorithm JWT attack."""
# Attacker creates a token with algorithm=none and no signature
header = {"alg": "none", "typ": "JWT"}
payload = {"sub": "admin", "scopes": ["admin"]}
import base64, json
def b64encode(data):
return base64.urlsafe_b64encode(
json.dumps(data).encode()
).rstrip(b"=").decode()
token = f"{b64encode(header)}.{b64encode(payload)}."
response = client.get(
"/users/me",
headers={"Authorization": f"Bearer {token}"}
)
assert response.status_code == 401
def test_missing_sub_claim_returns_401(self, client):
# Token without 'sub' claim
token = jwt.encode(
{"exp": datetime.utcnow() + timedelta(hours=1), "role": "admin"},
SECRET_KEY,
algorithm=ALGORITHM
)
response = client.get(
"/users/me",
headers={"Authorization": f"Bearer {token}"}
)
assert response.status_code == 401
def test_no_authorization_header_returns_401(self, client):
response = client.get("/users/me")
assert response.status_code == 401
def test_malformed_bearer_prefix_returns_401(self, client):
token = make_token()
# Missing "Bearer " prefix
response = client.get(
"/users/me",
headers={"Authorization": token}
)
assert response.status_code == 401Testing Scopes and Permissions
OAuth2 scopes let you define fine-grained permissions. Test that scopes are enforced correctly — that a token with read:posts cannot access write:posts endpoints:
# app/auth.py (additions)
from fastapi.security import SecurityScopes
async def get_current_user_with_scopes(
security_scopes: SecurityScopes,
token: str = Depends(oauth2_scheme),
):
if security_scopes.scopes:
authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
else:
authenticate_value = "Bearer"
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username = payload.get("sub")
token_scopes = payload.get("scopes", [])
except JWTError:
raise HTTPException(status_code=401, detail="Invalid token")
for scope in security_scopes.scopes:
if scope not in token_scopes:
raise HTTPException(
status_code=403,
detail=f"Not enough permissions. Required: {scope}",
headers={"WWW-Authenticate": authenticate_value},
)
return {"username": username, "scopes": token_scopes}
# In your router:
from fastapi import Security
from app.auth import get_current_user_with_scopes
@router.get("/posts", dependencies=[Security(get_current_user_with_scopes, scopes=["read:posts"])])
async def list_posts():
return []
@router.post("/posts", dependencies=[Security(get_current_user_with_scopes, scopes=["write:posts"])])
async def create_post(post: PostCreate):
...
@router.delete("/posts/{post_id}", dependencies=[Security(get_current_user_with_scopes, scopes=["admin"])])
async def delete_post(post_id: int):
...# tests/test_scopes.py
import pytest
from fastapi.testclient import TestClient
from app.main import app
@pytest.fixture
def client():
return TestClient(app)
def make_scoped_token(scopes: list[str]) -> str:
from app.auth import create_access_token
from datetime import timedelta
return create_access_token(
data={"sub": "testuser", "scopes": scopes},
expires_delta=timedelta(minutes=30)
)
class TestScopeEnforcement:
def test_read_scope_allows_list_posts(self, client):
token = make_scoped_token(["read:posts"])
response = client.get(
"/posts",
headers={"Authorization": f"Bearer {token}"}
)
assert response.status_code == 200
def test_read_scope_blocks_create_post(self, client):
token = make_scoped_token(["read:posts"])
response = client.post(
"/posts",
json={"title": "Test", "content": "Body"},
headers={"Authorization": f"Bearer {token}"}
)
assert response.status_code == 403
assert "write:posts" in response.json()["detail"]
def test_write_scope_allows_create_post(self, client):
token = make_scoped_token(["read:posts", "write:posts"])
response = client.post(
"/posts",
json={"title": "Test", "content": "Body"},
headers={"Authorization": f"Bearer {token}"}
)
assert response.status_code == 201
def test_admin_scope_required_for_delete(self, client):
token = make_scoped_token(["read:posts", "write:posts"])
response = client.delete(
"/posts/1",
headers={"Authorization": f"Bearer {token}"}
)
assert response.status_code == 403
def test_admin_scope_allows_delete(self, client):
token = make_scoped_token(["admin"])
response = client.delete(
"/posts/1",
headers={"Authorization": f"Bearer {token}"}
)
# 200 or 404 depending on whether the post exists, but NOT 403
assert response.status_code != 403
def test_empty_scopes_blocked_from_all(self, client):
token = make_scoped_token([])
for method, path in [
("get", "/posts"),
("post", "/posts"),
("delete", "/posts/1"),
]:
response = getattr(client, method)(
path,
headers={"Authorization": f"Bearer {token}"},
json={}
)
assert response.status_code == 403, f"{method.upper()} {path} should be forbidden"Testing API Key Authentication
API key auth is common for service-to-service communication. Test key validation, key rotation, and per-key rate limits:
# app/auth.py (API key auth)
from fastapi import Security, HTTPException
from fastapi.security.api_key import APIKeyHeader
from app.repositories import ApiKeyRepository
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
async def get_api_key(
api_key: str = Security(api_key_header),
repo: ApiKeyRepository = Depends(get_api_key_repo),
):
if api_key is None:
raise HTTPException(status_code=401, detail="API key required")
key_record = await repo.get_by_key(api_key)
if key_record is None or not key_record.is_active:
raise HTTPException(status_code=401, detail="Invalid or inactive API key")
if key_record.is_expired:
raise HTTPException(status_code=401, detail="API key has expired")
await repo.record_usage(key_record.id)
return key_record# tests/test_api_key_auth.py
import pytest
from fastapi.testclient import TestClient
from unittest.mock import AsyncMock, MagicMock
from datetime import datetime, timedelta
from app.main import app
from app.dependencies import get_api_key_repo
def make_api_key_repo(key_record=None):
repo = MagicMock()
repo.get_by_key = AsyncMock(return_value=key_record)
repo.record_usage = AsyncMock(return_value=None)
return repo
def make_key_record(
key="valid-key-123",
is_active=True,
expires_at=None
):
record = MagicMock()
record.id = 1
record.key = key
record.is_active = is_active
record.is_expired = expires_at is not None and expires_at < datetime.utcnow()
record.scopes = ["read:data"]
return record
@pytest.fixture
def client():
return TestClient(app)
class TestAPIKeyAuth:
def test_valid_api_key_grants_access(self, client):
key_record = make_key_record()
repo = make_api_key_repo(key_record)
app.dependency_overrides[get_api_key_repo] = lambda: repo
response = client.get(
"/api/data",
headers={"X-API-Key": "valid-key-123"}
)
assert response.status_code == 200
app.dependency_overrides.clear()
def test_missing_api_key_returns_401(self, client):
repo = make_api_key_repo(None)
app.dependency_overrides[get_api_key_repo] = lambda: repo
response = client.get("/api/data") # No X-API-Key header
assert response.status_code == 401
assert "required" in response.json()["detail"]
app.dependency_overrides.clear()
def test_unknown_api_key_returns_401(self, client):
repo = make_api_key_repo(None) # key not found
app.dependency_overrides[get_api_key_repo] = lambda: repo
response = client.get(
"/api/data",
headers={"X-API-Key": "nonexistent-key"}
)
assert response.status_code == 401
app.dependency_overrides.clear()
def test_inactive_api_key_returns_401(self, client):
key_record = make_key_record(is_active=False)
repo = make_api_key_repo(key_record)
app.dependency_overrides[get_api_key_repo] = lambda: repo
response = client.get(
"/api/data",
headers={"X-API-Key": "inactive-key"}
)
assert response.status_code == 401
assert "inactive" in response.json()["detail"].lower()
app.dependency_overrides.clear()
def test_expired_api_key_returns_401(self, client):
key_record = make_key_record(
expires_at=datetime.utcnow() - timedelta(days=1)
)
repo = make_api_key_repo(key_record)
app.dependency_overrides[get_api_key_repo] = lambda: repo
response = client.get(
"/api/data",
headers={"X-API-Key": "expired-key"}
)
assert response.status_code == 401
assert "expired" in response.json()["detail"].lower()
app.dependency_overrides.clear()
def test_valid_key_records_usage(self, client):
key_record = make_key_record()
repo = make_api_key_repo(key_record)
app.dependency_overrides[get_api_key_repo] = lambda: repo
client.get("/api/data", headers={"X-API-Key": "valid-key-123"})
repo.record_usage.assert_called_once_with(1)
app.dependency_overrides.clear()Testing HTTPS-Only Cookies
Secure cookies — HttpOnly, Secure, SameSite — are essential for session-based auth. Test that your app sets them correctly and that it rejects requests without them:
# app/auth.py (cookie auth)
from fastapi import Response, Cookie, HTTPException
def set_auth_cookie(response: Response, token: str):
response.set_cookie(
key="session_token",
value=token,
httponly=True,
secure=True,
samesite="strict",
max_age=3600,
path="/",
)
async def get_session_user(
session_token: str | None = Cookie(default=None)
):
if session_token is None:
raise HTTPException(status_code=401, detail="Not authenticated")
try:
payload = jwt.decode(session_token, SECRET_KEY, algorithms=[ALGORITHM])
return {"username": payload["sub"]}
except JWTError:
raise HTTPException(status_code=401, detail="Invalid session")# tests/test_cookie_auth.py
import pytest
from fastapi.testclient import TestClient
from app.main import app
from app.auth import create_access_token
from datetime import timedelta
@pytest.fixture
def client():
# TestClient allows http by default, but cookies are still set
return TestClient(app, base_url="https://testserver")
def test_login_sets_secure_httponly_cookie(client):
response = client.post("/auth/login", data={
"username": "testuser",
"password": "password123"
})
assert response.status_code == 200
# Verify cookie attributes
set_cookie = response.headers.get("set-cookie")
assert set_cookie is not None
assert "httponly" in set_cookie.lower()
assert "samesite=strict" in set_cookie.lower()
assert "max-age=3600" in set_cookie.lower()
def test_authenticated_endpoint_works_with_cookie(client):
# Log in to get cookie
login_resp = client.post("/auth/login", data={
"username": "testuser",
"password": "password123"
})
assert login_resp.status_code == 200
# TestClient automatically carries cookies
response = client.get("/dashboard")
assert response.status_code == 200
def test_endpoint_requires_session_cookie(client):
# No cookie — should be rejected
response = client.get("/dashboard")
assert response.status_code == 401
def test_tampered_session_cookie_rejected(client):
token = create_access_token(
{"sub": "testuser"},
expires_delta=timedelta(hours=1)
)
# Tamper with the token
tampered = token[:-10] + "XXXXXXXXXX"
client.cookies.set("session_token", tampered)
response = client.get("/dashboard")
assert response.status_code == 401
def test_logout_clears_session_cookie(client):
# Log in
client.post("/auth/login", data={
"username": "testuser",
"password": "password123"
})
# Log out
response = client.post("/auth/logout")
assert response.status_code == 200
# Cookie should be cleared (max-age=0 or deleted)
set_cookie = response.headers.get("set-cookie", "")
assert "max-age=0" in set_cookie.lower() or "session_token=;" in set_cookie.lower()
# Subsequent requests should fail
response = client.get("/dashboard")
assert response.status_code == 401Role-Based Access Control Tests
RBAC is easy to implement but easy to test wrong. The critical tests are the negative cases — verifying that roles that should NOT have access are actually blocked:
# app/auth.py (RBAC)
from enum import Enum
from functools import wraps
class Role(str, Enum):
VIEWER = "viewer"
EDITOR = "editor"
ADMIN = "admin"
SUPERADMIN = "superadmin"
ROLE_HIERARCHY = {
Role.VIEWER: 0,
Role.EDITOR: 1,
Role.ADMIN: 2,
Role.SUPERADMIN: 3,
}
def require_role(minimum_role: Role):
async def dependency(current_user = Depends(get_current_user)):
user_role = Role(current_user.get("role", "viewer"))
if ROLE_HIERARCHY[user_role] < ROLE_HIERARCHY[minimum_role]:
raise HTTPException(
status_code=403,
detail=f"Role '{minimum_role}' required, you have '{user_role}'"
)
return current_user
return Depends(dependency)# tests/test_rbac.py
import pytest
from fastapi.testclient import TestClient
from app.main import app
from app.auth import get_current_user
def make_user(role: str, user_id: int = 1):
return {"id": user_id, "username": f"{role}_user", "role": role}
def client_as(role: str):
user = make_user(role)
app.dependency_overrides[get_current_user] = lambda: user
client = TestClient(app)
return client
@pytest.fixture(autouse=True)
def clear_overrides():
yield
app.dependency_overrides.clear()
class TestRBACEndpoints:
"""
Test matrix for each endpoint × each role.
Every cell in this matrix should be an explicit test.
"""
# POST /posts — requires EDITOR or higher
def test_viewer_cannot_create_post(self):
client = client_as("viewer")
response = client.post("/posts", json={"title": "T", "content": "C"})
assert response.status_code == 403
def test_editor_can_create_post(self):
client = client_as("editor")
response = client.post("/posts", json={"title": "T", "content": "C"})
assert response.status_code in (200, 201)
def test_admin_can_create_post(self):
client = client_as("admin")
response = client.post("/posts", json={"title": "T", "content": "C"})
assert response.status_code in (200, 201)
# DELETE /users/{id} — requires ADMIN or higher
def test_viewer_cannot_delete_user(self):
client = client_as("viewer")
response = client.delete("/users/99")
assert response.status_code == 403
def test_editor_cannot_delete_user(self):
client = client_as("editor")
response = client.delete("/users/99")
assert response.status_code == 403
def test_admin_can_delete_user(self):
client = client_as("admin")
response = client.delete("/users/99")
# 200 or 404, but NOT 403
assert response.status_code != 403
# GET /admin/settings — requires SUPERADMIN
def test_admin_cannot_access_superadmin_settings(self):
client = client_as("admin")
response = client.get("/admin/settings")
assert response.status_code == 403
def test_superadmin_can_access_settings(self):
client = client_as("superadmin")
response = client.get("/admin/settings")
assert response.status_code == 200
# Resource ownership: users can only edit their own resources
def test_user_cannot_edit_another_users_post(self):
# Create post owned by user 1
user1_client = client_as("editor")
create_resp = user1_client.post("/posts", json={"title": "User1 Post", "content": "C"})
post_id = create_resp.json()["id"]
# Try to edit as user 2
app.dependency_overrides[get_current_user] = lambda: make_user("editor", user_id=2)
user2_client = TestClient(app)
response = user2_client.patch(f"/posts/{post_id}", json={"title": "Hijacked"})
assert response.status_code == 403
def test_user_can_edit_own_post(self):
user_client = client_as("editor")
create_resp = user_client.post("/posts", json={"title": "My Post", "content": "C"})
post_id = create_resp.json()["id"]
# Same user edits it
response = user_client.patch(f"/posts/{post_id}", json={"title": "Updated Title"})
assert response.status_code == 200The ownership test is the one most teams skip. Without it, you won't catch horizontal privilege escalation — where a user with the right role but wrong identity can access another user's resources.
Testing Auth Error Response Format
Authentication errors should return consistent, machine-readable responses that your API clients can handle:
# tests/test_auth_error_format.py
import pytest
from fastapi.testclient import TestClient
from app.main import app
@pytest.fixture
def client():
return TestClient(app, raise_server_exceptions=False)
class TestAuthErrorResponses:
def test_401_includes_www_authenticate_header(self, client):
response = client.get("/users/me")
assert response.status_code == 401
assert "www-authenticate" in response.headers
def test_403_response_has_detail_field(self, client):
from app.auth import get_current_user
# Log in as viewer
app.dependency_overrides[get_current_user] = lambda: {
"id": 1, "role": "viewer"
}
response = client.delete("/admin/nuclear-option")
assert response.status_code == 403
assert "detail" in response.json()
app.dependency_overrides.clear()
def test_auth_errors_dont_leak_internal_details(self, client):
# Tampered token
response = client.get(
"/users/me",
headers={"Authorization": "Bearer tampered.token.value"}
)
body = response.json()["detail"].lower()
# Should not reveal implementation details
assert "traceback" not in body
assert "sqlalchemy" not in body
assert "internal" not in bodyEvery auth test you skip is a potential security vulnerability. The investment in a thorough auth test suite pays for itself the first time you refactor your middleware and don't break anything.