OWASP API Security Top 10: A Practical Guide to Securing Your APIs

March 26, 2026 21 min read Security

securityapiowasprestgraphqlauthenticationauthorization

Contents

OWASP API Security Top 10: A Practical Guide to Securing Your APIs

APIs are the backbone of modern software. They power mobile apps, enable third-party integrations, and connect microservices. They’re also the primary attack surface for modern applications — the Verizon Data Breach Investigations Report consistently shows that web application attacks (APIs included) are the leading cause of breaches.

The OWASP API Security Top 10 was created specifically because traditional web application vulnerabilities don’t fully capture what goes wrong with APIs. An API doesn’t render HTML, so XSS is less relevant. But an API that exposes every field of your user table because a developer returned user.* instead of specific fields? That happens constantly.

This guide covers each vulnerability class with real attack scenarios, detection approaches, and concrete defensive code. By the end, you’ll be able to review an API with security in mind and avoid the most costly mistakes.

API1:2023 — Broken Object Level Authorization (BOLA)

Also known as Insecure Direct Object Reference (IDOR), BOLA is the #1 API vulnerability for good reason: it’s easy to introduce, hard to catch in code review, and devastating when exploited.

What it is

An endpoint that accepts an object ID from the caller but doesn’t verify the caller actually owns or has access to that object.

1
2


GET /api/orders/12345
Authorization: Bearer <token for user A>

User A can change 12345 to 67890 and read user B’s order — because the API trusts the ID without checking ownership.

Real-world example

In 2019, T-Mobile’s API exposed customer data because their account management API accepted account IDs without verifying the authenticated user owned that account. Attackers enumerated sequential IDs and exfiltrated millions of records.

Vulnerable code

1
2
3
4
5
6
7


# FastAPI — VULNERABLE
@app.get("/api/orders/{order_id}")
async def get_order(order_id: int, current_user: User = Depends(get_current_user)):
    order = await db.get_order(order_id)  # No ownership check!
    if not order:
        raise HTTPException(404)
    return order

Fixed code

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11


# FastAPI — SECURE
@app.get("/api/orders/{order_id}")
async def get_order(order_id: int, current_user: User = Depends(get_current_user)):
    order = await db.get_order(order_id)
    if not order:
        raise HTTPException(404)
    # Verify ownership — this is the critical check
    if order.user_id != current_user.id:
        # Return 404, not 403 — don't confirm the resource exists to unauthorized users
        raise HTTPException(404)
    return order

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18


// Go — SECURE with a query that enforces ownership at the DB level
func (h *Handler) GetOrder(w http.ResponseWriter, r *http.Request) {
    orderID := chi.URLParam(r, "orderID")
    userID := auth.UserIDFromContext(r.Context())

    // The WHERE clause enforces ownership — the DB won't return rows belonging to other users
    order, err := h.db.QueryRowContext(r.Context(),
        `SELECT id, items, total, status FROM orders
         WHERE id = $1 AND user_id = $2`,  // user_id constraint is the key
        orderID, userID,
    ).Scan(&order.ID, &order.Items, &order.Total, &order.Status)

    if err == sql.ErrNoRows {
        http.Error(w, "not found", http.StatusNotFound)
        return
    }
    // ...
}

Testing for BOLA

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11


# Create two accounts and get their tokens
TOKEN_A="eyJ..."
TOKEN_B="eyJ..."

# Get user B's resource ID (from their own authenticated session)
curl -H "Authorization: Bearer $TOKEN_B" https://api.example.com/api/orders | jq '.[0].id'
# Returns: 67890

# Try to access user B's resource with user A's token
curl -H "Authorization: Bearer $TOKEN_A" https://api.example.com/api/orders/67890
# Should return 404, not 200

Prevention

Always filter by the authenticated user’s ID at the query level, not in application code after fetching
Use non-sequential, non-guessable IDs (UUIDs v4) to raise the bar for enumeration
Write automated tests that cross access objects between test accounts and assert 404

API2:2023 — Broken Authentication

Authentication vulnerabilities in APIs range from weak credential policies to fundamentally flawed token implementations.

Common patterns

Weak JWT implementation:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15


# VULNERABLE — accepts "none" algorithm (JWT confusion attack)
import jwt
decoded = jwt.decode(token, options={"verify_signature": False})

# VULNERABLE — doesn't validate expiry
decoded = jwt.decode(token, SECRET, algorithms=["HS256"],
                     options={"verify_exp": False})

# SECURE
decoded = jwt.decode(
    token,
    SECRET,
    algorithms=["HS256"],  # Explicitly allowlist algorithms
    options={"require": ["exp", "iat", "sub"]},
)

Missing brute-force protection:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31


# VULNERABLE — no rate limiting on login
@app.post("/api/auth/login")
async def login(credentials: LoginRequest):
    user = await db.get_user_by_email(credentials.email)
    if user and verify_password(credentials.password, user.password_hash):
        return {"token": create_token(user)}
    raise HTTPException(401, "Invalid credentials")

# SECURE — rate limited with exponential backoff tracking
from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)

@app.post("/api/auth/login")
@limiter.limit("5/minute")  # 5 attempts per minute per IP
async def login(credentials: LoginRequest, request: Request):
    # Also track per-email to defend against distributed attacks
    attempts = await redis.incr(f"login_attempts:{credentials.email}")
    await redis.expire(f"login_attempts:{credentials.email}", 900)  # 15 minutes

    if attempts > 10:
        raise HTTPException(429, "Too many login attempts. Try again in 15 minutes.")

    user = await db.get_user_by_email(credentials.email)
    # Constant-time comparison to prevent timing attacks
    if not user or not verify_password(credentials.password, user.password_hash):
        raise HTTPException(401, "Invalid credentials")  # Same message for both cases

    await redis.delete(f"login_attempts:{credentials.email}")
    return {"token": create_token(user)}

Token storage and transmission:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22


# JWT creation best practices
import jwt
from datetime import datetime, timedelta, timezone
import secrets

def create_access_token(user_id: str) -> str:
    now = datetime.now(timezone.utc)
    payload = {
        "sub": user_id,
        "iat": now,
        "exp": now + timedelta(minutes=15),  # Short-lived
        "jti": secrets.token_hex(16),  # JWT ID for revocation
        "type": "access",
    }
    return jwt.encode(payload, ACCESS_SECRET, algorithm="RS256")  # Asymmetric > symmetric

def create_refresh_token(user_id: str) -> str:
    token = secrets.token_urlsafe(32)
    # Store in DB with expiry — can be revoked
    db.store_refresh_token(user_id=user_id, token_hash=hash(token),
                           expires_at=datetime.now() + timedelta(days=30))
    return token

API3:2023 — Broken Object Property Level Authorization

A subtler variant of BOLA: the attacker can access the right object but sees or modifies properties they shouldn’t.

Excessive data exposure

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23


# VULNERABLE — returns the full ORM object
@app.get("/api/users/{user_id}/profile")
async def get_profile(user_id: int):
    user = await db.get_user(user_id)
    return user  # Exposes: password_hash, internal_notes, stripe_customer_id, admin_flags, ...

# SECURE — explicit response model
from pydantic import BaseModel

class PublicProfile(BaseModel):
    id: int
    name: str
    bio: str
    avatar_url: str
    created_at: datetime

    class Config:
        from_attributes = True

@app.get("/api/users/{user_id}/profile", response_model=PublicProfile)
async def get_profile(user_id: int):
    user = await db.get_user(user_id)
    return user  # Pydantic serializes only the PublicProfile fields

Mass assignment

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21


# VULNERABLE — lets callers set any field including privileged ones
@app.put("/api/users/me")
async def update_profile(updates: dict, current_user: User = Depends(get_current_user)):
    await db.update_user(current_user.id, **updates)
    # Attacker sends: {"is_admin": true, "subscription_tier": "enterprise"}

# SECURE — allowlist exactly what users can update
class UpdateProfileRequest(BaseModel):
    name: Optional[str] = None
    bio: Optional[str] = None
    avatar_url: Optional[str] = None
    # is_admin: NOT here
    # subscription_tier: NOT here

@app.put("/api/users/me")
async def update_profile(
    updates: UpdateProfileRequest,
    current_user: User = Depends(get_current_user)
):
    # Only the allowlisted fields can be changed
    await db.update_user(current_user.id, **updates.model_dump(exclude_none=True))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17


// Go — never bind directly to your DB model
type UpdateUserRequest struct {
    Name      *string `json:"name"`
    Bio       *string `json:"bio"`
    AvatarURL *string `json:"avatar_url"`
    // IsAdmin is deliberately absent — it cannot be set via this endpoint
}

func (h *Handler) UpdateProfile(w http.ResponseWriter, r *http.Request) {
    var req UpdateUserRequest
    if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
        http.Error(w, "invalid request", http.StatusBadRequest)
        return
    }
    userID := auth.UserIDFromContext(r.Context())
    h.db.UpdateUser(r.Context(), userID, req) // Only UpdateUserRequest fields
}

API4:2023 — Unrestricted Resource Consumption

APIs that don’t enforce limits on query complexity, response size, or request rate can be exhausted — either maliciously or accidentally.

Request rate limiting (token bucket)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11


# Nginx — rate limiting at the ingress level
limit_req_zone $binary_remote_addr zone=api:10m rate=100r/m;
limit_req_zone $http_authorization zone=api_user:10m rate=1000r/m;

server {
    location /api/ {
        limit_req zone=api burst=20 nodelay;
        limit_req zone=api_user burst=100 nodelay;
        limit_req_status 429;
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26


# Application-level limits with Redis sliding window
import time

async def check_rate_limit(key: str, limit: int, window_seconds: int) -> bool:
    now = time.time()
    pipeline = redis.pipeline()
    pipeline.zremrangebyscore(key, 0, now - window_seconds)  # Remove old entries
    pipeline.zadd(key, {str(now): now})                       # Add current request
    pipeline.zcard(key)                                        # Count requests in window
    pipeline.expire(key, window_seconds)
    results = await pipeline.execute()
    count = results[2]
    return count <= limit

@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    user_id = get_user_id_from_token(request)
    key = f"rate_limit:{user_id or request.client.host}"

    if not await check_rate_limit(key, limit=100, window_seconds=60):
        return JSONResponse(
            status_code=429,
            content={"error": "Rate limit exceeded"},
            headers={"Retry-After": "60", "X-RateLimit-Limit": "100"},
        )
    return await call_next(request)

GraphQL query depth limiting

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17


// GraphQL — prevent deeply nested queries (N+1 amplification)
const { createComplexityLimitRule } = require('graphql-validation-complexity');
const depthLimit = require('graphql-depth-limit');

const schema = makeExecutableSchema({ typeDefs, resolvers });

app.use('/graphql', graphqlHTTP({
  schema,
  validationRules: [
    depthLimit(7),                              // Max 7 levels of nesting
    createComplexityLimitRule(1000, {           // Max complexity score of 1000
      scalarCost: 1,
      objectCost: 10,
      listFactor: 10,
    }),
  ],
}));

Payload size limits

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14


# FastAPI — enforce request body size limits
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware

class MaxBodySizeMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, max_upload_size: int = 10 * 1024 * 1024):  # 10 MB
        super().__init__(app)
        self.max_upload_size = max_upload_size

    async def dispatch(self, request: Request, call_next):
        content_length = request.headers.get("content-length")
        if content_length and int(content_length) > self.max_upload_size:
            return JSONResponse(status_code=413, content={"error": "Payload too large"})
        return await call_next(request)

API5:2023 — Broken Function Level Authorization (BFLA)

Where BOLA is about accessing the wrong object, BFLA is about accessing the wrong function — typically admin or privileged operations.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24


# VULNERABLE — admin endpoint without role check
@app.delete("/api/users/{user_id}")
async def delete_user(user_id: int, current_user: User = Depends(get_current_user)):
    await db.delete_user(user_id)  # Any authenticated user can delete any user!
    return {"status": "deleted"}

# SECURE — explicit role requirement
from functools import wraps

def require_role(*roles: str):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, current_user: User = Depends(get_current_user), **kwargs):
            if current_user.role not in roles:
                raise HTTPException(403, "Insufficient permissions")
            return await func(*args, current_user=current_user, **kwargs)
        return wrapper
    return decorator

@app.delete("/api/users/{user_id}")
@require_role("admin", "super_admin")
async def delete_user(user_id: int, current_user: User = Depends(get_current_user)):
    await db.delete_user(user_id)
    return {"status": "deleted"}

HTTP method confusion

1
2
3
4
5
6
7


# Secure routing — be explicit about allowed methods
@app.api_route("/api/orders/{order_id}", methods=["GET"])
async def get_order(order_id: int): ...

# Don't use @app.route without methods= — it matches ALL verbs by default
# Flask example of the trap:
# @app.route("/admin/users")  # Matches GET, POST, DELETE, etc.

Hidden administrative endpoints

A common BFLA pattern is endpoints that “shouldn’t be accessible” but are exposed because the API framework routes them:

1
2
3
4
5
6
7


# Attackers probe for common admin patterns
/api/admin/users
/api/internal/metrics
/api/v1/admin
/api/debug/config
/_admin
/management/env  # Spring Boot Actuator

Mitigation: bind administrative endpoints to a separate port or network interface that isn’t publicly exposed:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11


# Separate admin app on internal port
admin_app = FastAPI()

@admin_app.get("/users")
async def list_all_users(): ...

if __name__ == "__main__":
    import uvicorn
    # Public API on 0.0.0.0:8000 (internet-facing)
    # Admin API on 127.0.0.1:8001 (localhost only — accessed via kubectl port-forward)
    uvicorn.run("main:admin_app", host="127.0.0.1", port=8001)

API6:2023 — Unrestricted Access to Sensitive Business Flows

APIs that expose business logic without rate limiting or flow control can be abused for:

Scalping: bulk-buying limited-inventory items faster than humans
Account enumeration: checking if emails exist via “forgot password” errors
Coupon abuse: applying the same discount code thousands of times in parallel

Inventory scalping defense

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29


# VULNERABLE — no purchase limits
@app.post("/api/checkout")
async def checkout(items: List[CartItem], current_user: User = Depends(...)):
    return await process_purchase(items, current_user)

# SECURE — per-user rate limiting and purchase caps for limited items
@app.post("/api/checkout")
async def checkout(items: List[CartItem], current_user: User = Depends(...)):
    for item in items:
        product = await db.get_product(item.product_id)

        if product.is_limited_release:
            # Check per-user purchase cap
            user_purchases = await db.count_purchases(
                user_id=current_user.id,
                product_id=item.product_id,
                window_hours=24
            )
            if user_purchases + item.quantity > product.per_user_limit:
                raise HTTPException(429, f"Purchase limit: {product.per_user_limit} per customer per day")

            # Distributed lock to prevent race conditions
            async with redis_lock(f"product:{item.product_id}:checkout", timeout=5):
                stock = await db.get_stock(item.product_id)
                if stock < item.quantity:
                    raise HTTPException(409, "Insufficient stock")
                await db.reserve_stock(item.product_id, item.quantity)

    return await process_purchase(items, current_user)

Account enumeration via error messages

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18


# VULNERABLE — leaks whether an email exists
@app.post("/api/auth/forgot-password")
async def forgot_password(email: str):
    user = await db.get_user_by_email(email)
    if not user:
        raise HTTPException(404, "No account found with this email")  # LEAKS!
    await send_reset_email(user)
    return {"message": "Reset email sent"}

# SECURE — always return the same response
@app.post("/api/auth/forgot-password")
@limiter.limit("3/hour")  # Rate limit tightly
async def forgot_password(email: str, request: Request):
    user = await db.get_user_by_email(email)
    if user:  # Only send if user exists, but don't leak the difference
        await send_reset_email(user)
    # Always return the same message regardless of whether user exists
    return {"message": "If an account exists with this email, you'll receive a reset link."}

API7:2023 — Server-Side Request Forgery (SSRF)

When an API fetches a URL on behalf of the caller, an attacker can point it at internal services — cloud metadata endpoints, internal databases, Kubernetes API server, etc.

Attack example

1
2
3
4


POST /api/webhooks/test
Content-Type: application/json

{"url": "http://169.254.169.254/latest/meta-data/iam/security-credentials/"}

If the server makes this request, the attacker gets the EC2 instance’s IAM credentials.

Vulnerable code

1
2
3
4
5
6
7
8


# VULNERABLE
import httpx

@app.post("/api/webhooks/test")
async def test_webhook(payload: WebhookTestRequest, current_user: User = Depends(...)):
    async with httpx.AsyncClient() as client:
        response = await client.post(payload.url, json={"test": True})  # SSRF!
    return {"status": response.status_code}

Fixed code

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55


import ipaddress
import socket
from urllib.parse import urlparse

BLOCKED_NETWORKS = [
    ipaddress.ip_network("10.0.0.0/8"),
    ipaddress.ip_network("172.16.0.0/12"),
    ipaddress.ip_network("192.168.0.0/16"),
    ipaddress.ip_network("127.0.0.0/8"),
    ipaddress.ip_network("169.254.0.0/16"),  # AWS/GCP metadata
    ipaddress.ip_network("100.64.0.0/10"),   # Tailscale/CGNAT
    ipaddress.ip_network("::1/128"),
    ipaddress.ip_network("fc00::/7"),
]

ALLOWED_SCHEMES = {"https"}
ALLOWED_PORTS = {443}

def validate_webhook_url(url: str) -> None:
    parsed = urlparse(url)

    if parsed.scheme not in ALLOWED_SCHEMES:
        raise ValueError(f"Only HTTPS webhooks are allowed")

    if parsed.port and parsed.port not in ALLOWED_PORTS:
        raise ValueError(f"Port {parsed.port} is not allowed")

    hostname = parsed.hostname
    if not hostname:
        raise ValueError("Invalid URL")

    # Resolve DNS — check final IP, not just hostname
    try:
        ip = ipaddress.ip_address(socket.gethostbyname(hostname))
    except (socket.gaierror, ValueError) as e:
        raise ValueError(f"Cannot resolve hostname: {e}")

    for blocked in BLOCKED_NETWORKS:
        if ip in blocked:
            raise ValueError(f"Private/internal IP addresses are not allowed")

@app.post("/api/webhooks/test")
async def test_webhook(payload: WebhookTestRequest, current_user: User = Depends(...)):
    try:
        validate_webhook_url(payload.url)
    except ValueError as e:
        raise HTTPException(400, str(e))

    async with httpx.AsyncClient(
        follow_redirects=False,   # Don't follow redirects — could bypass IP checks
        timeout=5.0,
    ) as client:
        response = await client.post(payload.url, json={"test": True})

    return {"status": response.status_code}

Additional SSRF mitigations

Run the webhook-fetching service in a network segment with no access to internal services
Use a dedicated egress proxy that enforces allowlists at the network layer
Disable automatic DNS rebinding protection at the OS level (use a dedicated resolver)

API8:2023 — Security Misconfiguration

Catch-all for the many ways APIs expose themselves through poor configuration.

CORS misconfiguration

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26


# VULNERABLE — reflects any origin
from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],           # Allows any origin
    allow_credentials=True,         # With credentials — this is especially dangerous
    allow_methods=["*"],
    allow_headers=["*"],
)
# Note: browsers won't send cookies to allow_origins=["*"] with credentials=True,
# but some custom clients will — and this opens reflected-origin attacks

# SECURE — explicit allowlist
ALLOWED_ORIGINS = [
    "https://app.example.com",
    "https://admin.example.com",
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=ALLOWED_ORIGINS,
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE", "PATCH"],
    allow_headers=["Authorization", "Content-Type"],
)

Debug mode and stack traces in production

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24


# VULNERABLE — exposes stack traces to callers
@app.exception_handler(Exception)
async def generic_exception_handler(request: Request, exc: Exception):
    import traceback
    return JSONResponse(
        status_code=500,
        content={"error": traceback.format_exc()}  # Full stack trace to attacker!
    )

# SECURE — log internally, return generic message
import logging
logger = logging.getLogger(__name__)

@app.exception_handler(Exception)
async def generic_exception_handler(request: Request, exc: Exception):
    # Log with full context for internal debugging
    logger.exception("Unhandled exception", extra={
        "path": request.url.path,
        "method": request.method,
    })
    return JSONResponse(
        status_code=500,
        content={"error": "Internal server error", "request_id": request.state.request_id}
    )

Security headers for APIs

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14


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["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
        response.headers["Cache-Control"] = "no-store"  # Don't cache API responses
        response.headers["Content-Security-Policy"] = "default-src 'none'"
        # Remove server banner
        response.headers.pop("server", None)
        response.headers.pop("x-powered-by", None)
        return response

Disabling unnecessary HTTP methods

1
2
3
4
5
6


# Ensure OPTIONS/TRACE/etc. are handled explicitly
@app.middleware("http")
async def method_allowlist(request: Request, call_next):
    if request.method not in {"GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"}:
        return JSONResponse(status_code=405, content={"error": "Method not allowed"})
    return await call_next(request)

API9:2023 — Improper Inventory Management

Shadow APIs — undocumented, forgotten, or old API versions — are a persistent problem. They often lack the security controls added to newer versions.

Version sprawl example

Production:
  /api/v1/users/{id}    ← still running, no MFA required, no rate limits
  /api/v2/users/{id}    ← current, MFA enforced
  /api/v3/users/{id}    ← new, full security stack

Legacy:
  /api/users/{id}       ← original unversioned endpoint, never decommissioned
  /internal/admin/...   ← internal debug endpoint exposed publicly

Mitigation: API inventory and lifecycle policy

Maintain a central API catalog (e.g., in Backstage or a simple OpenAPI registry):

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13


# api-catalog.yaml — track every version
apis:
  - name: users-api
    versions:
      - version: v1
        status: deprecated
        sunset_date: 2026-06-01
        endpoints: /api/v1/users
        security_review: false  # Flag for attention
      - version: v2
        status: current
        endpoints: /api/v2/users
        security_review: true

1
2
3
4
5
6
7
8
9


# Add deprecation headers to old versions
@app.middleware("http")
async def deprecation_warning(request: Request, call_next):
    response = await call_next(request)
    if request.url.path.startswith("/api/v1/"):
        response.headers["Deprecation"] = "true"
        response.headers["Sunset"] = "Sat, 01 Jun 2026 00:00:00 GMT"
        response.headers["Link"] = '</api/v2/>; rel="successor-version"'
    return response

Automate discovery to find undocumented endpoints:

1
2
3
4
5


# Use OpenAPI diff tooling to find endpoints not in your spec
npx @stoplight/spectral-cli lint openapi.yaml

# Use tracing data to find endpoints that receive traffic but aren't in your spec
# (Tempo/Jaeger query for span.http.route values not in your OpenAPI paths)

API10:2023 — Unsafe Consumption of APIs

When your API consumes third-party APIs, you inherit their security posture. Blindly trusting upstream data is a vulnerability.

Third-party data injection

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48


# VULNERABLE — trusts external API response
import httpx

@app.get("/api/products/{product_id}/reviews")
async def get_reviews(product_id: str):
    async with httpx.AsyncClient() as client:
        # Third-party review aggregator
        resp = await client.get(f"https://reviews-api.example.com/products/{product_id}")
    data = resp.json()
    return data  # Could contain XSS payloads, SQL injection, malformed data...

# SECURE — validate and sanitize upstream data
from pydantic import BaseModel, validator
import bleach

class ExternalReview(BaseModel):
    rating: int
    comment: str
    reviewer_name: str

    @validator("rating")
    def rating_range(cls, v):
        if not 1 <= v <= 5:
            raise ValueError("Rating must be 1-5")
        return v

    @validator("comment", "reviewer_name")
    def sanitize_html(cls, v):
        # Strip all HTML from user-generated content from external sources
        return bleach.clean(v, tags=[], strip=True)[:1000]  # Also truncate

@app.get("/api/products/{product_id}/reviews")
async def get_reviews(product_id: str):
    async with httpx.AsyncClient(timeout=5.0) as client:
        try:
            resp = await client.get(
                f"https://reviews-api.example.com/products/{product_id}",
                headers={"Authorization": f"Bearer {REVIEWS_API_KEY}"},
            )
            resp.raise_for_status()
        except httpx.HTTPError as e:
            # Fail gracefully — don't let upstream failures crash your API
            logger.warning(f"Reviews API error: {e}")
            return []

    raw_reviews = resp.json()
    # Validate each review through the Pydantic model
    return [ExternalReview(**r).model_dump() for r in raw_reviews if isinstance(r, dict)]

Webhook signature verification

When accepting webhooks from third-party services, always verify the signature:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38


import hmac
import hashlib

STRIPE_WEBHOOK_SECRET = os.environ["STRIPE_WEBHOOK_SECRET"]

@app.post("/webhooks/stripe")
async def stripe_webhook(request: Request):
    payload = await request.body()
    sig_header = request.headers.get("stripe-signature")

    if not sig_header:
        raise HTTPException(400, "Missing signature")

    # Extract timestamp and signatures
    parts = {k: v for k, v in (x.split("=", 1) for x in sig_header.split(","))}
    timestamp = int(parts.get("t", 0))
    signatures = [v for k, v in parts.items() if k == "v1"]

    # Reject old webhooks to prevent replay attacks
    if abs(time.time() - timestamp) > 300:  # 5 minute tolerance
        raise HTTPException(400, "Webhook timestamp too old")

    # Compute expected signature
    signed_payload = f"{timestamp}.{payload.decode()}"
    expected_sig = hmac.new(
        STRIPE_WEBHOOK_SECRET.encode(),
        signed_payload.encode(),
        hashlib.sha256
    ).hexdigest()

    # Constant-time comparison
    if not any(hmac.compare_digest(expected_sig, sig) for sig in signatures):
        raise HTTPException(400, "Invalid signature")

    # Now safe to process
    event = json.loads(payload)
    await handle_stripe_event(event)
    return {"status": "ok"}

Building a Defense-in-Depth API Security Stack

No single control is sufficient. Layer these defenses:

Internet → WAF → API Gateway → Service Mesh (mTLS) → Application

Layer 1: WAF and DDoS protection

Use Cloudflare, AWS WAF, or similar at the edge:

Block known malicious IPs and botnets
Rate limiting by IP and user-agent
Block OWASP common attack patterns

Layer 2: API Gateway

Centralize cross-cutting security concerns:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17


# Kong API Gateway — rate limiting and auth plugin
plugins:
  - name: rate-limiting
    config:
      minute: 100
      policy: redis
      redis_host: redis

  - name: jwt
    config:
      key_claim_name: kid
      claims_to_verify:
        - exp

  - name: request-size-limiting
    config:
      allowed_payload_size: 10  # MB

Layer 3: Application-level

The defensive patterns shown throughout this guide:

BOLA checks (ownership at query level)
Input validation (Pydantic/validation libraries)
Explicit response schemas (no SELECT *)

Layer 4: Observability

Security without visibility is blind. Log and alert on:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13


# Structured security event logging
def log_security_event(event_type: str, request: Request, **details):
    logger.warning("security_event", extra={
        "event_type": event_type,
        "ip": request.client.host,
        "user_agent": request.headers.get("user-agent"),
        "path": str(request.url.path),
        "user_id": getattr(request.state, "user_id", None),
        **details,
    })

# Alert on: repeated 401s from same IP, BOLA attempts (404 after auth),
#           mass data exports, impossible geography logins

API Security Testing Checklist

Use this when reviewing or testing an API:

Authentication & Authorization

Can I access other users’ resources by changing IDs? (BOLA)
Can I call admin endpoints as a regular user? (BFLA)
Can I set privileged fields like is_admin? (mass assignment)
Does the API return fields I shouldn’t see? (excessive data exposure)
Can I brute-force login without getting rate limited?
Do JWT tokens have short expiry and proper algorithm validation?

Input Handling

Can I submit massive payloads without hitting size limits?
Can deeply nested GraphQL queries cause performance issues?
Can I trigger SSRF by providing URLs to internal services?

Configuration

Does CORS reflect arbitrary origins?
Do error responses include stack traces or internal paths?
Are old API versions still accessible with fewer controls?
Is there a debug/admin endpoint exposed publicly?

Business Logic

Can I register the same email twice via race condition?
Does “forgot password” reveal whether an email exists?
Can I apply coupons or discounts multiple times?

Summary

API security failures follow predictable patterns. The OWASP API Security Top 10 is valuable not as a compliance checkbox but as a mental model for how attackers think:

BOLA: Every endpoint that takes an ID must verify ownership
Broken Auth: Short tokens, rate-limited login, asymmetric JWTs
Property-Level Auth: Explicit response schemas, allowlisted write fields
Resource Consumption: Rate limits, payload caps, query complexity limits
BFLA: Every endpoint needs explicit authorization, not just authentication
Business Flow Abuse: Throttle sensitive flows, same responses for existence checks
SSRF: Validate and resolve URLs, block private IP ranges
Misconfiguration: Explicit CORS origins, no stack traces, security headers
Inventory: Track and sunset old API versions, detect shadow APIs
Unsafe Consumption: Validate third-party data, verify webhook signatures

The most impactful investment: implement BOLA checks everywhere (it’s the most common, most exploitable), use explicit response models (prevents accidental data exposure), and add rate limiting at the gateway before any single team’s service gets compromised by brute force or resource exhaustion.

OWASP API Security Top 10: A Practical Guide to Securing Your APIs

API1:2023 — Broken Object Level Authorization (BOLA)

What it is

Real-world example

Vulnerable code

Fixed code

Testing for BOLA

Prevention

API2:2023 — Broken Authentication

Common patterns

API3:2023 — Broken Object Property Level Authorization

Excessive data exposure

Mass assignment

API4:2023 — Unrestricted Resource Consumption

Request rate limiting (token bucket)

GraphQL query depth limiting

Payload size limits

API5:2023 — Broken Function Level Authorization (BFLA)

HTTP method confusion

Hidden administrative endpoints

API6:2023 — Unrestricted Access to Sensitive Business Flows

Inventory scalping defense

Account enumeration via error messages

API7:2023 — Server-Side Request Forgery (SSRF)

Attack example

Vulnerable code

Fixed code

Additional SSRF mitigations

API8:2023 — Security Misconfiguration

CORS misconfiguration

Debug mode and stack traces in production

Security headers for APIs

Disabling unnecessary HTTP methods

API9:2023 — Improper Inventory Management

Version sprawl example

Mitigation: API inventory and lifecycle policy

API10:2023 — Unsafe Consumption of APIs

Third-party data injection

Webhook signature verification

Building a Defense-in-Depth API Security Stack

Layer 1: WAF and DDoS protection

Layer 2: API Gateway

Layer 3: Application-level

Layer 4: Observability

API Security Testing Checklist

Summary

Comments