# Fastapi Security

> FastAPI security best practices — JWT authentication with python-jose, OAuth2, role-based access, CORS, rate limiting, security headers, and input validation.

- Skill: `aleonsa/fastapi-security` (Agent Skill)
- Install (CLI): `npx skillmds@latest add aleonsa/fastapi-security`
- Raw SKILL.md: https://api.skillmd.com/api/skills/aleonsa/fastapi-security/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Security
- Author: aleonsa (https://skillmd.com/u/aleonsa)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/aleonsa/fastapi-security

---


# FastAPI Security Best Practices

Comprehensive security patterns for FastAPI APIs using python-jose, OAuth2, and production hardening.

## When to Activate

- Setting up authentication and authorization in FastAPI
- Implementing JWT-based auth with python-jose
- Configuring CORS, rate limiting, security headers
- Reviewing FastAPI apps for security issues
- Deploying FastAPI to production

## JWT Authentication with python-jose

### Token Utilities

```python
# app/auth/tokens.py
from datetime import datetime, timedelta, timezone
from jose import JWTError, jwt
from app.config import settings


def create_access_token(subject: str | int, expires_delta: timedelta | None = None) -> str:
    expire = datetime.now(timezone.utc) + (
        expires_delta or timedelta(minutes=settings.access_token_expire_minutes)
    )
    payload = {"sub": str(subject), "exp": expire, "iat": datetime.now(timezone.utc)}
    return jwt.encode(payload, settings.secret_key, algorithm=settings.algorithm)


def decode_access_token(token: str) -> str:
    """Decode and validate JWT. Returns subject or raises JWTError."""
    payload = jwt.decode(token, settings.secret_key, algorithms=[settings.algorithm])
    sub: str | None = payload.get("sub")
    if sub is None:
        raise JWTError("Missing subject")
    return sub
```

### Password Hashing

```python
# app/auth/passwords.py
from passlib.context import CryptContext

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


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


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

### OAuth2 Dependencies

```python
# app/dependencies.py
from typing import Annotated
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError
from sqlalchemy.ext.asyncio import AsyncSession

from app.auth.tokens import decode_access_token
from app.models.user import User
from app.database import get_db

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


async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
    db: Annotated[AsyncSession, Depends(get_db)],
) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        user_id = decode_access_token(token)
    except JWTError:
        raise credentials_exception

    user = await db.get(User, int(user_id))
    if user is None:
        raise credentials_exception
    return user


async def get_current_active_user(
    current_user: Annotated[User, Depends(get_current_user)],
) -> User:
    if not current_user.is_active:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Inactive user")
    return current_user


# Reusable annotated types
CurrentUser = Annotated[User, Depends(get_current_active_user)]
```

### Auth Router

```python
# app/routers/auth.py
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.ext.asyncio import AsyncSession

from app.auth.passwords import verify_password
from app.auth.tokens import create_access_token
from app.dependencies import DbSession
from app.models.user import User
from app.schemas.auth import Token
from sqlalchemy import select

router = APIRouter(prefix="/auth", tags=["auth"])


@router.post("/token", response_model=Token)
async def login(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
    db: DbSession,
):
    result = await db.execute(select(User).where(User.email == form_data.username))
    user = result.scalar_one_or_none()

    if not user or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect email or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    if not user.is_active:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Inactive user")

    access_token = create_access_token(subject=user.id)
    return Token(access_token=access_token, token_type="bearer")


# app/schemas/auth.py
from pydantic import BaseModel

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

## Role-Based Access Control (RBAC)

```python
# app/auth/roles.py
from enum import StrEnum
from fastapi import HTTPException, status
from app.models.user import User


class Role(StrEnum):
    USER = "user"
    MODERATOR = "moderator"
    ADMIN = "admin"


def require_role(*roles: Role):
    """Dependency factory that requires one of the given roles."""
    async def check_role(current_user: CurrentUser) -> User:
        if current_user.role not in roles:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="Insufficient permissions",
            )
        return current_user
    return check_role


# Usage
AdminOnly = Annotated[User, Depends(require_role(Role.ADMIN))]
ModeratorOrAbove = Annotated[User, Depends(require_role(Role.MODERATOR, Role.ADMIN))]


@router.delete("/{user_id}", status_code=204)
async def delete_user(user_id: int, _: AdminOnly, db: DbSession):
    ...
```

## CORS Configuration

```python
# app/main.py
from fastapi.middleware.cors import CORSMiddleware
from app.config import settings

app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.cors_origins,   # Never use ["*"] in production
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE"],
    allow_headers=["Authorization", "Content-Type"],
    max_age=600,
)

# config.py — parse from env
cors_origins: list[str] = ["https://app.example.com"]
```

## Rate Limiting with slowapi

```python
# app/main.py
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)


# app/routers/auth.py
from app.main import limiter
from fastapi import Request

@router.post("/token")
@limiter.limit("10/minute")
async def login(request: Request, ...):
    ...

# Stricter limit for sensitive endpoints
@router.post("/register")
@limiter.limit("5/hour")
async def register(request: Request, ...):
    ...
```

## Security Headers Middleware

```python
# app/middleware/security_headers.py
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware


class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        response = await call_next(request)
        response.headers["X-Content-Type-Options"] = "nosniff"
        response.headers["X-Frame-Options"] = "DENY"
        response.headers["X-XSS-Protection"] = "1; mode=block"
        response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
        response.headers["Content-Security-Policy"] = "default-src 'self'"
        response.headers["Strict-Transport-Security"] = (
            "max-age=31536000; includeSubDomains; preload"
        )
        return response


# app/main.py
app.add_middleware(SecurityHeadersMiddleware)
```

## Input Validation

```python
# Pydantic v2 — use Field constraints, never trust user input
from pydantic import BaseModel, Field, EmailStr, field_validator
import re

class UserCreate(BaseModel):
    email: EmailStr
    password: str = Field(..., min_length=8, max_length=128)
    username: str = Field(..., min_length=3, max_length=50, pattern=r"^[a-zA-Z0-9_-]+$")

    @field_validator("password")
    @classmethod
    def password_strength(cls, v: str) -> str:
        if not re.search(r"[A-Z]", v):
            raise ValueError("Password must contain at least one uppercase letter")
        if not re.search(r"\d", v):
            raise ValueError("Password must contain at least one digit")
        return v
```

## File Upload Security

```python
import magic  # python-magic
from fastapi import UploadFile, HTTPException

ALLOWED_CONTENT_TYPES = {"image/jpeg", "image/png", "image/gif", "application/pdf"}
MAX_FILE_SIZE = 5 * 1024 * 1024  # 5 MB


async def validate_upload(file: UploadFile) -> bytes:
    content = await file.read()

    if len(content) > MAX_FILE_SIZE:
        raise HTTPException(status_code=413, detail="File too large (max 5MB)")

    # Validate actual content, not just the declared content-type
    detected = magic.from_buffer(content, mime=True)
    if detected not in ALLOWED_CONTENT_TYPES:
        raise HTTPException(status_code=415, detail=f"Unsupported file type: {detected}")

    return content
```

## Secrets Management

```python
# app/config.py — all secrets from environment, never hardcoded
from pydantic_settings import BaseSettings, SettingsConfigDict
from pydantic import SecretStr


class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env")

    # SecretStr prevents accidental logging
    secret_key: SecretStr
    database_url: SecretStr
    redis_url: SecretStr | None = None

    def get_secret_key(self) -> str:
        return self.secret_key.get_secret_value()
```

## Logging Security Events

```python
# app/middleware/audit_log.py
import logging
import time
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware

security_logger = logging.getLogger("security")


class AuditLogMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start = time.perf_counter()
        response = await call_next(request)
        duration = time.perf_counter() - start

        # Log auth failures and suspicious activity
        if response.status_code in (401, 403):
            security_logger.warning(
                "Auth failure",
                extra={
                    "method": request.method,
                    "path": request.url.path,
                    "status": response.status_code,
                    "ip": request.client.host if request.client else "unknown",
                    "duration_ms": round(duration * 1000, 2),
                },
            )
        return response
```

## Security Checklist

| Check | How |
|-------|-----|
| Never expose `DEBUG=True` in prod | `settings.debug = False` |
| JWT secret via env var | `SecretStr` in settings |
| Password hashing | `bcrypt` via `passlib` |
| HTTPS only | Enforce at reverse proxy (nginx/Caddy) |
| CORS locked down | Explicit origins, not `["*"]` |
| Rate limit auth endpoints | `slowapi` on `/token`, `/register` |
| Validate file uploads | Content-type via `python-magic`, size limit |
| Security headers | `SecurityHeadersMiddleware` |
| Input validation | Pydantic `Field` constraints + validators |
| No SQL injection | SQLAlchemy ORM or parameterized queries only |
| Log auth failures | `AuditLogMiddleware` |
| Dependency updates | `pip-audit` in CI |

Remember: No CSRF protection needed for stateless JWT APIs (no cookies carrying session), but always verify `Origin` for cookie-based auth.