Chapter 21: API Security — Guarding the Gates of Your Data

"An API is a contract. A poorly secured API is a contract with the attacker." — Troy Hunt

Imagine scrolling through a log of HTTP requests — thousands per minute, all hitting the same endpoint pattern. Someone is enumerating every user ID from 1 to 500,000 through your API. They have been at it for six hours. Why didn't rate limiting catch it? Because there is no rate limiting. And because the API returns full user profiles including email addresses and phone numbers for any valid ID. No authorization check — if you know the number, you get the data.

That is BOLA — Broken Object-Level Authorization. The number one vulnerability in the OWASP API Security Top 10. APIs are the attack surface of the modern web. If your web UI is the front door, your API is every window, vent, and service entrance combined. This chapter covers the OWASP API Security Top 10, authentication and validation patterns, and the architectural controls that prevent these vulnerabilities.

API Authentication Patterns Compared

Authentication answers the question: "Who are you?" For APIs, there are several mechanisms, each with fundamentally different security properties.

graph LR
    subgraph "Authentication Mechanisms"
        A[API Keys] --> |Simple but limited| D[Identifies App]
        B[OAuth 2.0 Tokens] --> |Flexible, scoped| E[Identifies User + Scopes]
        C[mTLS Certificates] --> |Transport-level| F[Identifies Service]
    end

    subgraph "Best For"
        D --> G[Server-to-server<br/>Rate limiting<br/>Billing]
        E --> H[User-facing APIs<br/>Third-party access<br/>Mobile/SPA apps]
        F --> I[Service mesh<br/>Internal APIs<br/>Zero Trust]
    end

API Keys

The simplest authentication mechanism — a long random string included in each request.

curl -H "X-API-Key: sk_live_4eC39HqLyjWDarjtT1zdp7dc" \
  https://api.example.com/v1/data

Best practices for API keys:

import secrets
import hashlib

# Generate with sufficient entropy (256 bits)
raw_key = secrets.token_urlsafe(32)
# Prefix for easy identification and scanning
api_key = f"sk_live_{raw_key}"
# Example: sk_live_4eC39HqLyjWDarjtT1zdp7dc_8f3kJm2Q

# Store HASHED in database (like a password)
key_hash = hashlib.sha256(api_key.encode()).hexdigest()

# Verification on each request
def verify_api_key(provided_key):
    provided_hash = hashlib.sha256(provided_key.encode()).hexdigest()
    # Constant-time comparison to prevent timing attacks
    return hmac.compare_digest(provided_hash, stored_hash)

Never put API keys in client-side code (JavaScript, mobile apps). Client-side code is fully inspectable — anyone can extract the key from a browser's developer tools or by decompiling an APK/IPA. Use API keys only for server-to-server communication. For client-to-server, use OAuth tokens with appropriate flows (Authorization Code + PKCE for SPAs and mobile apps).

OAuth 2.0 Tokens

OAuth 2.0 separates the concerns of identity, authorization, and resource access. The user authenticates with an identity provider, which issues time-limited tokens with specific scopes.

sequenceDiagram
    participant User
    participant SPA as Client App<br/>(SPA / Mobile)
    participant Auth as Auth Server<br/>(IdP)
    participant API as API Server

    User->>SPA: 1. Click "Login"
    SPA->>Auth: 2. Redirect to /authorize<br/>response_type=code<br/>code_challenge=SHA256(verifier)<br/>scope=read:profile write:orders

    User->>Auth: 3. Enter credentials + MFA
    Auth->>Auth: 4. Verify identity
    Auth-->>SPA: 5. Redirect with authorization code

    SPA->>Auth: 6. POST /token<br/>grant_type=authorization_code<br/>code=abc123<br/>code_verifier=original_verifier
    Auth-->>SPA: 7. Access token (15min) + Refresh token (7d)

    SPA->>API: 8. GET /api/profile<br/>Authorization: Bearer eyJhbG...
    API->>API: 9. Verify JWT signature + expiry + scopes
    API-->>SPA: 10. {"name": "User", "email": "..."}

    Note over SPA,Auth: When access token expires:
    SPA->>Auth: POST /token<br/>grant_type=refresh_token<br/>refresh_token=def456
    Auth-->>SPA: New access token (15min)

JWT (JSON Web Tokens) as access tokens — security pitfalls:

JWT vulnerabilities that have caused real breaches:

**1. `alg: none` attack:**
Some JWT libraries accepted tokens with `"alg": "none"`, meaning no signature verification. An attacker could forge any token by setting the algorithm to `none` and removing the signature:

eyJhbGciOiJub25lIn0.eyJzdWIiOiJhZG1pbiIsInJvbGUiOiJhZG1pbiJ9.

**Defense:** Always validate the algorithm server-side and reject `none`. Use an allowlist of accepted algorithms.

**2. Algorithm confusion (RSA → HMAC):**
If the server expects RSA-signed tokens (asymmetric: sign with private key, verify with public key) but the library also accepts HMAC (symmetric: same key for both), an attacker can:
- Download the server's public RSA key
- Create a token signed with HMAC using the public key as the HMAC secret
- The server uses the same public key to "verify" the HMAC signature — and accepts it

**Defense:** Pin the expected algorithm in verification code, never derive it from the token's header.

**3. `kid` injection:**
The `kid` (Key ID) header parameter specifies which key to use for verification. If the server uses `kid` in a file path or database query without sanitization:
```json
{"alg": "HS256", "kid": "../../etc/passwd"}

Defense: Validate kid against an allowlist of known key IDs.

4. jwk and jku injection: The token header can include a JWK (JSON Web Key) or JKU (JWK Set URL) pointing to the attacker's key. If the server fetches and trusts this key, the attacker can sign tokens with their own key. Defense: Never trust keys embedded in or referenced by the token itself. Use only pre-configured keys.

### Mutual TLS (mTLS)

In standard TLS, only the server presents a certificate. In mTLS, the client also presents a certificate, providing strong mutual authentication at the transport layer — before any application code runs.

```bash
# Generate a client certificate for mTLS
# 1. Create an internal CA
openssl genrsa -out ca.key 4096
openssl req -new -x509 -days 365 -key ca.key -out ca.crt \
  -subj "/CN=Internal API CA"

# 2. Generate client key and CSR
openssl genrsa -out client.key 4096
openssl req -new -key client.key -out client.csr \
  -subj "/CN=orders-service/O=production"

# 3. Sign the client certificate
openssl x509 -req -days 90 -in client.csr \
  -CA ca.crt -CAkey ca.key -CAcreateserial -out client.crt

# 4. Make an mTLS request
curl --cert client.crt --key client.key --cacert ca.crt \
  https://api.internal.example.com/v1/data

mTLS comparison with other methods:

Rate Limiting Algorithms — In Depth

How should rate limiting actually work at the algorithm level? Here are the three most important approaches.

Token Bucket Algorithm

The most widely used rate limiting algorithm. A bucket holds N tokens and refills at a steady rate. Each request consumes a token. When the bucket is empty, requests are rejected.

stateDiagram-v2
    [*] --> Full: Initialize bucket<br/>capacity=10, rate=2/sec

    Full --> Available: Request arrives<br/>consume 1 token
    Available --> Available: Request arrives<br/>consume 1 token

    Available --> Empty: Last token consumed

    Empty --> Available: Wait → tokens refill<br/>(2 per second)
    Empty --> Rejected: Request arrives<br/>no tokens available

    Rejected --> Available: Wait → tokens refill

    state Full {
        [*] --> Tokens10: 10/10 tokens
    }
    state Available {
        [*] --> TokensN: 1-9 tokens
    }
    state Rejected {
        [*] --> Return429: 429 Too Many Requests<br/>Retry-After: N seconds
    }

import time
import threading

class TokenBucket:
    """Thread-safe token bucket rate limiter."""

    def __init__(self, capacity: int, refill_rate: float):
        self.capacity = capacity
        self.tokens = capacity
        self.refill_rate = refill_rate  # tokens per second
        self.last_refill = time.monotonic()
        self.lock = threading.Lock()

    def consume(self, tokens: int = 1) -> bool:
        with self.lock:
            now = time.monotonic()
            # Refill tokens based on elapsed time
            elapsed = now - self.last_refill
            self.tokens = min(
                self.capacity,
                self.tokens + elapsed * self.refill_rate
            )
            self.last_refill = now

            if self.tokens >= tokens:
                self.tokens -= tokens
                return True  # Request allowed
            return False  # Request rejected

    @property
    def retry_after(self) -> float:
        """Seconds until a token is available."""
        if self.tokens >= 1:
            return 0
        return (1 - self.tokens) / self.refill_rate

Sliding Window Log

Tracks the timestamp of every request. More accurate than fixed windows but uses more memory:

import time
from collections import defaultdict

class SlidingWindowLog:
    """Sliding window rate limiter using request timestamps."""

    def __init__(self, max_requests: int, window_seconds: int):
        self.max_requests = max_requests
        self.window = window_seconds
        self.requests = defaultdict(list)  # client_id -> [timestamps]

    def allow(self, client_id: str) -> bool:
        now = time.time()
        cutoff = now - self.window

        # Remove expired timestamps
        self.requests[client_id] = [
            ts for ts in self.requests[client_id] if ts > cutoff
        ]

        if len(self.requests[client_id]) < self.max_requests:
            self.requests[client_id].append(now)
            return True
        return False

Sliding Window Counter

A memory-efficient approximation that combines the current and previous fixed windows:

class SlidingWindowCounter:
    """Approximate sliding window using weighted fixed windows."""

    def __init__(self, max_requests: int, window_seconds: int):
        self.max_requests = max_requests
        self.window = window_seconds
        self.counters = {}  # client_id -> {window_key: count}

    def allow(self, client_id: str) -> bool:
        now = time.time()
        current_window = int(now // self.window)
        window_position = (now % self.window) / self.window

        current_count = self.counters.get(client_id, {}).get(current_window, 0)
        previous_count = self.counters.get(client_id, {}).get(current_window - 1, 0)

        # Weighted estimate: full current + proportional previous
        estimated = current_count + previous_count * (1 - window_position)

        if estimated < self.max_requests:
            if client_id not in self.counters:
                self.counters[client_id] = {}
            self.counters[client_id][current_window] = current_count + 1
            return True
        return False

What to Rate Limit By

Rate Limit Response Headers

HTTP/1.1 200 OK
X-RateLimit-Limit: 100
X-RateLimit-Remaining: 42
X-RateLimit-Reset: 1709245600

When limit is exceeded:

HTTP/1.1 429 Too Many Requests
Retry-After: 30
Content-Type: application/json

{
    "error": "rate_limit_exceeded",
    "message": "Too many requests. Please retry after 30 seconds.",
    "retry_after": 30
}

Rate limiting alone does not solve authentication bypass or authorization failures. An attacker with valid credentials making one request per minute but accessing other users' data is not a rate-limiting problem — it is an authorization problem. Rate limiting protects availability. Authorization protects confidentiality and integrity. Both are needed, and neither substitutes for the other.

OWASP API Security Top 10 — With Exploitation Examples

OWASP released a dedicated API Security Top 10 because API vulnerabilities differ from traditional web app vulnerabilities. APIs expose data and operations directly — there is no UI layer to accidentally obscure attack surfaces. Here is each one with real exploitation examples.

API1: Broken Object-Level Authorization (BOLA/IDOR)

The API does not verify that the authenticated user is authorized to access the specific object they requested. This is the most common API vulnerability.

# VULNERABLE — any authenticated user can access any order
@app.route('/api/v1/orders/<order_id>')
@require_auth
def get_order(order_id):
    order = db.orders.find_one({'_id': order_id})
    return jsonify(order)
    # User A can view User B's orders by changing the ID

# FIXED — filter by authenticated user
@app.route('/api/v1/orders/<order_id>')
@require_auth
def get_order(order_id):
    order = db.orders.find_one({
        '_id': order_id,
        'user_id': current_user.id  # Filter by authenticated user
    })
    if not order:
        return jsonify({'error': 'not found'}), 404
    return jsonify(order)

Exploitation automation:

# Enumerate orders — sequential IDs make this trivial
for id in $(seq 1 10000); do
    response=$(curl -s -H "Authorization: Bearer $TOKEN" \
        "https://api.example.com/v1/orders/$id")
    if echo "$response" | grep -q '"id"'; then
        echo "Found order $id: $response"
    fi
done

# UUID-based IDs are harder to enumerate but not immune
# Attacker finds one UUID from a legitimate interaction
# then tries variations or checks other endpoints that leak UUIDs

A ride-sharing API returned complete trip details — including driver name, phone number, GPS route, and fare — for any trip ID. The IDs were sequential integers. A script that iterated from 1 to 10,000 downloaded full trip histories for thousands of riders. All it would have taken to prevent this was adding `AND rider_id = :current_user` to the database query. The company had to notify 2.3 million affected users.

The fix is architecturally simple but requires discipline: every data access query must include a WHERE clause that restricts results to the authenticated user's data. No exceptions. No "but this endpoint is only used by our mobile app" — all endpoints are used by anyone who can send HTTP requests.

API2: Broken Authentication

Weak or missing authentication mechanisms:

# Test for missing authentication
curl -s https://api.example.com/v1/admin/users
# Should return 401, not data

# Test for weak token validation
# Modify a JWT payload without re-signing
echo '{"sub":"admin","role":"admin"}' | base64 | \
  xargs -I{} curl -s -H "Authorization: Bearer eyJhbGciOiJub25lIn0.{}.fake" \
  https://api.example.com/v1/admin/users

# Test for credential stuffing protection
for i in $(seq 1 100); do
    curl -s -o /dev/null -w "%{http_code}" \
        -X POST https://api.example.com/v1/login \
        -d '{"email":"target@example.com","password":"attempt'$i'"}'
done | sort | uniq -c
# If all return 200/401 with no 429 — no brute force protection

API3: Broken Object Property Level Authorization

Excessive data exposure — the API returns more data than the client needs:

// GET /api/v1/users/123 — returns EVERYTHING
{
    "id": 123,
    "name": "Alice",
    "email": "alice@example.com",
    "role": "user",
    "password_hash": "$2b$12$LJ3m4...",
    "ssn": "123-45-6789",
    "salary": 95000,
    "internal_notes": "VIP customer, CEO's nephew",
    "credit_card_last4": "4242",
    "failed_login_count": 0
}

Mass assignment — the API blindly accepts all fields from the client:

# User updates their profile — includes unauthorized field
curl -X PUT https://api.example.com/v1/users/123 \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"name": "Alice", "role": "admin", "salary": 500000}'
# If the API blindly applies all fields, the user just promoted themselves

Defense — explicit field control:

# Django REST Framework — explicit serializers
class UserReadSerializer(serializers.ModelSerializer):
    class Meta:
        model = User
        fields = ['id', 'name', 'email', 'avatar_url']
        # password_hash, ssn, role, salary — NOT exposed

class UserWriteSerializer(serializers.ModelSerializer):
    class Meta:
        model = User
        fields = ['name', 'email', 'avatar_url']
        # role, salary — NOT writable via API

# Pydantic (FastAPI)
class UserResponse(BaseModel):
    id: int
    name: str
    email: str

    class Config:
        orm_mode = True
        # Only these fields are serialized — nothing else leaks

class UserUpdate(BaseModel):
    name: Optional[str] = Field(None, max_length=100)
    email: Optional[str] = Field(None, max_length=254)
    # No role, salary, or any sensitive fields accepted

API4: Unrestricted Resource Consumption

No limits on the size or number of resources a client can request:

# Pagination abuse — request the entire database
curl "https://api.example.com/v1/users?limit=1000000&offset=0"

# Upload size abuse — fill the server's disk
curl -X POST https://api.example.com/v1/upload \
  -F "file=@gigantic_file.bin"  # No file size limit

# GraphQL complexity abuse (covered in detail below)

API5: Broken Function-Level Authorization

Regular users accessing admin endpoints:

# Regular user discovers admin endpoints
curl -H "Authorization: Bearer $USER_TOKEN" \
  https://api.example.com/v1/admin/users    # Should be 403
curl -H "Authorization: Bearer $USER_TOKEN" \
  -X DELETE https://api.example.com/v1/users/456  # Should be 403
curl -H "Authorization: Bearer $USER_TOKEN" \
  -X PUT https://api.example.com/v1/settings/global  # Should be 403

# HTTP method bypass
curl -X PUT -H "Authorization: Bearer $USER_TOKEN" \
  https://api.example.com/v1/users/456
# PUT and DELETE often have weaker auth checks than GET

API6: Unrestricted Access to Sensitive Business Flows

Automated abuse of legitimate functionality:

• Bot purchasing of limited-edition sneakers
• Automated account creation for spam
• Mass coupon/promo code redemption
• Automated ticket scalping

Defense: CAPTCHA, device fingerprinting, behavioral analysis, business-logic rate limits (e.g., max 2 purchases of the same item per account per day).

API7: Server-Side Request Forgery (SSRF)

APIs are especially vulnerable because they often accept URLs as parameters:

# Webhook registration — attacker points to internal service
curl -X POST https://api.example.com/v1/webhooks \
  -d '{"url": "http://169.254.169.254/latest/meta-data/"}'

# File import from URL
curl -X POST https://api.example.com/v1/import \
  -d '{"source_url": "http://10.0.1.50:6379/"}'

# Avatar/image URL
curl -X PUT https://api.example.com/v1/users/me \
  -d '{"avatar_url": "http://localhost:8500/v1/agent/members"}'

API8: Security Misconfiguration

# Check for verbose error messages
curl -s https://api.example.com/v1/users/invalid
# BAD: {"error": "PG::UndefinedTable: ERROR: relation 'users' does not exist"}
# GOOD: {"error": "not_found", "message": "Resource not found"}

# Check for exposed debug endpoints
curl -s https://api.example.com/debug/vars
curl -s https://api.example.com/actuator/health
curl -s https://api.example.com/metrics
curl -s https://api.example.com/__debug__/

# Check for unnecessary HTTP methods
curl -sI -X TRACE https://api.example.com/v1/users
curl -sI -X OPTIONS https://api.example.com/v1/users

API9: Improper Inventory Management

# Discover old API versions
curl -s https://api.example.com/v1/users   # Current (secured)
curl -s https://api.example.com/v0/users   # Old (no auth?)
curl -s https://api.example.com/v2-beta/users  # Pre-release (no auth?)

# Discover documentation and schema
curl -s https://api.example.com/swagger.json
curl -s https://api.example.com/openapi.yaml
curl -s https://api.example.com/docs
curl -s https://api.example.com/.well-known/openapi
curl -s https://api.example.com/graphql  # Introspection

# Discover internal/staging endpoints
curl -s https://api-staging.example.com/v1/users
curl -s https://api-internal.example.com/v1/users

API10: Unsafe Consumption of APIs

Your API trusts data from third-party APIs without validation:

# VULNERABLE — trusts third-party response blindly
def process_payment(payment_id):
    # Call payment provider
    response = requests.get(f"https://payments.example.com/api/{payment_id}")
    data = response.json()

    # Directly uses provider's data without validation
    db.execute(f"UPDATE orders SET amount = {data['amount']}")  # SQL injection!
    db.execute(f"UPDATE orders SET status = '{data['status']}'")

# SAFE — validate third-party data like user input
def process_payment(payment_id):
    response = requests.get(f"https://payments.example.com/api/{payment_id}")
    data = response.json()

    # Validate
    amount = Decimal(str(data.get('amount', 0)))
    if amount < 0 or amount > 1_000_000:
        raise ValueError("Invalid amount from payment provider")

    status = data.get('status', '')
    if status not in ('completed', 'failed', 'pending'):
        raise ValueError("Invalid status from payment provider")

    # Parameterized query
    db.execute(
        "UPDATE orders SET amount = %s, status = %s WHERE payment_id = %s",
        (amount, status, payment_id)
    )

GraphQL-Specific Security Concerns

GraphQL's flexibility is both its strength and its security challenge. Every feature that makes GraphQL powerful for developers also makes it powerful for attackers.

Introspection — Your Schema Is Showing

GraphQL supports introspection — querying the schema itself:

{
  __schema {
    types {
      name
      fields {
        name
        type { name kind }
        args { name type { name } }
      }
    }
    mutationType {
      fields { name }
    }
  }
}

This reveals every type, field, query, mutation, and argument in your API. Disable introspection in production:

// Apollo Server
const server = new ApolloServer({
    typeDefs,
    resolvers,
    introspection: process.env.NODE_ENV !== 'production',
});

// GraphQL Yoga
const yoga = createYoga({
    schema,
    graphiql: process.env.NODE_ENV !== 'production',
});

Query Depth Attacks

GraphQL allows nested queries that can consume exponential resources:

# Each nesting level multiplies database queries
# Depth 5 on a social graph = potentially millions of records
{
  user(id: 1) {
    friends {       # 100 friends
      friends {     # 100 * 100 = 10,000
        friends {   # 100^3 = 1,000,000
          friends {  # 100^4 = 100,000,000
            name
            email
          }
        }
      }
    }
  }
}

Batching Attacks

GraphQL allows multiple operations in a single request, bypassing per-request rate limiting:

[
  {"query": "mutation { login(user:\"admin\", pass:\"password1\") { token } }"},
  {"query": "mutation { login(user:\"admin\", pass:\"password2\") { token } }"},
  {"query": "mutation { login(user:\"admin\", pass:\"password3\") { token } }"},
  {"query": "mutation { login(user:\"admin\", pass:\"password4\") { token } }"}
]

One HTTP request, four login attempts. Per-HTTP-request rate limiting sees one request.

Alias-Based Attacks

Even without batching, GraphQL aliases allow multiple operations per query:

{
  a1: login(user: "admin", pass: "pass1") { token }
  a2: login(user: "admin", pass: "pass2") { token }
  a3: login(user: "admin", pass: "pass3") { token }
  # ... 100 aliases = 100 login attempts in 1 query
}

GraphQL Defense

# Query complexity analysis — assign cost to each field
from graphql import GraphQLError

class ComplexityAnalyzer:
    """Reject queries exceeding a complexity budget."""

    FIELD_COSTS = {
        'friends': 10,      # Expensive — joins, pagination
        'orders': 5,        # Moderate
        'name': 1,          # Cheap — single column
        'email': 1,
    }
    MAX_COMPLEXITY = 1000

    def analyze(self, query_ast):
        complexity = self._calculate(query_ast)
        if complexity > self.MAX_COMPLEXITY:
            raise GraphQLError(
                f"Query complexity {complexity} exceeds maximum {self.MAX_COMPLEXITY}"
            )
        return complexity

    def _calculate(self, node, depth=0, parent_multiplier=1):
        # Recursive calculation with depth and list multiplication
        total = 0
        for field in node.selection_set.selections:
            cost = self.FIELD_COSTS.get(field.name.value, 1)
            multiplier = self._get_limit_arg(field) or 10  # Default list size
            field_cost = cost * parent_multiplier

            if field.selection_set:
                field_cost += self._calculate(
                    field, depth + 1, parent_multiplier * multiplier
                )
            total += field_cost
        return total

Test GraphQL security:

~~~bash
# Test introspection (should be disabled in production)
curl -s -X POST https://api.example.com/graphql \
  -H "Content-Type: application/json" \
  -d '{"query": "{ __schema { types { name } } }"}'

# Test query depth
curl -s -X POST https://api.example.com/graphql \
  -H "Content-Type: application/json" \
  -d '{"query": "{ user(id:1) { friends { friends { friends { friends { name } } } } } }"}'

# Test batching
curl -s -X POST https://api.example.com/graphql \
  -H "Content-Type: application/json" \
  -d '[{"query":"{ user(id:1) { name } }"},{"query":"{ user(id:2) { name } }"}]'

# Test alias abuse
curl -s -X POST https://api.example.com/graphql \
  -H "Content-Type: application/json" \
  -d '{"query": "{ a:user(id:1){name} b:user(id:2){name} c:user(id:3){name} }"}'
~~~

API Gateway Architecture

An API gateway sits between clients and backend services, providing a centralized enforcement point for cross-cutting security concerns.

graph TD
    subgraph "Clients"
        W[Web App]
        M[Mobile App]
        P[Partner API]
        I[IoT Device]
    end

    subgraph "API Gateway Layer"
        GW[API Gateway]
        AUTH[Authentication<br/>JWT / mTLS / API Key]
        RL[Rate Limiting<br/>Token Bucket per client]
        WAF_G[WAF Rules<br/>OWASP CRS]
        LOG[Access Logging<br/>& Audit Trail]
        TRANSFORM[Request Validation<br/>& Transformation]
        CACHE[Response Cache]
    end

    subgraph "Backend Services"
        US[User Service]
        OS[Order Service]
        PS[Payment Service]
        NS[Notification Service]
    end

    W --> GW
    M --> GW
    P --> GW
    I --> GW

    GW --> AUTH
    AUTH --> RL
    RL --> WAF_G
    WAF_G --> TRANSFORM
    TRANSFORM --> LOG

    LOG --> US
    LOG --> OS
    LOG --> PS
    LOG --> NS

    style GW fill:#3498db,stroke:#2980b9,color:#fff

Gateway security functions:

The gateway handles cross-cutting security concerns, while individual services handle business-level authorization. The gateway is your perimeter defense — it answers "is this a valid, authenticated, rate-limited request?" Services still must answer "can this user access this specific resource?" The gateway cannot know that User A should not see User B's orders — that is business logic that belongs in the service.

Input Validation at the API Boundary

Every piece of data entering your API must be validated before it touches business logic or persistence. Trust nothing.

# Pydantic v2 (FastAPI) — strict schema validation
from pydantic import BaseModel, Field, field_validator, ConfigDict
from typing import Optional
import re

class CreateUserRequest(BaseModel):
    model_config = ConfigDict(extra='forbid')  # Reject unknown fields

    name: str = Field(..., min_length=1, max_length=100)
    email: str = Field(..., max_length=254)
    age: Optional[int] = Field(None, ge=0, le=150)
    role: None = Field(None, exclude=True)  # Explicitly block this field

    @field_validator('email')
    @classmethod
    def validate_email(cls, v):
        pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        if not re.match(pattern, v):
            raise ValueError('Invalid email format')
        return v.lower()

    @field_validator('name')
    @classmethod
    def validate_name(cls, v):
        if re.search(r'[<>"\';]', v):
            raise ValueError('Name contains invalid characters')
        return v.strip()

// JSON Schema — for language-agnostic validation
{
    "type": "object",
    "required": ["name", "email"],
    "properties": {
        "name": {
            "type": "string",
            "minLength": 1,
            "maxLength": 100,
            "pattern": "^[a-zA-Z0-9 .'-]+$"
        },
        "email": {
            "type": "string",
            "format": "email",
            "maxLength": 254
        },
        "age": {
            "type": "integer",
            "minimum": 0,
            "maximum": 150
        }
    },
    "additionalProperties": false
}

The "additionalProperties": false is critical — it rejects any fields not explicitly defined, preventing mass assignment attacks.

Validation layers in a well-designed API:

1. **Transport layer:** TLS version, certificate validation
2. **Gateway layer:** Request size limits, content-type enforcement, rate limiting
3. **Schema layer:** JSON/XML schema validation, type checking, field constraints
4. **Business logic layer:** Domain-specific rules (order quantity ≤ stock, dates in future)
5. **Persistence layer:** Database constraints, foreign key validation, unique constraints

Each layer catches different classes of invalid input. Do not rely on any single layer. A request might pass schema validation (valid JSON, correct types) but fail business validation (negative order quantity). It might pass business validation but fail a database constraint (duplicate email).

API Versioning and Security Implications

A payment processor maintained three API versions simultaneously. Version 3 had proper authentication, rate limiting, and PCI compliance. Version 1 had been "deprecated" two years earlier — meaning they sent a deprecation header but never shut it down. An attacker discovered v1, which had no rate limiting and returned unmasked credit card numbers in responses.

The v1 API was serving 40 requests per second to an IP address in a known botnet — for two months. The deprecation notice was in the documentation. Nobody reads documentation. Nobody monitored v1 traffic.

**The lesson:** Deprecation without removal is not deprecation. Set hard sunset dates. Monitor traffic to old versions. When you say "deprecated," mean "deleted."

# Nginx — block deprecated API version with a clear response
location /api/v1/ {
    return 410 '{"error": "gone", "message": "API v1 was permanently removed on 2025-01-01. Use /api/v3/"}';
    add_header Content-Type application/json always;
}

# Redirect old version to current with a warning
location /api/v2/ {
    return 301 /api/v3/$request_uri;
    add_header X-API-Warning "API v2 is deprecated. Please migrate to v3.";
}

What You've Learned

This chapter covered the security landscape specific to APIs, which are now the primary attack surface for most applications:

• Authentication patterns range from API keys (simple, server-to-server) to OAuth 2.0 tokens (user-scoped, expiring, with PKCE for public clients) to mTLS (transport-level, certificate-based). Each has specific security considerations, failure modes, and appropriate use cases.

• Rate limiting algorithms (token bucket, sliding window log, sliding window counter) protect availability but must be applied at the right level — per-user, per-endpoint, per-operation. Rate limiting does not substitute for authorization.

• The OWASP API Security Top 10 highlights that BOLA (broken object-level authorization) is the number one API vulnerability. Authentication is not authorization. Every data access must verify the user owns the resource. Every field exposed must be intentional.

• Input validation must happen at the API boundary using strict schemas. Reject unknown fields (additionalProperties: false) to prevent mass assignment. Validate third-party API responses with the same rigor as user input.

• GraphQL introduces unique security concerns: introspection exposure, query depth and complexity attacks, batching and alias abuse, and per-field authorization requirements. Disable introspection in production and enforce complexity budgets.

• API gateways centralize cross-cutting security (auth, rate limiting, WAF) but do not replace per-service authorization. The gateway answers "is this request valid?" The service answers "is this user allowed to access this specific resource?"

• API versioning creates security debt. Old versions must be actively decommissioned, not just deprecated in documentation. Monitor traffic to all versions.

Authenticate everything, authorize every object access, validate every input, rate limit every endpoint, decommission old versions — and log everything. When — not if — something slips through, your logs are the difference between a one-day incident and a one-year undetected breach. APIs are designed for machines to consume at scale. Attackers are machines too. Design your defenses accordingly.