Implement advanced API security patterns to protect against common vulnerabilities.

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:

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

Step 4: Configure Rate Limiting

Apply rate limits at multiple layers:

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

Algorithms:

Algorithm	Best For	Burst Tolerance
Fixed Window	Simple, low-memory	Allows 2x burst at window boundary
Sliding Window Log	Precise, audit-friendly	None (exact count)
Sliding Window Counter	Good balance of precision and memory	Minimal
Token Bucket	Steady rate with controlled bursts	Configurable
Leaky Bucket	Strict output rate smoothing	None

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

Api Security Patterns