FastAPI Authentication and Authorization Testing

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 == 422

JWT 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 == 401

Testing 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 == 401

Role-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 == 200

The 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 body

Every 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.

Read more

Start now free