API Security Patterns

name: api-security-patterns description: Advanced API security patterns covering OWASP API Top 10, authentication, authorization, rate limiting, and gateway hardening tags: [api, security]

Advanced API Security Patterns

Overview

APIs are the primary attack surface for modern applications. Over 90% of web-enabled applications expose more attack surface through APIs than through traditional UI. This skill provides defense-in-depth patterns for API security, mapping OWASP API Security Top 10 risks to concrete prevention code.

Core principle: Every API endpoint is an entry point for attackers. Authenticate, authorize, validate, and rate-limit every request -- no exceptions.

This skill enhances the existing @owasp-api-top10 skill with implementation-level patterns for Python/FastAPI, OPA policy-as-code, Redis-based rate limiting, and GraphQL-specific defenses.

When to Use This Skill

• Designing a new API or microservice
• Conducting a security review of existing API endpoints
• Implementing authentication or authorization middleware
• Setting up rate limiting or abuse prevention
• Hardening a GraphQL or WebSocket API
• Responding to an API-related security finding or penetration test result
• Preparing for SOC 2, PCI DSS, or HIPAA compliance audits that cover API security

How It Works

Step 1: Threat Model the API Surface

Before writing security code, enumerate your attack surface:

API Threat Model Checklist:
============================
[ ] List all endpoints (including undocumented/debug endpoints)
[ ] Classify data sensitivity per endpoint (public, internal, confidential, restricted)
[ ] Identify authentication requirements per endpoint
[ ] Map authorization rules (who can access what)
[ ] Document rate limiting requirements per endpoint/consumer
[ ] Identify external dependencies and trust boundaries
[ ] Review OWASP API Top 10 against each endpoint:
    API1  - Broken Object Level Authorization (BOLA)
    API2  - Broken Authentication
    API3  - Broken Object Property Level Authorization
    API4  - Unrestricted Resource Consumption
    API5  - Broken Function Level Authorization
    API6  - Unrestricted Access to Sensitive Business Flows
    API7  - Server-Side Request Forgery (SSRF)
    API8  - Security Misconfiguration
    API9  - Improper Inventory Management
    API10 - Unsafe Consumption of APIs

Step 2: Implement the Authentication Layer

JWT Best Practices:

• Use short-lived access tokens (5-15 minutes)
• Implement refresh token rotation (one-time use refresh tokens)
• Validate iss, aud, exp, nbf claims on every request
• Use asymmetric signing (RS256/ES256) for distributed systems
• Store refresh tokens server-side with revocation capability

OAuth 2.0 + PKCE (for public clients):

• Always use Authorization Code flow with PKCE for SPAs and mobile apps
• Never use Implicit flow (deprecated in OAuth 2.1)
• Validate state parameter to prevent CSRF
• Use code_verifier / code_challenge (S256 method)

mTLS (for service-to-service):

• Require client certificates for internal microservice communication
• Pin certificates or use a private CA
• Rotate certificates before expiry with automated cert management

Step 3: Add Input Validation

Validate every input at the API boundary. Never trust client data.

Validation layers:

1. Content-Type enforcement -- reject requests with unexpected content types
2. Request size limits -- prevent resource exhaustion from oversized payloads
3. JSON Schema validation -- enforce structure, types, and constraints
4. Business logic validation -- domain-specific rules (ranges, formats, references)

Step 4: Configure Rate Limiting

Apply rate limits at multiple layers:

1. Global -- protect infrastructure (e.g., 10,000 req/min per IP)
2. Per-user -- prevent abuse by authenticated users (e.g., 100 req/min per user)
3. Per-endpoint -- protect expensive operations (e.g., 5 req/min for password reset)
4. Per-tenant -- enforce SLA-based limits in multi-tenant systems

Algorithms:

Step 5: Set Up Monitoring and Alerting

Monitor for:

• Authentication failure spikes (brute force indicator)
• Authorization failures (BOLA/BFLA probing)
• Rate limit hits by consumer (abuse detection)
• Unusual request patterns (scraping, enumeration)
• Response size anomalies (data exfiltration indicator)
• Latency anomalies (injection, DoS indicator)

Examples

Example 1: FastAPI Middleware for JWT Validation with Role-Based Access

"""
JWT authentication middleware for FastAPI with RBAC.
Validates tokens, extracts claims, and enforces role-based access control.
"""

from datetime import datetime, timezone
from enum import Enum
from functools import wraps
from typing import Optional

import httpx
from fastapi import Depends, HTTPException, Request, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from jose import JWTError, jwt
from pydantic import BaseModel

# --- Configuration ---

class AuthConfig(BaseModel):
    jwks_url: str          # e.g., "https://auth.example.com/.well-known/jwks.json"
    issuer: str            # e.g., "https://auth.example.com/"
    audience: str          # e.g., "https://api.example.com"
    algorithm: str = "RS256"
    token_ttl_leeway: int = 30  # seconds of clock skew tolerance


class Role(str, Enum):
    VIEWER = "viewer"
    EDITOR = "editor"
    ADMIN = "admin"
    SUPER_ADMIN = "super_admin"


ROLE_HIERARCHY = {
    Role.SUPER_ADMIN: 4,
    Role.ADMIN: 3,
    Role.EDITOR: 2,
    Role.VIEWER: 1,
}


class TokenClaims(BaseModel):
    sub: str
    tenant_id: str
    roles: list[str]
    exp: int
    iss: str
    aud: str
    jti: Optional[str] = None


# --- JWKS Key Cache ---

class JWKSClient:
    """Caches JWKS keys with automatic refresh."""

    def __init__(self, jwks_url: str, cache_ttl: int = 3600):
        self._jwks_url = jwks_url
        self._cache_ttl = cache_ttl
        self._keys: dict = {}
        self._last_fetched: float = 0

    async def get_signing_key(self, kid: str) -> dict:
        now = datetime.now(timezone.utc).timestamp()
        if now - self._last_fetched > self._cache_ttl or kid not in self._keys:
            await self._refresh_keys()
        if kid not in self._keys:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Unknown signing key",
            )
        return self._keys[kid]

    async def _refresh_keys(self) -> None:
        async with httpx.AsyncClient(timeout=10) as client:
            resp = await client.get(self._jwks_url)
            resp.raise_for_status()
            jwks = resp.json()
        self._keys = {key["kid"]: key for key in jwks.get("keys", [])}
        self._last_fetched = datetime.now(timezone.utc).timestamp()


# --- Token Validation ---

security_scheme = HTTPBearer()


async def validate_token(
    request: Request,
    credentials: HTTPAuthorizationCredentials = Depends(security_scheme),
) -> TokenClaims:
    """Validate JWT and extract claims. Use as a FastAPI dependency."""
    config: AuthConfig = request.app.state.auth_config
    jwks_client: JWKSClient = request.app.state.jwks_client
    token = credentials.credentials

    try:
        # Decode header without verification to get kid
        unverified_header = jwt.get_unverified_header(token)
        kid = unverified_header.get("kid")
        if not kid:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Token missing key ID",
            )

        # Fetch the signing key
        signing_key = await jwks_client.get_signing_key(kid)

        # Verify and decode the token
        payload = jwt.decode(
            token,
            signing_key,
            algorithms=[config.algorithm],
            audience=config.audience,
            issuer=config.issuer,
            options={
                "verify_exp": True,
                "verify_aud": True,
                "verify_iss": True,
                "verify_nbf": True,
                "leeway": config.token_ttl_leeway,
            },
        )

        claims = TokenClaims(**payload)

        # Bind tenant_id and sub to request state for downstream use
        request.state.tenant_id = claims.tenant_id
        request.state.user_id = claims.sub
        request.state.roles = claims.roles

        return claims

    except JWTError as e:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=f"Invalid token: {e}",
        )


# --- Role-Based Access Control Decorator ---

def require_role(minimum_role: Role):
    """Decorator that enforces minimum role level on a route."""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, claims: TokenClaims = Depends(validate_token), **kwargs):
            user_max_role = max(
                (ROLE_HIERARCHY.get(Role(r), 0) for r in claims.roles),
                default=0,
            )
            if user_max_role < ROLE_HIERARCHY[minimum_role]:
                raise HTTPException(
                    status_code=status.HTTP_403_FORBIDDEN,
                    detail=f"Requires role: {minimum_role.value}",
                )
            return await func(*args, claims=claims, **kwargs)
        return wrapper
    return decorator


# --- Usage in Routes ---

# @router.get("/admin/users")
# @require_role(Role.ADMIN)
# async def list_users(claims: TokenClaims = Depends(validate_token)):
#     """Only admins and above can list users."""
#     return await user_service.list_by_tenant(claims.tenant_id)
#
# @router.get("/documents/{doc_id}")
# async def get_document(doc_id: str, claims: TokenClaims = Depends(validate_token)):
#     """BOLA prevention: always filter by tenant_id from token, never from request."""
#     doc = await doc_service.get(doc_id, tenant_id=claims.tenant_id)
#     if not doc:
#         raise HTTPException(status_code=404, detail="Not found")
#     return doc

Example 2: OPA Policy for API Authorization (Rego)

# policy/api_authz.rego
#
# Open Policy Agent (OPA) policy for API authorization.
# Enforces tenant isolation, role-based access, and resource-level permissions.

package api.authz

import future.keywords.if
import future.keywords.in

default allow := false

# --- Tenant Isolation (BOLA Prevention) ---
# Every request must include a tenant_id that matches the token's tenant.

allow if {
    valid_tenant
    valid_role
    valid_resource_access
}

valid_tenant if {
    # The requested resource's tenant must match the token's tenant
    input.resource.tenant_id == input.token.tenant_id
}

# --- Role-Based Access Control ---

role_permissions := {
    "viewer":      {"GET"},
    "editor":      {"GET", "POST", "PUT", "PATCH"},
    "admin":       {"GET", "POST", "PUT", "PATCH", "DELETE"},
    "super_admin": {"GET", "POST", "PUT", "PATCH", "DELETE"},
}

valid_role if {
    some role in input.token.roles
    input.request.method in role_permissions[role]
}

# --- Resource-Level Permissions ---

# Admin-only endpoints
admin_paths := {"/api/v1/admin", "/api/v1/users", "/api/v1/audit-logs"}

valid_resource_access if {
    not startswith_any(input.request.path, admin_paths)
}

valid_resource_access if {
    startswith_any(input.request.path, admin_paths)
    "admin" in input.token.roles
}

valid_resource_access if {
    startswith_any(input.request.path, admin_paths)
    "super_admin" in input.token.roles
}

# --- Rate Limit Tier (advisory, enforced by gateway) ---

rate_limit_tier := tier if {
    "super_admin" in input.token.roles
    tier := "unlimited"
} else := tier if {
    "admin" in input.token.roles
    tier := "high"
} else := tier if {
    tier := "standard"
}

# --- Helper Functions ---

startswith_any(str, prefixes) if {
    some prefix in prefixes
    startswith(str, prefix)
}

# --- Deny Reasons (for debugging and audit logging) ---

reasons[reason] if {
    not valid_tenant
    reason := "tenant_id mismatch: cross-tenant access denied"
}

reasons[reason] if {
    not valid_role
    reason := sprintf("role '%v' not authorized for method '%v'", [input.token.roles, input.request.method])
}

reasons[reason] if {
    not valid_resource_access
    reason := sprintf("path '%v' requires admin role", [input.request.path])
}

Example 3: Redis-Based Distributed Rate Limiter

"""
Sliding window rate limiter using Redis sorted sets.
Suitable for distributed API deployments behind a load balancer.
"""

import time
from dataclasses import dataclass
from typing import Optional

import redis.asyncio as redis
from fastapi import HTTPException, Request, status


@dataclass
class RateLimitConfig:
    """Rate limit configuration per endpoint or consumer tier."""
    requests: int       # max requests allowed
    window_seconds: int # time window in seconds
    burst_multiplier: float = 1.0  # allow burst up to N * requests in any sub-window


@dataclass
class RateLimitResult:
    allowed: bool
    limit: int
    remaining: int
    retry_after: Optional[int]  # seconds until next allowed request
    reset_at: float             # Unix timestamp when the window resets


class SlidingWindowRateLimiter:
    """
    Sliding window rate limiter using Redis sorted sets.

    Algorithm:
    1. Use a sorted set per (identifier, endpoint) pair
    2. Score = request timestamp
    3. Remove entries older than the window
    4. Count remaining entries
    5. If count < limit, allow and add new entry
    6. All operations in a single Redis pipeline (atomic)
    """

    def __init__(self, redis_client: redis.Redis, key_prefix: str = "faos:ratelimit"):
        self._redis = redis_client
        self._prefix = key_prefix

    async def check(
        self,
        identifier: str,
        endpoint: str,
        config: RateLimitConfig,
    ) -> RateLimitResult:
        """Check and consume a rate limit token. Returns result with headers."""
        now = time.time()
        window_start = now - config.window_seconds
        key = f"{self._prefix}:{identifier}:{endpoint}"

        async with self._redis.pipeline(transaction=True) as pipe:
            # Remove expired entries
            pipe.zremrangebyscore(key, 0, window_start)
            # Count current entries in window
            pipe.zcard(key)
            # Add current request (optimistically)
            pipe.zadd(key, {f"{now}:{id(now)}": now})
            # Set TTL to auto-cleanup
            pipe.expire(key, config.window_seconds + 1)
            results = await pipe.execute()

        current_count = results[1]  # zcard result (before adding new entry)

        if current_count >= config.requests:
            # Over limit -- remove the optimistically added entry
            await self._redis.zremrangebyscore(key, now, now + 0.001)

            # Calculate retry-after from oldest entry in window
            oldest = await self._redis.zrange(key, 0, 0, withscores=True)
            retry_after = int(oldest[0][1] + config.window_seconds - now) + 1 if oldest else config.window_seconds

            return RateLimitResult(
                allowed=False,
                limit=config.requests,
                remaining=0,
                retry_after=retry_after,
                reset_at=now + retry_after,
            )

        return RateLimitResult(
            allowed=True,
            limit=config.requests,
            remaining=config.requests - current_count - 1,
            retry_after=None,
            reset_at=now + config.window_seconds,
        )


# --- Rate Limit Configurations by Tier ---

RATE_LIMITS = {
    "default": RateLimitConfig(requests=100, window_seconds=60),
    "auth_endpoints": RateLimitConfig(requests=10, window_seconds=60),
    "search": RateLimitConfig(requests=30, window_seconds=60),
    "export": RateLimitConfig(requests=5, window_seconds=300),
    "premium_tier": RateLimitConfig(requests=500, window_seconds=60),
}


# --- FastAPI Middleware ---

async def rate_limit_middleware(request: Request, call_next):
    """FastAPI middleware that enforces rate limits and sets response headers."""
    limiter: SlidingWindowRateLimiter = request.app.state.rate_limiter

    # Determine identifier: authenticated user > API key > IP
    identifier = getattr(request.state, "user_id", None)
    if not identifier:
        identifier = request.headers.get("X-API-Key", request.client.host)

    # Determine rate limit config based on endpoint
    endpoint_key = "default"
    if request.url.path.startswith("/api/v1/auth"):
        endpoint_key = "auth_endpoints"
    elif request.url.path.startswith("/api/v1/search"):
        endpoint_key = "search"
    elif request.url.path.startswith("/api/v1/export"):
        endpoint_key = "export"

    config = RATE_LIMITS[endpoint_key]
    result = await limiter.check(identifier, endpoint_key, config)

    if not result.allowed:
        raise HTTPException(
            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
            detail="Rate limit exceeded",
            headers={
                "X-RateLimit-Limit": str(result.limit),
                "X-RateLimit-Remaining": "0",
                "X-RateLimit-Reset": str(int(result.reset_at)),
                "Retry-After": str(result.retry_after),
            },
        )

    response = await call_next(request)
    response.headers["X-RateLimit-Limit"] = str(result.limit)
    response.headers["X-RateLimit-Remaining"] = str(result.remaining)
    response.headers["X-RateLimit-Reset"] = str(int(result.reset_at))
    return response

Best Practices

Do This

• Always validate tenant_id from the token, never from request parameters -- this is the single most effective defense against BOLA (API1)
• Use allowlists for input validation, not blocklists -- reject everything not explicitly permitted
• Enforce Content-Type headers -- reject requests with mismatched or missing content types
• Return generic error messages to clients -- detailed errors go to logs, not to attackers
• Implement request signing for webhook endpoints -- verify HMAC signatures to prevent forgery
• Use parameterized queries exclusively -- never concatenate user input into queries
• Apply rate limits at multiple layers -- gateway, application, and database
• Log all authentication and authorization failures -- these are attack indicators
• Version your APIs and deprecate old versions -- unmanaged API versions are a top source of breaches (API9)
• Validate JWTs on every request -- never cache authorization decisions client-side

Don't Do This

• Don't use API keys as the sole authentication mechanism -- API keys are credentials, not identity; combine with OAuth 2.0
• Don't expose sequential/guessable IDs -- use UUIDs or opaque identifiers to prevent enumeration
• Don't trust client-side rate limiting or validation -- all security controls must be enforced server-side
• Don't return stack traces or internal error details in production -- this reveals your technology stack and code paths
• Don't store JWTs in localStorage -- use httpOnly, Secure, SameSite=Strict cookies instead
• Don't allow unlimited query depth in GraphQL -- set depth limits (typically 5-7) and complexity budgets
• Don't skip TLS certificate validation for internal services -- internal networks are not trusted zones
• Don't hardcode API keys or secrets in source code -- use environment variables or a secrets manager

Security Checklist

Authentication

• All endpoints require authentication except explicitly public ones
• JWTs use asymmetric signing (RS256 or ES256)
• Access token lifetime is 15 minutes or less
• Refresh tokens are single-use and stored server-side
• Token revocation is implemented and tested
• PKCE is enforced for all public OAuth clients
• mTLS is configured for service-to-service communication
• API keys are scoped to specific operations and have expiration dates

Authorization

• Object-level authorization checks on every data access (BOLA prevention)
• Function-level authorization enforced (admin endpoints restricted)
• Tenant isolation enforced at the query layer (tenant_id from token, not request)
• Policy-as-code implemented (OPA, Cedar, or equivalent)
• Authorization decisions are logged for audit
• Principle of least privilege applied to all API consumers

Input Validation

• JSON Schema validation on all request bodies
• Request size limits configured (body, headers, URL length)
• Content-Type enforcement (reject unexpected types)
• Path parameters validated (type, format, range)
• Query parameters validated and bounded (pagination limits)
• File upload validation (type, size, content scanning)

Rate Limiting and Abuse Prevention

• Rate limits configured per endpoint, per user, and per IP
• Authentication endpoints have strict rate limits (brute force prevention)
• Rate limit headers returned in responses (X-RateLimit-*)
• Retry-After header sent on 429 responses
• Bot detection implemented for public endpoints
• Distributed rate limiting for multi-node deployments

Transport and Infrastructure

• TLS 1.2+ enforced on all endpoints
• HSTS headers configured with long max-age
• CORS policy restricts allowed origins to known domains
• Security headers set (X-Content-Type-Options, X-Frame-Options, CSP)
• API gateway WAF rules configured and tuned
• Health check and debug endpoints are not publicly accessible
• API inventory is maintained and reviewed quarterly (API9 prevention)

Related Skills

• @owasp-api-top10 -- detailed reference for each OWASP API Security Top 10 risk
• @owasp-top10 -- web application security risks (overlaps with API security)
• @cwe-sans-top25 -- common weakness enumeration for secure coding practices

Additional Resources

• OWASP API Security Top 10 (2023) -- primary risk taxonomy for API security
• OWASP API Security Testing Guide -- methodology for testing API security controls
• OAuth 2.1 Draft Specification -- consolidated best practices for OAuth implementations
• NIST SP 800-204 -- Security Strategies for Microservices-based Application Systems
• Open Policy Agent Documentation -- policy-as-code engine for API authorization
• RFC 6750 -- Bearer Token Usage in OAuth 2.0
• RFC 7519 -- JSON Web Token (JWT) specification
• RFC 9110 -- HTTP Semantics (status codes, methods, headers)
• CIS API Security Benchmark -- hardening checklist for API infrastructure