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:

| 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  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 equ

…

Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: frank-luongt
• Source: frank-luongt/faos-skills-marketplace
• License: Apache-2.0
• Homepage: https://faosx.ai/skills

Install and usage instructions live in the source repository linked above.