FastAPI: Authorization & API security

Most API breaches are not clever cryptography attacks. Someone changes /orders/1042 to /orders/1043 and gets another customer’s order. Or a customer calls an admin endpoint nobody protected, or requests a million rows, or makes your server fetch http://169.254.169.254/. Authorization (who may do what to which object) is where APIs fail most, and the OWASP API Security Top 10 is the checklist. This part implements the defences that matter most, each running.

1. The OWASP API Security Top 10 (2023) → where this guide covers it

2. Object-level authorization (API1)

the vulnerable endpoint, and the fix: scope every query by the caller

from typing import Annotated

from fastapi import Depends, FastAPI, Header, HTTPException
from fastapi.testclient import TestClient

ORDERS = {1042: {"id": 1042, "customer_id": 7, "total": 1250},
          1043: {"id": 1043, "customer_id": 9, "total": 88_000, "address": "22 Palm Grove, Kochi"}}
TOKENS = {"t-asha": {"id": 7, "role": "customer"}, "t-ops": {"id": 1, "role": "support"}}

def current_user(x_token: Annotated[str, Header()]):
    if x_token not in TOKENS:
        raise HTTPException(401, "not authenticated")
    return TOKENS[x_token]

User = Annotated[dict, Depends(current_user)]
app = FastAPI()

@app.get("/v0/orders/{order_id}")                       # VULNERABLE: authenticated, but not authorized
def get_order_vulnerable(order_id: int, user: User):
    return ORDERS.get(order_id) or {}

def find_order_for(user: dict, order_id: int) -> dict:
    order = ORDERS.get(order_id)
    visible = order is not None and (order["customer_id"] == user["id"] or user["role"] == "support")
    if not visible:
        raise HTTPException(404, "order not found")     # 404, not 403: do not confirm it exists
    return order                                         # in SQL: WHERE id = :id AND customer_id = :user_id

@app.get("/v1/orders/{order_id}")
def get_order(order_id: int, user: User):
    return find_order_for(user, order_id)

client = TestClient(app)
asha = {"X-Token": "t-asha"}
print("v0, Asha reads order 1043:", client.get("/v0/orders/1043", headers=asha).json())   # leaks!
print("v1, Asha reads order 1043:", client.get("/v1/orders/1043", headers=asha).status_code)
print("v1, Asha reads her order: ", client.get("/v1/orders/1042", headers=asha).json()["total"])
print("v1, support reads 1043:   ", client.get("/v1/orders/1043", headers={"X-Token": "t-ops"}).status_code)

Output

v0, Asha reads order 1043: {'id': 1043, 'customer_id': 9, 'total': 88000, 'address': '22 Palm Grove, Kochi'}
v1, Asha reads order 1043: 404
v1, Asha reads her order:  1250
v1, support reads 1043:    200

3. Function-level authorization (API5): permissions, not role names

roles map to permissions; endpoints require permissions

from typing import Annotated

from fastapi import APIRouter, Depends, FastAPI, Header, HTTPException
from fastapi.testclient import TestClient

ROLE_PERMISSIONS = {
    "customer": {"orders:read:own", "orders:create"},
    "support":  {"orders:read:any", "refunds:create"},
    "admin":    {"orders:read:any", "refunds:create", "users:manage", "reports:export"},
}
USERS = {"t-asha": ("asha", "customer"), "t-meera": ("meera", "support"), "t-ravi": ("ravi", "admin")}

def current_user(x_token: Annotated[str, Header()]):
    name, role = USERS.get(x_token, (None, None))
    if name is None:
        raise HTTPException(401, "not authenticated")
    return {"name": name, "permissions": ROLE_PERMISSIONS[role]}

def require(permission: str):
    def check(user: Annotated[dict, Depends(current_user)]):
        if permission not in user["permissions"]:
            raise HTTPException(403, f"missing permission {permission}")
        return user
    return check

admin = APIRouter(prefix="/admin", dependencies=[Depends(require("users:manage"))])   # the whole area

@admin.get("/users")
def list_users():
    return ["asha", "meera", "ravi"]

app = FastAPI()
app.include_router(admin)

@app.post("/refunds")
def refund(user: Annotated[dict, Depends(require("refunds:create"))]):
    return {"approved_by": user["name"]}

client = TestClient(app)
for token in ("t-asha", "t-meera", "t-ravi"):
    h = {"X-Token": token}
    print(f"{token:8} refund: {client.post('/refunds', headers=h).status_code}  admin/users: {client.get('/admin/users', headers=h).status_code}")

Output

t-asha   refund: 403  admin/users: 403
t-meera  refund: 200  admin/users: 403
t-ravi   refund: 200  admin/users: 200

4. Field-level exposure (API3): different views of one object

customers and staff get different response models

from fastapi import FastAPI, Header
from fastapi.testclient import TestClient
from pydantic import BaseModel

class ProductPublic(BaseModel):
    sku: str
    name: str
    price: float

class ProductStaff(ProductPublic):
    cost_price: float
    supplier: str
    margin_pct: float

ROW = {"sku": "TEA-01", "name": "Assam Tea", "price": 240.0, "cost_price": 131.0, "supplier": "Hill Estates",
       "margin_pct": 45.4}

app = FastAPI()

@app.get("/products/{sku}", response_model=ProductStaff | ProductPublic)
def product(sku: str, x_role: str = Header("customer")):
    model = ProductStaff if x_role == "staff" else ProductPublic      # choose the view explicitly
    return model.model_validate(ROW)

client = TestClient(app)
print("customer:", client.get("/products/TEA-01").json())
print("staff:   ", client.get("/products/TEA-01", headers={"X-Role": "staff"}).json())

Output

customer: {'sku': 'TEA-01', 'name': 'Assam Tea', 'price': 240.0}
staff:    {'sku': 'TEA-01', 'name': 'Assam Tea', 'price': 240.0, 'cost_price': 131.0, 'supplier': 'Hill Estates', 'margin_pct': 45.4}

5. Rate limiting (API4, API2, API6)

a sliding-window limiter as a dependency, with Retry-After

import time
from collections import defaultdict, deque
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, Request, Response
from fastapi.testclient import TestClient

class SlidingWindowLimiter:
    """In-memory, per process. For a fleet, keep the counters in Redis (same algorithm)."""
    def __init__(self, limit: int, window_s: float):
        self.limit, self.window = limit, window_s
        self.hits: dict[str, deque] = defaultdict(deque)

    def __call__(self, request: Request, response: Response):
        key = f"{request.client.host}:{request.url.path}"       # use the user id when authenticated
        now = time.monotonic()
        q = self.hits[key]
        while q and now - q[0] > self.window:
            q.popleft()
        if len(q) >= self.limit:
            retry = max(1, int(self.window - (now - q[0])) + 1)
            raise HTTPException(429, "too many requests", headers={"Retry-After": str(retry)})
        q.append(now)
        response.headers["RateLimit-Limit"] = str(self.limit)
        response.headers["RateLimit-Remaining"] = str(self.limit - len(q))

login_limit = SlidingWindowLimiter(limit=5, window_s=60)      # strict on credential endpoints
app = FastAPI()

@app.post("/login", dependencies=[Depends(login_limit)])
def login():
    return {"ok": False}

client = TestClient(app)
for attempt in range(1, 8):
    r = client.post("/login")
    print(attempt, r.status_code, r.headers.get("ratelimit-remaining"), r.headers.get("retry-after"))

Output

1 200 4 None
2 200 3 None
3 200 2 None
4 200 1 None
5 200 0 None
6 429 None 60
7 429 None 60

6. SSRF: when users give you URLs (API7)

validate webhook URLs: https only, resolved addresses must be public

import ipaddress
from urllib.parse import urlsplit

FAKE_DNS = {"hooks.partner.example": ["93.184.215.14"],           # stand-in for socket.getaddrinfo
            "evil.example": ["10.0.0.5"],                          # resolves to an internal address
            "metadata.evil": ["169.254.169.254"]}

def resolve(host):
    try:
        return [str(ipaddress.ip_address(host))]                    # literal IPs resolve to themselves
    except ValueError:
        return FAKE_DNS.get(host, [])

def validate_webhook(url: str) -> str:
    parts = urlsplit(url)
    if parts.scheme != "https":
        raise ValueError("webhook must use https")
    if parts.port not in (None, 443):
        raise ValueError("only port 443 is allowed")
    addresses = resolve(parts.hostname or "")
    if not addresses:
        raise ValueError("host does not resolve")
    for addr in addresses:
        ip = ipaddress.ip_address(addr)
        if not ip.is_global:                                          # private, loopback, link-local, reserved
            raise ValueError(f"{parts.hostname} resolves to a non-public address ({ip})")
    return url

for url in ["https://hooks.partner.example/orders", "http://hooks.partner.example/x",
            "https://127.0.0.1/admin", "https://metadata.evil/latest/meta-data/", "https://evil.example/"]:
    try:
        print("allowed:", validate_webhook(url))
    except ValueError as e:
        print("blocked:", url, "→", e)

Output

allowed: https://hooks.partner.example/orders
blocked: http://hooks.partner.example/x → webhook must use https
blocked: https://127.0.0.1/admin → 127.0.0.1 resolves to a non-public address (127.0.0.1)
blocked: https://metadata.evil/latest/meta-data/ → metadata.evil resolves to a non-public address (169.254.169.254)
blocked: https://evil.example/ → evil.example resolves to a non-public address (10.0.0.5)

Resolve and check at send time too, and pin the connection to the address you checked. Otherwise “DNS rebinding” (a hostname that resolves to a public IP when validated and to a private one when fetched) bypasses the check. Disable redirects on outbound calls, or re-validate each hop. Better still, send webhooks from an isolated egress proxy that cannot reach internal networks at all.

7. Hardening checklist

Recap