Api Security

API Security Development Guide

Structured approach to building secure APIs, covering OWASP API Security Top 10 (2023), secure design patterns, and verification checklists. Apply these guidelines throughout the API development lifecycle — from threat modeling to deployment monitoring.

Secure API Development Lifecycle

Phase 1: API Threat Modeling and Design

• Identify API attack surfaces: public endpoints, authenticated endpoints, admin endpoints, webhooks, third-party integrations
• Map data flows: what sensitive data crosses each API boundary
• Define authorization model: which users/roles access which resources and properties
• Design security controls:
• Centralized authentication (OAuth2/OIDC at API gateway)
• Object-level authorization at data access layer
• Schema-based input validation at every endpoint
• Rate limiting per endpoint sensitivity
• API versioning with deprecation strategy

Phase 2: Secure Implementation

Critical API Security Rules

Reference detailed guides:

• For OWASP API Top 10 with code examples: See references/owasp-api-top-10.md
• For secure API design patterns: See references/secure-api-design.md

Phase 3: API Security Verification

1. Schema Validation — Lint OpenAPI spec for security issues (Spectral) 2. Static Analysis — Run SAST on API code (Semgrep, bandit) 3. Contract Testing — Verify API behavior matches spec (Schemathesis, Dredd) 4. Dynamic Testing — Run DAST against running API (OWASP ZAP, Burp Suite) 5. Authorization Testing — Test every endpoint with wrong user/role/anonymous 6. Rate Limit Testing — Verify all endpoints enforce limits 7. Code Review — Apply API security checklists

Reference: See references/api-security-checklist.md

Phase 4: Deployment and Monitoring

• API gateway: auth offloading, rate limiting, request logging
• TLS 1.2+ enforcement, HSTS, security headers
• Structured logging (no tokens/PII in logs)
• Anomaly detection and alerting on security events
• API inventory management and version deprecation
• Incident response plan for API breaches

OWASP API Security Top 10 (2023) Quick Reference

For detailed attack scenarios and code examples: See references/owasp-api-top-10.md

API Security Review Workflow

Step-by-step procedure for reviewing API security:

1. Map the API surface — List all endpoints, methods, auth requirements, and data flows. Check for undocumented/shadow endpoints. 2. Check authentication — Verify every non-public endpoint requires valid authentication. Test with missing/expired/malformed tokens. 3. Check object-level authorization (BOLA) — For every endpoint accepting resource IDs, verify users can only access their own resources. 4. Check function-level authorization (BFLA) — Verify admin endpoints reject non-admin users. Test horizontal and vertical privilege escalation. 5. Check property-level authorization — Verify responses only include authorized fields. Test mass assignment by sending extra fields in updates. 6. Validate input handling — Check schema validation on all inputs. Test with oversized payloads, unexpected types, injection payloads. 7. Check rate limiting — Verify limits on all endpoints, especially auth, business-critical, and resource-intensive operations. 8. Check error handling — Verify no sensitive info in error responses. Test with invalid inputs, missing resources, server errors. 9. Review third-party integrations — Verify external API responses are validated. Check for SSRF in URL-accepting endpoints. 10. Check API inventory — Verify no deprecated/shadow endpoints are live. Check documentation matches reality. 11. Report findings — Severity (Critical/High/Medium/Low), endpoint, vulnerable request, explanation, fix with code example.

API Security Testing Quick Commands

# === OpenAPI Spec Linting ===
npm install -g @stoplight/spectral-cli && spectral lint openapi.yaml

# === Property-based API Testing ===
pip install schemathesis && schemathesis run --checks all http://localhost:8000/openapi.json

# === Dynamic Security Scanning ===
docker run -t ghcr.io/zaproxy/zaproxy:stable zap-api-scan.py -t http://target:8000/openapi.json -f openapi

# === API Fuzzing ===
# nuclei -u http://target:8000 -t api/

# === Static Analysis (Python API) ===
pip install bandit && bandit -r src/ -f json
pip install semgrep && semgrep --config=p/python --config=p/owasp-top-ten src/

Reference Files

• [references/owasp-api-top-10.md](references/owasp-api-top-10.md) — Detailed OWASP API Security Top 10 (2023) with attack scenarios, vulnerable → secure code examples for REST and GraphQL APIs
• [references/secure-api-design.md](references/secure-api-design.md) — Secure API design patterns: authentication (OAuth2, JWT, API keys, mTLS), authorization (RBAC/ABAC), input validation, rate limiting, CORS, error handling, API gateway, monitoring
• [references/api-security-checklist.md](references/api-security-checklist.md) — Actionable checklists for API design review, auth, input validation, transport security, rate limiting, inventory, deployment, logging, and testing

API Security Verification Checklists

Comprehensive, actionable checklists for securing APIs throughout the development lifecycle.

Use these checklists during design reviews, code reviews, pre-deployment gates, and periodic audits.

>

Every item uses imperative form and is independently verifiable.

---

Table of Contents

• 1. API Design Review Checklist
• 2. Authentication and Authorization Checklist
• 3. Input Validation Checklist
• 4. Transport and Data Security Checklist
• 5. Rate Limiting and Resource Protection Checklist
• 6. API Inventory and Documentation Checklist
• 7. Deployment and Infrastructure Checklist
• 8. Logging and Monitoring Checklist
• 9. Testing Security Checklist
• 10. Security Tools Reference

---

1. API Design Review Checklist

Verify these items during API design and specification review, before implementation begins.

Authentication and Authorization by Design

• [ ] Require authentication on every endpoint (unless explicitly documented as public)
• [ ] Document and justify every public (unauthenticated) endpoint
• [ ] Enforce object-level authorization (BOLA) checks on every endpoint accepting resource IDs
• [ ] Enforce function-level authorization (BFLA) — separate and protect admin endpoints
• [ ] Enforce property-level authorization — explicitly define returned fields in response schemas (prohibit blanket to_dict()/to_json())
• [ ] Protect against mass assignment — accept only allowlisted fields on update endpoints
• [ ] Use UUIDs or non-sequential identifiers for resource IDs to reduce enumeration risk

Request and Response Design

• [ ] Define input validation schema for every endpoint (OpenAPI/JSON Schema)
• [ ] Configure rate limiting per endpoint based on sensitivity
• [ ] Enforce pagination with a defined maximum page size
• [ ] Define file upload limits (size, type, count) for every upload endpoint
• [ ] Standardize error response format (use RFC 7807 Problem Details)
• [ ] Ensure error responses never expose stack traces, internal paths, or library versions
• [ ] Define explicit Content-Type for all responses (e.g., application/json; charset=utf-8)

API Lifecycle and Architecture

• [ ] Document API versioning strategy
• [ ] Identify sensitive business flows and protect them against automation abuse
• [ ] Limit GraphQL query depth and complexity; disable introspection in production
• [ ] Configure CORS with explicit origins (never use wildcards with credentials)
• [ ] Configure security headers (HSTS, X-Content-Type-Options, X-Frame-Options, etc.)
• [ ] Define and enforce request/response schemas for all inter-service communication
• [ ] Establish a security review gate in the API design approval process

---

2. Authentication and Authorization Checklist

Verify identity and access controls at every layer of the API.

Authentication Mechanisms

• [ ] Use OAuth 2.0 / OpenID Connect for user authentication
• [ ] Enforce JWT algorithm explicitly (RS256/ES256 — reject "none" and HS256 with public keys)
• [ ] Set JWT expiry to a reasonable duration (access tokens: 15–60 minutes)
• [ ] Implement refresh token rotation (invalidate old refresh tokens on use)
• [ ] Bind refresh tokens to the client (device fingerprint or client ID)
• [ ] Hash API keys in storage (never store plaintext)
• [ ] Reject API keys passed in URL query parameters
• [ ] Use mTLS or signed tokens for service-to-service authentication
• [ ] Validate the aud (audience) and iss (issuer) claims in every JWT

Authentication Security

• [ ] Return generic errors on authentication failure — do not reveal whether a user exists
• [ ] Use timing-safe comparison for credential validation
• [ ] Implement account lockout or progressive delays after repeated failed attempts
• [ ] Invalidate all sessions and tokens on logout or password change
• [ ] Require step-up authentication for sensitive operations (e.g., MFA re-prompt)
• [ ] Set absolute session timeout in addition to idle timeout
• [ ] Reject tokens issued before the last password change (check iat claim)

Authorization Controls

• [ ] Enforce RBAC/ABAC policies at the middleware level (not in individual handlers)
• [ ] Test authorization with multiple roles: admin, regular user, anonymous, service account

---

3. Input Validation Checklist

Validate and sanitize all input at the API boundary to prevent injection and abuse.

Header and Body Validation

• [ ] Validate the Content-Type header — reject unexpected content types
• [ ] Limit request body size at the server/gateway level
• [ ] Validate all path parameters (type, format, range)
• [ ] Validate all query parameters — use allowlists where possible
• [ ] Apply JSON schema validation on all request bodies

Injection Prevention

• [ ] Prevent SQL/NoSQL injection — never use string interpolation for queries
• [ ] Prevent command injection — never pass unsanitized input to shell commands
• [ ] Prevent XXE — disable XML external entity processing or use defusedxml
• [ ] Prevent LDAP injection — escape special characters in LDAP queries
• [ ] Prevent template injection — never pass user input directly to template engines

File and URL Handling

• [ ] Restrict file uploads: enforce type whitelist, size limit, sanitize filenames, store outside webroot
• [ ] Validate URL parameters against SSRF — block private/internal IPs, validate URL schemes
• [ ] Enforce GraphQL-specific limits: query depth, complexity, and batch query count

---

4. Transport and Data Security Checklist

Protect data in transit and at rest across all API communications.

Transport Security

• [ ] Enforce TLS 1.2 or higher on all endpoints
• [ ] Disable weak cipher suites (RC4, DES, 3DES, export ciphers)
• [ ] Set HSTS header with max-age ≥ 31536000 (one year)
• [ ] Never include sensitive data in URL query parameters (tokens, passwords, PII)
• [ ] Pin certificates or use Certificate Transparency monitoring for critical services

Data Protection

• [ ] Encrypt sensitive data at rest
• [ ] Classify API data fields by sensitivity level (public, internal, confidential, restricted)
• [ ] Minimize PII in API responses (apply data minimization principle)
• [ ] Verify webhook signatures using HMAC or shared secret validation
• [ ] Implement field-level encryption for highly sensitive data (e.g., SSN, payment card numbers)

Outbound API Call Security

• [ ] Validate third-party API responses before processing
• [ ] Disable automatic HTTP redirect following for upstream API calls
• [ ] Set explicit timeouts on all outbound API calls
• [ ] Enable certificate validation on all outbound HTTPS calls

---

5. Rate Limiting and Resource Protection Checklist

Prevent abuse, denial-of-service, and resource exhaustion across all API surfaces.

Rate Limiting Configuration

• [ ] Configure global rate limiting (requests per minute per client)
• [ ] Apply stricter limits on authentication endpoints (login, password reset, OTP verification)
• [ ] Apply stricter limits on sensitive business flows (purchase, transfer, invitation)
• [ ] Return rate limit headers to clients (X-RateLimit-Limit, X-RateLimit-Remaining, X-RateLimit-Reset)
• [ ] Return 429 Too Many Requests with a Retry-After header when limits are exceeded

Resource Constraints

• [ ] Enforce response pagination with a maximum page size
• [ ] Apply query complexity limits for GraphQL APIs
• [ ] Enforce upload size limits at the server/gateway level
• [ ] Set execution timeouts on long-running operations
• [ ] Configure spending limits and alerts on third-party API integrations

---

6. API Inventory and Documentation Checklist

Maintain a complete, accurate inventory of all APIs and their security posture.

Documentation and Specification

• [ ] Document all APIs in an OpenAPI/Swagger specification
• [ ] Generate API documentation from code to ensure it stays in sync
• [ ] Assign sunset dates and add deprecation headers to deprecated APIs
• [ ] Document the API version retirement plan
• [ ] Include security requirements (scopes, roles, auth methods) in API specification
• [ ] Document data classification for every endpoint's request and response fields

API Inventory Management

• [ ] Audit for and remove undocumented shadow/zombie endpoints in production
• [ ] Apply identical security controls to beta/staging APIs as production APIs
• [ ] Verify internal APIs are not exposed to the public network
• [ ] Maintain a registry of all API consumers and their access scopes
• [ ] Review and rotate API keys/credentials on a defined schedule

Dependency Tracking

• [ ] Inventory all third-party API integrations with data flow documentation
• [ ] Include API dependencies and client libraries in the SBOM

---

7. Deployment and Infrastructure Checklist

Harden the runtime environment and infrastructure supporting the API.

Gateway and Network Security

• [ ] Configure the API gateway for auth offloading, rate limiting, and request logging
• [ ] Activate WAF rules for common API attacks (injection, SSRF, path traversal)
• [ ] Disable debug and diagnostic endpoints in production
• [ ] Change all default credentials before deployment
• [ ] Disable unnecessary HTTP methods (TRACE, OPTIONS unless required by CORS)
• [ ] Remove server version banners from HTTP response headers (Server, X-Powered-By)

Container and Runtime Security

• [ ] Scan container images for known vulnerabilities before deployment
• [ ] Run the API process with minimal privileges (non-root user)
• [ ] Enforce network segmentation between API tiers (frontend, backend, database)
• [ ] Use read-only filesystems where possible in container deployments
• [ ] Define resource limits (CPU, memory) for API containers to prevent resource abuse

Secrets Management

• [ ] Inject secrets via environment variables or a secrets vault (never embed in code or config files)
• [ ] Verify health check endpoints do not expose sensitive information (version, config, env)
• [ ] Rotate secrets and credentials on a defined schedule
• [ ] Audit access to secrets vault and log secret retrieval events

---

8. Logging and Monitoring Checklist

Establish visibility into API activity for detection, response, and forensics.

Request Logging

• [ ] Log all API requests with method, path, status code, latency, and user ID
• [ ] Log all authentication events (success, failure, lockout)
• [ ] Log all authorization failures with request context
• [ ] Exclude sensitive data from logs (tokens, passwords, PII, full request bodies)
• [ ] Track correlation IDs across services for distributed tracing
• [ ] Log API key usage per consumer for audit and anomaly detection

Alerting and Response

• [ ] Configure alerting on anomalies (error rate spikes, auth failure bursts, unusual traffic patterns)
• [ ] Alert on sudden changes in API usage patterns per consumer
• [ ] Protect log integrity (use append-only storage, ship to external log aggregation)
• [ ] Define log retention policy that meets compliance requirements
• [ ] Document an incident response plan specific to API breaches
• [ ] Include API-specific tests in penetration testing scope
• [ ] Conduct periodic tabletop exercises for API security incidents

---

9. Testing Security Checklist

Verify security controls through automated and manual testing at every stage.

Authorization Testing

• [ ] Test every endpoint with an incorrect user/role — verify 403 Forbidden response
• [ ] Test BOLA: attempt to access resources by manipulating IDs with a different user
• [ ] Test BFLA: attempt to access admin endpoints as a regular user
• [ ] Test mass assignment: send extra/unexpected fields and verify they are rejected or ignored
• [ ] Test horizontal privilege escalation: access peer user resources across tenant boundaries

Abuse and Limit Testing

• [ ] Test rate limits: exceed configured limits and verify 429 response
• [ ] Test injection payloads: SQL, NoSQL, and command injection vectors
• [ ] Test SSRF: submit internal IP addresses and cloud metadata URLs (e.g., 169.254.169.254)
• [ ] Test large payload handling: submit oversized request bodies and verify rejection
• [ ] Test pagination boundary conditions: request page size 0, negative, and extremely large values

Authentication and Error Handling Testing

• [ ] Test authentication bypass: submit missing, expired, and malformed tokens
• [ ] Test token reuse after logout or password change — verify rejection
• [ ] Test error responses: verify no sensitive information is leaked in error bodies or headers
• [ ] Verify consistent error format across all endpoints (RFC 7807 compliance)
• [ ] Test CORS: verify preflight responses reject unauthorized origins

Automated Security Testing

• [ ] Integrate SAST and DAST tools in the CI/CD pipeline
• [ ] Run API fuzzing with dedicated tools (Burp Suite, OWASP ZAP, Schemathesis)
• [ ] Validate API contract compliance against the OpenAPI specification in CI
• [ ] Run dependency vulnerability scans (Dependabot, Snyk, Trivy) on every build
• [ ] Maintain a baseline of known security test results and track regressions

---

10. Security Tools Reference

Recommended tools for API security testing, enforcement, and monitoring.

Tool Selection Guidelines

• Use Schemathesis or dredd for automated contract and property-based testing in CI/CD
• Use OWASP ZAP for automated DAST scans; use Burp Suite for manual penetration testing
• Use Spectral as a pre-commit or CI lint gate for OpenAPI specification quality
• Use nuclei for broad vulnerability scanning across multiple API endpoints
• Choose rate limiting middleware based on the application framework and deployment architecture

Additional Tools by Category

Static Analysis:

• Use semgrep with API-security rulesets for language-specific SAST
• Use bandit (Python) or eslint-plugin-security (Node.js) for framework-specific checks

Runtime Protection:

• Use ModSecurity or cloud-native WAF for runtime API attack mitigation
• Use Falco for runtime container security monitoring

API Discovery:

• Use kiterunner for discovering hidden API endpoints during penetration testing
• Use Akto or Salt Security for continuous API discovery and posture management

---

Quick Reference: OWASP API Security Top 10 (2023) Mapping

---

Usage Notes

• Apply checklists incrementally — prioritize items based on threat model and risk assessment
• Treat each checkbox as a gate: mark complete only when the control is verified and documented
• Review these checklists at every major milestone: design review, code review, pre-deployment, and periodic audit
• Update checklists as new threats emerge and standards evolve (reference OWASP API Security Top 10)

OWASP API Security Top 10 (2023) — Reference

Agent reference for secure API development. All 10 categories with vulnerable/secure

code patterns. Use imperative form. Language-agnostic principles with practical examples.

Table of Contents

• API1:2023 — Broken Object Level Authorization (BOLA)
• API2:2023 — Broken Authentication
• API3:2023 — Broken Object Property Level Authorization
• API4:2023 — Unrestricted Resource Consumption
• API5:2023 — Broken Function Level Authorization (BFLA)
• API6:2023 — Unrestricted Access to Sensitive Business Flows
• API7:2023 — Server Side Request Forgery (SSRF)
• API8:2023 — Security Misconfiguration
• API9:2023 — Improper Inventory Management
• API10:2023 — Unsafe Consumption of APIs

---

API1:2023 — Broken Object Level Authorization (BOLA)

The #1 API security risk. Attackers manipulate object IDs in API requests to access resources belonging to other users. APIs expose endpoints that handle object identifiers, creating a wide attack surface for access control issues.

API-Specific Risks

• Sequential/predictable IDs allow enumeration (/api/orders/1001, /api/orders/1002)
• Missing ownership checks at data access layer
• GraphQL node queries bypassing REST-layer authorization
• Batch/list endpoints leaking other users' objects
• Nested resource access (/users/123/documents/456) skipping parent ownership verification

Vulnerable Example

# FastAPI — no ownership check
@app.get("/api/orders/{order_id}")
async def get_order(order_id: int):
    order = db.query(Order).filter(Order.id == order_id).first()
    if not order:
        raise HTTPException(status_code=404)
    return order  # Any authenticated user can access any order

// Express — no ownership check
app.get("/api/orders/:orderId", authenticate, async (req, res) => {
  const order = await Order.findById(req.params.orderId);
  if (!order) return res.status(404).json({ error: "Not found" });
  res.json(order); // Any authenticated user can access any order
});

# GraphQL — no ownership check in resolver
query {
  order(id: "order-belonging-to-another-user") {
    id totalAmount items { name price }
  }
}

# Attack: enumerate sequential IDs
curl -H "Authorization: Bearer <user_a_token>" https://api.example.com/api/orders/1001
curl -H "Authorization: Bearer <user_a_token>" https://api.example.com/api/orders/1002

Secure Example

# FastAPI — enforce ownership at data access layer
@app.get("/api/orders/{order_id}")
async def get_order(order_id: uuid.UUID, current_user: User = Depends(get_current_user)):
    order = db.query(Order).filter(
        Order.id == order_id,
        Order.user_id == current_user.id  # ownership check
    ).first()
    if not order:
        raise HTTPException(status_code=404)  # 404, not 403 — avoid confirming existence
    return OrderResponse.model_validate(order)

// Express — enforce ownership
app.get("/api/orders/:orderId", authenticate, async (req, res) => {
  const order = await Order.findOne({
    _id: req.params.orderId,
    userId: req.user.id, // ownership check
  });
  if (!order) return res.status(404).json({ error: "Not found" });
  res.json(order);
});

# GraphQL (Strawberry) — ownership in resolver
@strawberry.type
class Query:
    @strawberry.field
    async def order(self, info: Info, id: strawberry.ID) -> Order:
        user = info.context.user
        order = await Order.objects.filter(id=id, user_id=user.id).first()
        if not order:
            raise NotFoundException("Order not found")
        return order

Prevention Strategies

• Enforce authorization checks per object at the data access layer, not at the routing layer
• Use UUIDs/GUIDs instead of sequential integer IDs
• Return 404 Not Found instead of 403 Forbidden to avoid confirming resource existence
• Write authorization tests: "User A must not access User B's resources"
• Implement a centralized authorization service or policy engine (OPA, Casbin)
• Add automated BOLA detection in integration tests

---

API2:2023 — Broken Authentication

Weak or improperly implemented authentication mechanisms allow attackers to assume other users' identities. APIs are especially vulnerable because authentication tokens are bearer credentials — whoever holds the token holds the identity.

API-Specific Risks

• Missing authentication on endpoints (assumed "internal" APIs)
• Weak JWT implementation: no algorithm enforcement, alg: none accepted, weak signing secrets
• Long-lived tokens without rotation or revocation
• API keys in URLs (logged in proxies, browser history, referer headers)
• No rate limiting on authentication endpoints enabling credential stuffing
• Credential leakage in client-side code or public repositories

Vulnerable Example

# Accepting any JWT algorithm — algorithm confusion attack
import jwt

def verify_token(token: str):
    # VULNERABLE: attacker can switch to HS256 using the public RSA key as secret
    payload = jwt.decode(token, PUBLIC_KEY, algorithms=["RS256", "HS256"])
    return payload

// Express — weak JWT, no expiry check
app.use((req, res, next) => {
  const token = req.headers.authorization?.split(" ")[1];
  // VULNERABLE: no algorithm enforcement, using weak secret
  const payload = jwt.verify(token, "secret123");
  req.user = payload;
  next();
});

# API key leaked in URL — visible in server logs, proxy logs, referer headers
curl "https://api.example.com/data?api_key=sk_live_abc123xyz"

Secure Example

# FastAPI — strict JWT validation with proper configuration
from jose import jwt, JWTError

ALGORITHM = "RS256"  # enforce single algorithm
JWKS_URL = "https://auth.example.com/.well-known/jwks.json"

async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    try:
        jwks = await fetch_jwks(JWKS_URL)
        payload = jwt.decode(
            token,
            jwks,
            algorithms=[ALGORITHM],  # strict algorithm enforcement
            audience="https://api.example.com",
            issuer="https://auth.example.com",
        )
        if payload.get("exp", 0) < time.time():
            raise HTTPException(status_code=401, detail="Token expired")
        user = await get_user_by_sub(payload["sub"])
        if not user:
            raise HTTPException(status_code=401)
        return user
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

// Express — OAuth2 with JWKS validation
const { expressjwt: jwtMiddleware } = require("express-jwt");
const jwksRsa = require("jwks-rsa");

const authenticate = jwtMiddleware({
  secret: jwksRsa.expressJwtSecret({
    jwksUri: "https://auth.example.com/.well-known/jwks.json",
    cache: true,
    rateLimit: true,
  }),
  audience: "https://api.example.com",
  issuer: "https://auth.example.com",
  algorithms: ["RS256"], // strict algorithm enforcement
});

# API key in header, not URL
curl -H "Authorization: Bearer sk_live_abc123xyz" https://api.example.com/data

Prevention Strategies

• Use established auth standards: OAuth 2.0, OpenID Connect
• Enforce a single signing algorithm in JWT validation — never accept alg: none
• Use short-lived access tokens (5–15 min) with refresh token rotation
• Send API keys in headers (Authorization, X-API-Key), never in URLs
• Rate limit authentication endpoints (login, token refresh, password reset)
• Implement token revocation (blocklist or short expiry + refresh)
• Store secrets in environment variables or secret managers, never in code
• Use strong, unique signing secrets (≥256 bits for HMAC, ≥2048-bit RSA)

---

API3:2023 — Broken Object Property Level Authorization

Combines "Excessive Data Exposure" and "Mass Assignment." APIs expose more object properties than the client needs, or accept properties the client should not be able to set. This leads to data leaks and privilege escalation via property manipulation.

API-Specific Risks

• Returning full database objects via to_dict(), to_json(), serialize()
• Exposing internal fields: password_hash, is_admin, internal_notes, ssn
• Mass assignment: accepting arbitrary fields on create/update (e.g., setting role: "admin")
• GraphQL introspection revealing all fields, including sensitive ones

Vulnerable Example

# Django REST — returning full model object
class UserViewSet(viewsets.ModelViewSet):
    queryset = User.objects.all()
    serializer_class = UserSerializer

class UserSerializer(serializers.ModelSerializer):
    class Meta:
        model = User
        fields = "__all__"  # VULNERABLE: exposes password_hash, is_admin, ssn

# Flask — mass assignment via dict unpacking
@app.put("/api/users/<user_id>")
def update_user(user_id):
    user = User.query.get_or_404(user_id)
    data = request.get_json()
    for key, value in data.items():
        setattr(user, key, value)  # VULNERABLE: attacker can set is_admin=True
    db.session.commit()
    return jsonify(user.to_dict())

# Attack: mass assignment — escalate to admin
curl -X PUT https://api.example.com/api/users/me \
  -H "Authorization: Bearer <token>" \
  -H "Content-Type: application/json" \
  -d '{"name": "Alice", "role": "admin", "is_verified": true}'

Secure Example

# FastAPI + Pydantic — explicit response and update schemas
class UserResponse(BaseModel):
    id: uuid.UUID
    name: str
    email: str
    # Exclude: password_hash, is_admin, ssn, internal_notes

class UserUpdate(BaseModel):
    name: str | None = None
    email: str | None = None
    # Only allow specific fields — role, is_admin are excluded

@app.put("/api/users/me", response_model=UserResponse)
async def update_user(
    updates: UserUpdate,
    current_user: User = Depends(get_current_user),
):
    update_data = updates.model_dump(exclude_unset=True)
    for field, value in update_data.items():
        setattr(current_user, field, value)
    db.commit()
    return UserResponse.model_validate(current_user)

// Express — allowlisted fields for update and response
const ALLOWED_UPDATE_FIELDS = ["name", "email", "avatar"];
const RESPONSE_FIELDS = ["id", "name", "email", "avatar", "createdAt"];

app.put("/api/users/me", authenticate, async (req, res) => {
  const updates = {};
  for (const field of ALLOWED_UPDATE_FIELDS) {
    if (req.body[field] !== undefined) {
      updates[field] = req.body[field];
    }
  }
  const user = await User.findByIdAndUpdate(req.user.id, updates, { new: true })
    .select(RESPONSE_FIELDS.join(" "));
  res.json(user);
});

Prevention Strategies

• Define explicit response schemas — cherry-pick returned fields, never use fields = "__all__"
• Define explicit input schemas with allowlisted fields for create/update operations
• Use schema-based validation (Pydantic, marshmallow, Joi, Zod) for both input and output
• Block mass assignment: never spread/unpack request body directly into model updates
• Disable GraphQL introspection in production or restrict to authorized roles
• Review API responses in CI for unintended field exposure

---

API4:2023 — Unrestricted Resource Consumption

APIs that do not limit resource consumption enable denial-of-service, financial drain, and data exfiltration via excessive requests. Without rate limits, pagination caps, and complexity controls, attackers can overwhelm infrastructure or extract large datasets.

API-Specific Risks

• No rate limiting per user, IP, or endpoint
• Unlimited pagination (?page_size=999999) or missing pagination entirely
• No file upload size limits
• GraphQL: unbounded query depth, breadth, and batching
• No spending caps on pay-per-use APIs (SMS, email, AI inference)
• Regex denial of service (ReDoS) via crafted input

Vulnerable Example

# FastAPI — no pagination limit
@app.get("/api/users")
async def list_users(page_size: int = 10):
    # VULNERABLE: attacker can request page_size=1000000
    return db.query(User).limit(page_size).all()

# GraphQL — deeply nested query (unbounded depth)
query {
  users {
    friends {
      friends {
        friends {
          friends { id name email }
        }
      }
    }
  }
}

# Attack: request excessive data
curl "https://api.example.com/api/users?page_size=1000000"

# Attack: GraphQL batching — send 1000 queries in one request
curl -X POST https://api.example.com/graphql \
  -H "Content-Type: application/json" \
  -d '[{"query":"{ user(id:1) { email } }"},{"query":"{ user(id:2) { email } }"},...]'

Secure Example

# FastAPI — enforce pagination caps and rate limiting
from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)

MAX_PAGE_SIZE = 100

@app.get("/api/users")
@limiter.limit("30/minute")
async def list_users(
    request: Request,
    page: int = Query(ge=1, default=1),
    page_size: int = Query(ge=1, le=MAX_PAGE_SIZE, default=20),
):
    offset = (page - 1) * page_size
    users = db.query(User).offset(offset).limit(page_size).all()
    total = db.query(User).count()
    return {
        "data": [UserResponse.model_validate(u) for u in users],
        "pagination": {"page": page, "page_size": page_size, "total": total},
    }

// Express — rate limiting middleware
const rateLimit = require("express-rate-limit");

const apiLimiter = rateLimit({
  windowMs: 15 * 60 * 1000, // 15 minutes
  max: 100, // limit each IP to 100 requests per window
  standardHeaders: true,
  legacyHeaders: false,
  message: { error: "Too many requests, try again later" },
});

app.use("/api/", apiLimiter);

// Strict pagination
app.get("/api/users", authenticate, async (req, res) => {
  const page = Math.max(1, parseInt(req.query.page) || 1);
  const pageSize = Math.min(100, Math.max(1, parseInt(req.query.page_size) || 20));
  const skip = (page - 1) * pageSize;
  const [users, total] = await Promise.all([
    User.find().skip(skip).limit(pageSize),
    User.countDocuments(),
  ]);
  res.json({ data: users, pagination: { page, pageSize, total } });
});

# GraphQL — depth and complexity limiting (Strawberry + graphql-core)
from graphql import validate
from graphql.validation import ASTValidationRule

MAX_DEPTH = 5
MAX_ALIASES = 10

schema = strawberry.Schema(
    query=Query,
    extensions=[
        QueryDepthLimiter(max_depth=MAX_DEPTH),
        MaxAliasesLimiter(max_aliases=MAX_ALIASES),
    ],
)

Prevention Strategies

• Enforce rate limiting per user/IP/API key — use sliding window or token bucket algorithms
• Cap pagination: enforce maximum page_size (e.g., 100), require cursor-based pagination for large datasets
• Limit file upload size at the reverse proxy and application layer
• GraphQL: limit query depth (5–7), complexity scoring, disable batching or cap batch size
• Set execution timeouts for API requests (30s max for typical endpoints)
• Implement spending alerts and caps on pay-per-use downstream services
• Use streaming/chunked responses for large payloads

---

API5:2023 — Broken Function Level Authorization (BFLA)

Attackers access administrative or privileged functions by directly calling API endpoints. APIs tend to expose more endpoints than web apps, making it critical to enforce function-level authorization. Relying on client-side UI hiding is insufficient.

API-Specific Risks

• Regular users accessing admin endpoints (/api/admin/users, DELETE /api/users/{id})
• Horizontal escalation: accessing same-level functions of another role
• HTTP method tampering: GET allowed but PUT/DELETE not checked
• Predictable admin URL patterns (/api/v1/admin/*, /api/internal/*)
• Missing role checks on sensitive operations (delete, export, bulk operations)

Vulnerable Example

# FastAPI — no role check on admin endpoint
@app.delete("/api/admin/users/{user_id}")
async def delete_user(user_id: uuid.UUID, current_user: User = Depends(get_current_user)):
    # VULNERABLE: any authenticated user can delete users
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404)
    db.delete(user)
    db.commit()
    return {"status": "deleted"}

# Attack: regular user calls admin endpoint
curl -X DELETE https://api.example.com/api/admin/users/some-uuid \
  -H "Authorization: Bearer <regular_user_token>"

Secure Example

# FastAPI — centralized RBAC decorator
from functools import wraps

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

@app.delete("/api/admin/users/{user_id}")
@require_role("admin", "superadmin")
async def delete_user(user_id: uuid.UUID, current_user: User = Depends(get_current_user)):
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404)
    db.delete(user)
    db.commit()
    return {"status": "deleted"}

// Express — role-based middleware
function requireRole(...roles) {
  return (req, res, next) => {
    if (!roles.includes(req.user.role)) {
      return res.status(403).json({ error: "Insufficient permissions" });
    }
    next();
  };
}

// Apply to admin routes
app.delete("/api/admin/users/:userId", authenticate, requireRole("admin"), async (req, res) => {
  await User.findByIdAndDelete(req.params.userId);
  res.json({ status: "deleted" });
});

// Apply to entire admin router
const adminRouter = express.Router();
adminRouter.use(authenticate, requireRole("admin"));
adminRouter.delete("/users/:userId", deleteUserHandler);
app.use("/api/admin", adminRouter);

Prevention Strategies

• Deny by default: require explicit authorization for every endpoint
• Enforce RBAC or ABAC server-side — never rely on client-side UI hiding
• Centralize authorization logic in middleware or policy engine
• Group admin endpoints under a common prefix with shared auth middleware
• Test authorization with multiple roles: admin, regular user, unauthenticated
• Audit: log all access to privileged functions
• Review: ensure every HTTP method on every route has explicit authorization

---

API6:2023 — Unrestricted Access to Sensitive Business Flows

Attackers automate access to business-critical flows (purchasing, account creation, referral programs) to cause harm at scale. This is not a traditional technical vulnerability — it is business logic abuse through API automation.

API-Specific Risks

• Ticket/inventory scalping via automated purchasing
• Credential stuffing and account takeover at scale
• Referral/coupon/bonus abuse through automated account creation
• Comment/review spam via API automation
• Mass data scraping exceeding intended use
• Automated bidding or price manipulation

Vulnerable Example

# FastAPI — purchase endpoint with no anti-automation
@app.post("/api/purchases")
async def purchase_item(
    item_id: uuid.UUID,
    quantity: int,
    current_user: User = Depends(get_current_user),
):
    # VULNERABLE: no per-user rate limit, no CAPTCHA, no device fingerprint
    item = db.query(Item).filter(Item.id == item_id).first()
    if item.stock < quantity:
        raise HTTPException(status_code=400, detail="Out of stock")
    item.stock -= quantity
    order = Order(user_id=current_user.id, item_id=item_id, quantity=quantity)
    db.add(order)
    db.commit()
    return {"order_id": order.id}

Secure Example

# FastAPI — purchase with anti-automation protections
from slowapi import Limiter

limiter = Limiter(key_func=lambda req: req.state.user.id)

@app.post("/api/purchases")
@limiter.limit("5/minute")  # per-user rate limit on purchases
async def purchase_item(
    request: Request,
    purchase: PurchaseRequest,
    current_user: User = Depends(get_current_user),
):
    # Verify CAPTCHA for high-value or suspicious transactions
    if purchase.quantity > 2 or await is_suspicious(current_user):
        await verify_captcha(purchase.captcha_token)

    # Check per-user purchase limits
    recent_purchases = await count_recent_purchases(current_user.id, hours=24)
    if recent_purchases + purchase.quantity > MAX_DAILY_PURCHASE:
        raise HTTPException(status_code=429, detail="Daily purchase limit reached")

    # Device fingerprint validation
    await validate_device_fingerprint(request, current_user)

    item = db.query(Item).with_for_update().filter(Item.id == purchase.item_id).first()
    if item.stock < purchase.quantity:
        raise HTTPException(status_code=400, detail="Out of stock")

    item.stock -= purchase.quantity
    order = Order(user_id=current_user.id, item_id=purchase.item_id, quantity=purchase.quantity)
    db.add(order)
    db.commit()
    return {"order_id": order.id}

Prevention Strategies

• Identify business-critical flows and apply targeted protections
• Rate limit per user per business action (not just per IP)
• Implement CAPTCHA or proof-of-work challenges for sensitive operations
• Use device fingerprinting and behavioral analysis to detect bots
• Set per-user and per-time-window limits on business actions (purchases, signups, referrals)
• Monitor for anomalous patterns: burst activity, headless browser signatures, unusual timing
• Require step-up authentication (MFA) for high-value transactions

---

API7:2023 — Server Side Request Forgery (SSRF)

APIs that fetch user-supplied URLs without validation allow attackers to make the server send requests to unintended destinations. Common in webhook handlers, URL preview features, and file-import-by-URL functionality. Particularly dangerous in cloud environments.

API-Specific Risks

• Accessing cloud metadata services (http://169.254.169.254/latest/meta-data/)
• Scanning internal networks and services via API as proxy
• Reading internal files via file:// protocol
• Webhook registration pointing to internal services
• URL preview/unfurl features fetching arbitrary URLs
• Bypassing firewalls by making the server initiate the connection

Vulnerable Example

# FastAPI — webhook URL fetched without validation
import httpx

@app.post("/api/webhooks")
async def register_webhook(url: str, current_user: User = Depends(get_current_user)):
    # VULNERABLE: no URL validation, can access internal services
    response = await httpx.get(url)  # attacker sends url=http://169.254.169.254/latest/meta-data/
    return {"status": "registered", "test_response": response.status_code}

# Attack: access cloud metadata
curl -X POST https://api.example.com/api/webhooks \
  -H "Authorization: Bearer <token>" \
  -H "Content-Type: application/json" \
  -d '{"url": "http://169.254.169.254/latest/meta-data/iam/security-credentials/"}'

# Attack: scan internal network
curl -X POST https://api.example.com/api/webhooks \
  -d '{"url": "http://192.168.1.1:8080/admin"}'

Secure Example

# FastAPI — URL validation with allowlist and IP blocking
import ipaddress
from urllib.parse import urlparse
import socket

ALLOWED_SCHEMES = {"https"}
BLOCKED_IP_RANGES = [
    ipaddress.ip_network("10.0.0.0/8"),
    ipaddress.ip_network("172.16.0.0/12"),
    ipaddress.ip_network("192.168.0.0/16"),
    ipaddress.ip_network("127.0.0.0/8"),
    ipaddress.ip_network("169.254.0.0/16"),  # link-local / cloud metadata
    ipaddress.ip_network("::1/128"),
]

def validate_url(url: str) -> str:
    parsed = urlparse(url)

    if parsed.scheme not in ALLOWED_SCHEMES:
        raise ValueError(f"Scheme '{parsed.scheme}' not allowed. Use HTTPS.")

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

    # Resolve hostname to IP and check against blocked ranges
    try:
        resolved_ips = socket.getaddrinfo(parsed.hostname, None)
    except socket.gaierror:
        raise ValueError("Cannot resolve hostname")

    for family, _, _, _, addr in resolved_ips:
        ip = ipaddress.ip_address(addr[0])
        for blocked in BLOCKED_IP_RANGES:
            if ip in blocked:
                raise ValueError(f"URL resolves to blocked IP range")

    return url

@app.post("/api/webhooks")
async def register_webhook(
    webhook: WebhookRequest,
    current_user: User = Depends(get_current_user),
):
    validated_url = validate_url(webhook.url)
    async with httpx.AsyncClient(
        follow_redirects=False,  # prevent redirect-based SSRF bypass
        timeout=5.0,
    ) as client:
        response = await client.get(validated_url)
    return {"status": "registered", "test_response": response.status_code}

// Express — SSRF protection
const { URL } = require("url");
const dns = require("dns").promises;
const ipRangeCheck = require("ip-range-check");

const BLOCKED_RANGES = [
  "10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16",
  "127.0.0.0/8", "169.254.0.0/16", "::1/128",
];

async function validateUrl(urlString) {
  const parsed = new URL(urlString);
  if (parsed.protocol !== "https:") throw new Error("Only HTTPS allowed");

  const { address } = await dns.lookup(parsed.hostname);
  if (ipRangeCheck(address, BLOCKED_RANGES)) {
    throw new Error("URL resolves to blocked IP range");
  }
  return urlString;
}

Prevention Strategies

• Validate and sanitize all user-supplied URLs
• Allowlist URL schemes (https only) — block file://, gopher://, ftp://
• Resolve hostnames and verify the resolved IP is not in private/reserved ranges
• Disable HTTP redirects or re-validate the destination after each redirect
• Use a dedicated egress proxy for outbound requests with allowlisted destinations
• Set strict timeouts on outbound requests
• Block access to cloud metadata IPs (169.254.169.254) at the network level
• Run webhook/URL-fetching services in isolated network segments

---

API8:2023 — Security Misconfiguration

Broad category covering misconfigurations at any stack layer: security headers, CORS, error handling, TLS, debug endpoints, default credentials. APIs are particularly susceptible due to the many configuration surfaces across API gateways, frameworks, and cloud services.

API-Specific Risks

• CORS misconfiguration: Access-Control-Allow-Origin: * with credentials
• Verbose error messages exposing stack traces, SQL queries, internal paths
• Debug/profiling endpoints left enabled in production
• Default credentials on API gateways, admin panels, databases
• Unnecessary HTTP methods enabled (TRACE, OPTIONS leaking info)
• Missing security headers (Strict-Transport-Security, X-Content-Type-Options)
• TLS misconfiguration: outdated protocols, weak cipher suites, missing HSTS
• Permissive Content-Type handling leading to deserialization attacks

Vulnerable Example

# Flask — verbose errors + wide CORS
from flask_cors import CORS

app = Flask(__name__)
app.config["DEBUG"] = True  # VULNERABLE: debug mode in production
CORS(app, origins="*", supports_credentials=True)  # VULNERABLE: wildcard with credentials

@app.errorhandler(500)
def handle_error(e):
    return jsonify({
        "error": str(e),
        "traceback": traceback.format_exc(),  # VULNERABLE: stack trace exposed
        "sql_query": str(e.statement) if hasattr(e, "statement") else None,
    }), 500

// Express — no security headers, verbose errors
app.use((err, req, res, next) => {
  // VULNERABLE: full stack trace in production
  res.status(500).json({
    message: err.message,
    stack: err.stack,
    query: err.sql,
  });
});

Secure Example

# Flask — hardened CORS + minimal errors + security headers
from flask_cors import CORS
from flask_talisman import Talisman

app = Flask(__name__)
app.config["DEBUG"] = False

# Strict CORS — explicit origin allowlist
CORS(app, origins=["https://app.example.com"], supports_credentials=True)

# Security headers via Talisman
Talisman(
    app,
    force_https=True,
    strict_transport_security=True,
    strict_transport_security_max_age=31536000,
    content_security_policy={"default-src": "'self'"},
)

@app.errorhandler(500)
def handle_error(e):
    app.logger.error(f"Internal error: {e}", exc_info=True)  # log internally
    return jsonify({"error": "Internal server error"}), 500  # minimal response

// Express — security headers via helmet, minimal errors
const helmet = require("helmet");

app.use(helmet());
app.use(helmet.hsts({ maxAge: 31536000, includeSubDomains: true, preload: true }));

// CORS — explicit allowlist
const cors = require("cors");
app.use(cors({
  origin: ["https://app.example.com"],
  credentials: true,
  methods: ["GET", "POST", "PUT", "DELETE"],
  allowedHeaders: ["Content-Type", "Authorization"],
}));

// Production error handler — no stack traces
app.use((err, req, res, next) => {
  console.error("Internal error:", err);
  res.status(500).json({ error: "Internal server error" });
});

# Verify security headers
curl -I https://api.example.com/api/health
# Expected headers:
# Strict-Transport-Security: max-age=31536000; includeSubDomains; preload
# X-Content-Type-Options: nosniff
# X-Frame-Options: DENY
# Content-Security-Policy: default-src 'self'
# Cache-Control: no-store

Prevention Strategies

• Disable debug/verbose modes in production — use environment-based configuration
• Configure CORS with explicit origin allowlists — never * with credentials
• Return minimal error messages to clients — log full details server-side
• Set security headers: HSTS, X-Content-Type-Options, X-Frame-Options, CSP
• Disable unnecessary HTTP methods (TRACE, CONNECT)
• Remove or protect debug/health/metrics endpoints in production
• Automate security configuration checks in CI/CD (e.g., check-headers, tfsec)
• Rotate default credentials and API gateway keys before deployment
• Enforce TLS 1.2+ with strong cipher suites

---

API9:2023 — Improper Inventory Management

Organizations lose track of which APIs exist, which versions are running, and what data flows to third parties. Shadow APIs, deprecated-but-still-running endpoints, and missing documentation create blind spots that attackers exploit.

API-Specific Risks

• Shadow APIs: undocumented endpoints deployed by teams without security review
• Zombie APIs: deprecated versions still accessible (/api/v1/ alongside /api/v3/)
• Beta/staging APIs with weaker security controls exposed to the internet
• Missing or outdated API documentation (no OpenAPI spec, stale docs)
• Third-party API integrations without data flow inventory
• Internal APIs accidentally exposed via misconfigured API gateways

Vulnerable Example

# Deprecated v1 still running — weaker auth, no rate limiting
curl https://api.example.com/api/v1/users  # no auth required (legacy)
curl https://api.example.com/api/v2/users -H "Authorization: Bearer <token>"  # current

# Staging API accessible publicly
curl https://staging-api.example.com/api/users  # no auth, debug enabled

# Undocumented endpoint found via directory enumeration
curl https://api.example.com/api/internal/debug/dump-config

Secure Example

# FastAPI — versioned API with deprecation headers and OpenAPI docs
from fastapi import FastAPI
from datetime import datetime

app = FastAPI(
    title="Example API",
    version="3.0.0",
    docs_url="/api/v3/docs",
    openapi_url="/api/v3/openapi.json",
)

# Redirect deprecated version with sunset header
@app.api_route("/api/v1/{path:path}", methods=["GET", "POST", "PUT", "DELETE"])
async def deprecated_v1(path: str):
    return JSONResponse(
        status_code=410,
        content={"error": "API v1 has been retired. Use /api/v3/"},
        headers={
            "Sunset": "Sat, 01 Jan 2024 00:00:00 GMT",
            "Deprecation": "true",
            "Link": '</api/v3/>; rel="successor-version"',
        },
    )

# Active version with full security
v3_router = APIRouter(prefix="/api/v3", dependencies=[Depends(authenticate)])

# CI/CD — validate OpenAPI spec exists and is current
# .github/workflows/api-docs.yml
- name: Validate OpenAPI spec
  run: |
    npx @stoplight/spectral-cli lint openapi.yaml
    # Fail if spec is missing routes that exist in code
    python scripts/check_api_coverage.py --spec openapi.yaml --app app:app

# Set deprecation headers on soon-to-retire endpoints
# Response headers for deprecated endpoint:
Deprecation: true
Sunset: Sat, 01 Jul 2025 00:00:00 GMT
Link: </api/v3/users>; rel="successor-version"

Prevention Strategies

• Maintain a centralized API inventory — track all environments (prod, staging, beta)
• Generate OpenAPI/Swagger specs from code and validate in CI/CD
• Enforce API versioning strategy: URI versioning (/v3/), header versioning, or content negotiation
• Set Sunset and Deprecation headers on deprecated endpoints
• Return 410 Gone for fully retired API versions — do not silently keep them running
• Restrict staging/beta APIs to internal networks or VPN
• Regularly audit for shadow APIs: scan networks, review API gateway configs, search code repos
• Document all third-party API integrations and data flows
• Require security review for new API endpoints before deployment

---

API10:2023 — Unsafe Consumption of APIs

APIs that consume data from third-party services without proper validation are vulnerable to attacks originating from those services. Developers often trust third-party API responses more than user input, but compromised or malicious upstream APIs can inject payloads.

API-Specific Risks

• Trusting third-party API response data without validation or sanitization
• Following redirects from external APIs to internal/malicious destinations
• Disabling TLS verification for third-party API calls
• No timeout or resource limits on third-party API consumption
• Storing third-party data without sanitization (XSS, SQL injection via upstream)
• Using third-party SDKs with known vulnerabilities

Vulnerable Example

# FastAPI — trusting third-party response without validation
import httpx

@app.get("/api/enriched-profile/{user_id}")
async def get_enriched_profile(user_id: uuid.UUID):
    user = db.query(User).filter(User.id == user_id).first()
    # VULNERABLE: no TLS verification, no timeout, no response validation
    response = httpx.get(
        f"https://third-party-api.com/enrich?email={user.email}",
        verify=False,  # VULNERABLE: TLS verification disabled
        follow_redirects=True,  # VULNERABLE: follows redirects blindly
    )
    enrichment = response.json()
    # VULNERABLE: storing unvalidated third-party data directly
    user.company = enrichment.get("company")
    user.title = enrichment.get("title")
    user.bio = enrichment.get("bio")  # could contain XSS payload
    db.commit()
    return user

// Express — consuming third-party API without validation
app.get("/api/weather/:city", async (req, res) => {
  // VULNERABLE: no input validation, no response validation, no timeout
  const response = await fetch(
    `http://weather-api.com/data?city=${req.params.city}`, // HTTP, not HTTPS
    { redirect: "follow" } // follows redirects blindly
  );
  const data = await response.json();
  // VULNERABLE: passing unvalidated third-party data to client
  res.json({ weather: data });
});

Secure Example

# FastAPI — safe third-party API consumption
import httpx
from pydantic import BaseModel, field_validator
import bleach

class EnrichmentResponse(BaseModel):
    company: str | None = None
    title: str | None = None
    bio: str | None = None

    @field_validator("bio", mode="before")
    @classmethod
    def sanitize_bio(cls, v: str | None) -> str | None:
        if v is None:
            return None
        return bleach.clean(v, tags=[], strip=True)  # strip all HTML

    @field_validator("company", "title", mode="before")
    @classmethod
    def truncate_fields(cls, v: str | None) -> str | None:
        if v is None:
            return None
        return v[:200]  # prevent oversized data storage

@app.get("/api/enriched-profile/{user_id}")
async def get_enriched_profile(
    user_id: uuid.UUID,
    current_user: User = Depends(get_current_user),
):
    user = db.query(User).filter(
        User.id == user_id, User.id == current_user.id
    ).first()
    if not user:
        raise HTTPException(status_code=404)

    async with httpx.AsyncClient(
        verify=True,             # enforce TLS verification
        follow_redirects=False,  # do not follow redirects
        timeout=5.0,             # strict timeout
    ) as client:
        try:
            response = await client.get(
                "https://third-party-api.com/enrich",
                params={"email": user.email},
            )
            response.raise_for_status()
        except httpx.HTTPError:
            raise HTTPException(status_code=502, detail="Upstream service error")

    # Validate response through schema
    enrichment = EnrichmentResponse.model_validate(response.json())
    user.company = enrichment.company
    user.title = enrichment.title
    user.bio = enrichment.bio
    db.commit()
    return UserResponse.model_validate(user)

// Express — safe third-party consumption
const { z } = require("zod");
const sanitizeHtml = require("sanitize-html");

const WeatherResponseSchema = z.object({
  temperature: z.number(),
  description: z.string().max(200),
  humidity: z.number().min(0).max(100),
});

app.get("/api/weather/:city", authenticate, async (req, res) => {
  const city = req.params.city.replace(/[^a-zA-Z\s-]/g, ""); // sanitize input

  const controller = new AbortController();
  const timeout = setTimeout(() => controller.abort(), 5000);

  try {
    const response = await fetch(
      `https://weather-api.com/data?city=${encodeURIComponent(city)}`,
      {
        signal: controller.signal,
        redirect: "error", // reject redirects
      }
    );
    clearTimeout(timeout);

    if (!response.ok) throw new Error("Upstream error");

    const raw = await response.json();
    const validated = WeatherResponseSchema.parse(raw); // validate response
    validated.description = sanitizeHtml(validated.description, {
      allowedTags: [], allowedAttributes: {},
    });
    res.json({ weather: validated });
  } catch (err) {
    clearTimeout(timeout);
    res.status(502).json({ error: "Weather service unavailable" });
  }
});

Prevention Strategies

• Validate and sanitize all third-party API response data through strict schemas
• Enforce TLS verification on all outbound requests — never set verify=False
• Do not follow redirects from third-party APIs — or re-validate the redirect target
• Set strict timeouts (connect and read) on all outbound HTTP requests
• Sanitize third-party data before storing (strip HTML, truncate, validate types)
• Use circuit breaker patterns for third-party API dependencies
• Pin and audit third-party SDK/library versions for known vulnerabilities
• Log and monitor third-party API interactions for anomalies
• Treat third-party data with the same suspicion as user input

---

Quick Reference Matrix

Secure API Design Patterns Reference

Reference for AI agents implementing secure API design across REST, GraphQL, and gRPC.

Table of Contents

• 1. Authentication Patterns
• 2. Authorization Patterns
• 3. Input Validation and Data Sanitization
• 4. Rate Limiting and Throttling
• 5. API Gateway and Edge Security
• 6. CORS and Cross-Origin Security
• 7. Error Handling and Information Disclosure
• 8. API Documentation and OpenAPI Security
• 9. Transport and Data Security
• 10. Logging, Monitoring, and Incident Response

---

1. Authentication Patterns

Verify the identity of every API caller using cryptographically sound mechanisms. Never roll custom authentication; use proven protocols and libraries.

OAuth 2.0 Flows

Use Authorization Code + PKCE for SPAs and mobile apps. Use Client Credentials for machine-to-machine (M2M) communication.

Anti-pattern — Implicit flow (deprecated, token in URL fragment):

GET /authorize?response_type=token&client_id=app123&redirect_uri=https://app.example.com/callback
# Token exposed in browser history, referrer headers, and logs

Correct — Authorization Code + PKCE:

# FastAPI OAuth2 with PKCE verification
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2AuthorizationCodeBearer

oauth2_scheme = OAuth2AuthorizationCodeBearer(
    authorizationUrl="/authorize",
    tokenUrl="/token",
    scopes={"read": "Read access", "write": "Write access"},
)

async def get_current_user(token: str = Depends(oauth2_scheme)):
    payload = verify_token(token)  # Validate signature, exp, aud, iss
    if not payload:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid or expired token",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return payload

// Express passport OAuth2 config
const passport = require("passport");
const { Strategy } = require("passport-oauth2");

passport.use(
  new Strategy(
    {
      authorizationURL: "https://auth.example.com/authorize",
      tokenURL: "https://auth.example.com/token",
      clientID: process.env.CLIENT_ID,
      clientSecret: process.env.CLIENT_SECRET,
      callbackURL: "/callback",
      pkce: true,
      state: true, // CSRF protection
    },
    (accessToken, refreshToken, profile, done) => {
      return done(null, profile);
    }
  )
);

JWT Best Practices

Anti-pattern — No algorithm validation:

# DANGEROUS: Accepts any algorithm, including "none"
payload = jwt.decode(token, SECRET_KEY)

Correct — Strict JWT validation:

from jwt import decode, InvalidTokenError

ALLOWED_ALGORITHMS = ["RS256", "ES256"]  # Asymmetric only

def verify_token(token: str) -> dict:
    try:
        return decode(
            token,
            PUBLIC_KEY,
            algorithms=ALLOWED_ALGORITHMS,  # Explicit allowlist
            audience="https://api.example.com",
            issuer="https://auth.example.com",
            options={
                "require": ["exp", "iat", "aud", "iss", "sub"],
                "verify_exp": True,
            },
        )
    except InvalidTokenError:
        return None

Key rules:

• Set access token expiry to 5–15 minutes; use refresh tokens for longer sessions
• Enforce algorithm allowlist — never accept "none" or HS256 with public keys
• Validate aud, iss, exp, iat, and sub on every request
• Implement refresh token rotation: invalidate the old token when issuing a new one
• Store refresh tokens hashed in the database; detect reuse to revoke token families

API Key Management

Anti-pattern — API key in URL:

GET /api/data?api_key=sk_live_abc123
# Key logged in access logs, browser history, referrer headers

Correct — Key in header, stored hashed:

import hashlib, secrets

def generate_api_key() -> tuple[str, str]:
    raw_key = secrets.token_urlsafe(32)
    hashed = hashlib.sha256(raw_key.encode()).hexdigest()
    return raw_key, hashed  # Return raw to user once; store hashed

def verify_api_key(provided_key: str, stored_hash: str) -> bool:
    return hashlib.sha256(provided_key.encode()).hexdigest() == stored_hash

# Client sends key in header
GET /api/data HTTP/1.1
X-API-Key: sk_live_abc123

Key rules:

• Never transmit API keys in URLs or query parameters
• Hash keys at rest (SHA-256 minimum); only show the full key once at creation
• Scope keys to specific endpoints, methods, and IP ranges
• Implement key rotation with grace periods (old key valid for 24–72 hours)
• Log key usage but never log the key value itself

mTLS for Service-to-Service

Use mutual TLS when services must prove identity to each other. Validate client certificates against an internal CA.

Session-Based vs Token-Based

• Use token-based (JWT/OAuth2) for stateless APIs, microservices, and mobile clients
• Use session-based for server-rendered apps with tight session control needs
• Never store JWTs in localStorage; use httpOnly, secure, sameSite cookies if browser-based

Multi-Factor Authentication

Require MFA step-up for sensitive operations (fund transfers, role changes, PII export). Issue short-lived, narrowly scoped tokens after MFA verification.

---

2. Authorization Patterns

Enforce access control on every request at object, function, and property levels. Never rely on client-side enforcement or security through obscurity.

RBAC vs ABAC vs ReBAC

Object-Level Authorization (BOLA Prevention)

Anti-pattern — No ownership check:

@app.get("/api/orders/{order_id}")
async def get_order(order_id: int):
    return db.query(Order).filter(Order.id == order_id).first()
    # Any authenticated user can access any order

Correct — Enforce ownership:

@app.get("/api/orders/{order_id}")
async def get_order(order_id: int, user: User = Depends(get_current_user)):
    order = db.query(Order).filter(
        Order.id == order_id,
        Order.user_id == user.id  # Scoped to requesting user
    ).first()
    if not order:
        raise HTTPException(status_code=404)  # 404, not 403 — prevent enumeration
    return order

Function-Level Authorization (BFLA Prevention)

Anti-pattern — UI-only restriction:

// Client hides admin button, but endpoint is unprotected
app.delete("/api/users/:id", async (req, res) => {
  await User.findByIdAndDelete(req.params.id);
  res.sendStatus(204);
});

Correct — Middleware-enforced authorization:

const requireRole = (...roles) => (req, res, next) => {
  if (!roles.includes(req.user.role)) {
    return res.status(403).json({ error: "Insufficient permissions" });
  }
  next();
};

app.delete("/api/users/:id", requireRole("admin"), async (req, res) => {
  await User.findByIdAndDelete(req.params.id);
  res.sendStatus(204);
});

Property-Level Authorization

Anti-pattern — Mass assignment:

@app.put("/api/users/{user_id}")
async def update_user(user_id: int, data: dict):
    db.query(User).filter(User.id == user_id).update(data)
    # Client can set is_admin=true, role="superuser", etc.

Correct — Explicit allowed fields with response filtering:

from pydantic import BaseModel

class UserUpdateRequest(BaseModel):
    display_name: str | None = None
    email: str | None = None
    # is_admin, role NOT included — cannot be set

class UserPublicResponse(BaseModel):
    id: int
    display_name: str
    email: str
    # ssn, password_hash NOT included — cannot be leaked

@app.put("/api/users/{user_id}", response_model=UserPublicResponse)
async def update_user(user_id: int, data: UserUpdateRequest, user=Depends(get_current_user)):
    if user_id != user.id:
        raise HTTPException(status_code=404)
    db.query(User).filter(User.id == user_id).update(data.model_dump(exclude_unset=True))

Policy Engines

Use external policy engines for complex authorization:

# OPA/Rego policy example
package api.authz

default allow := false

allow if {
    input.method == "GET"
    input.path == ["api", "orders", order_id]
    data.orders[order_id].owner == input.user.id
}

Key rules:

• Apply authorization checks server-side on every request — never trust client-side gating
• Return 404 (not 403) for resources the user should not know exist
• Use allowlist for writable/readable fields per role — never blocklist
• Centralize policy logic; avoid scattering authorization checks across handlers
• Audit and test authorization with automated integration tests per role

---

3. Input Validation and Data Sanitization

Validate all input at the API boundary using strict schemas. Reject anything that does not match the expected shape, type, and range.

Schema Validation

Anti-pattern — No validation, raw request body:

@app.post("/api/items")
async def create_item(request: Request):
    data = await request.json()  # Arbitrary keys, types, sizes
    db.insert(data)

Correct — Pydantic schema enforcement (FastAPI):

from pydantic import BaseModel, Field, field_validator
import re

class ItemCreate(BaseModel):
    name: str = Field(..., min_length=1, max_length=200)
    price: float = Field(..., gt=0, le=1_000_000)
    category: str = Field(..., pattern=r"^[a-z_]+$")

    @field_validator("name")
    @classmethod
    def sanitize_name(cls, v: str) -> str:
        if re.search(r"[<>\"';]", v):
            raise ValueError("Invalid characters in name")
        return v.strip()

@app.post("/api/items", status_code=201)
async def create_item(item: ItemCreate):
    return db.insert(item.model_dump())

Correct — Zod validation (Express):

const { z } = require("zod");

const ItemSchema = z.object({
  name: z.string().min(1).max(200),
  price: z.number().positive().max(1_000_000),
  category: z.string().regex(/^[a-z_]+$/),
});

app.post("/api/items", (req, res) => {
  const result = ItemSchema.safeParse(req.body);
  if (!result.success) {
    return res.status(422).json({ errors: result.error.issues });
  }
  db.insert(result.data);
  res.status(201).json(result.data);
});

Content-Type Enforcement

Anti-pattern — Accept any content type:

# No Content-Type check — vulnerable to XXE, SSRF via XML payloads

Correct — Reject unexpected content types:

from fastapi import Request, HTTPException

@app.middleware("http")
async def enforce_content_type(request: Request, call_next):
    if request.method in ("POST", "PUT", "PATCH"):
        content_type = request.headers.get("content-type", "")
        if not content_type.startswith("application/json"):
            raise HTTPException(status_code=415, detail="Unsupported Media Type")
    return await call_next(request)

Request Body Size Limits

# FastAPI / Starlette
app = FastAPI()
app.add_middleware(
    TrustedHostMiddleware, allowed_hosts=["api.example.com"]
)
# Set via reverse proxy (nginx: client_max_body_size 1m;)

// Express
app.use(express.json({ limit: "1mb" }));
app.use(express.urlencoded({ limit: "1mb", extended: false }));

GraphQL-Specific Validation

Anti-pattern — Unbounded query depth:

# Malicious deeply nested query
query {
  user { orders { items { reviews { author { orders { items { ... } } } } } } }
}

Correct — Depth and complexity limits:

const depthLimit = require("graphql-depth-limit");
const { createComplexityLimitRule } = require("graphql-validation-complexity");

const server = new ApolloServer({
  schema,
  validationRules: [
    depthLimit(5),
    createComplexityLimitRule(1000),
  ],
  introspection: process.env.NODE_ENV !== "production", // Disable in production
});

File Upload Security

import magic

ALLOWED_MIME_TYPES = {"image/png", "image/jpeg", "application/pdf"}
MAX_FILE_SIZE = 10 * 1024 * 1024  # 10 MB

async def validate_upload(file: UploadFile):
    if file.size > MAX_FILE_SIZE:
        raise HTTPException(status_code=413, detail="File too large")
    content = await file.read(2048)
    mime = magic.from_buffer(content, mime=True)
    if mime not in ALLOWED_MIME_TYPES:
        raise HTTPException(status_code=422, detail="Invalid file type")
    await file.seek(0)
    # Send to malware scanner before storing

Key rules:

• Validate on the server side even if the client validates
• Use allowlists for types, ranges, and patterns — never blocklists
• Enforce Content-Type headers; reject unexpected media types
• Set request body size limits at both application and reverse proxy layers
• For GraphQL: limit query depth (≤5), complexity (≤1000), and disable introspection in production
• Validate file uploads by magic bytes, not file extension
• Prevent parameter pollution: use first-value-wins or reject duplicates

---

4. Rate Limiting and Throttling

Protect APIs from abuse, brute force, and resource exhaustion by limiting request rates per identity and per resource.

Rate Limit Strategies

Implementation

Anti-pattern — No rate limiting:

@app.post("/api/login")
async def login(creds: LoginRequest):
    # No limit — vulnerable to credential stuffing
    return authenticate(creds)

Correct — Redis-based sliding window (FastAPI with slowapi):

from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter

@app.post("/api/login")
@limiter.limit("5/minute")  # 5 attempts per minute per IP
async def login(request: Request, creds: LoginRequest):
    return authenticate(creds)

@app.get("/api/data")
@limiter.limit("100/minute")  # General endpoint
async def get_data(request: Request, user=Depends(get_current_user)):
    return fetch_data(user)

Correct — Express rate limiter:

const rateLimit = require("express-rate-limit");
const RedisStore = require("rate-limit-redis");
const Redis = require("ioredis");

const loginLimiter = rateLimit({
  store: new RedisStore({ sendCommand: (...args) => redisClient.call(...args) }),
  windowMs: 60 * 1000,
  max: 5,
  standardHeaders: true, // X-RateLimit-Limit, X-RateLimit-Remaining
  legacyHeaders: false,
  message: { error: "Too many login attempts, try again later" },
  keyGenerator: (req) => req.ip,
});

app.post("/api/login", loginLimiter, loginHandler);

Rate Limit Headers

Always include standard rate limit headers in responses:

HTTP/1.1 200 OK
X-RateLimit-Limit: 100
X-RateLimit-Remaining: 42
X-RateLimit-Reset: 1699900000
Retry-After: 30  # Include on 429 responses

GraphQL Cost-Based Rate Limiting

Assign cost to operations instead of counting raw requests:

const costMap = {
  Query: { users: { complexity: 10 }, user: { complexity: 1 } },
  User: { orders: { complexity: 5 } },
};
// Budget: 1000 points per minute per user

Key rules:

• Apply stricter limits to authentication endpoints (5–10/minute)
• Use per-user limits for authenticated endpoints; per-IP for unauthenticated
• Return 429 Too Many Requests with Retry-After header
• Use distributed stores (Redis) for rate limiting across multiple instances
• Implement per-endpoint limits — not just global limits
• For GraphQL, use cost-based limiting rather than request counting
• Add anti-automation measures (CAPTCHA, device fingerprinting) for business-critical flows

---

5. API Gateway and Edge Security

Offload cross-cutting security concerns to the API gateway. Apply defense-in-depth — never rely solely on the gateway.

Gateway Responsibilities

Client → WAF → API Gateway → Backend Service
              │
              ├── TLS termination
              ├── Authentication (JWT validation)
              ├── Rate limiting
              ├── Request size enforcement
              ├── Request/response logging
              └── Request transformation

API Versioning

Prefer URL path versioning for clarity:

GET /api/v1/users HTTP/1.1
GET /api/v2/users HTTP/1.1

Set deprecation timelines and return Deprecation and Sunset headers:

HTTP/1.1 200 OK
Deprecation: true
Sunset: Sat, 01 Mar 2025 00:00:00 GMT
Link: <https://api.example.com/v2/users>; rel="successor-version"

Circuit Breaker Pattern

Prevent cascading failures when downstream services degrade:

import circuitbreaker

@circuitbreaker.circuit(failure_threshold=5, recovery_timeout=30)
def call_downstream_service(request_data):
    response = httpx.post("https://internal-service/api", json=request_data, timeout=5.0)
    response.raise_for_status()
    return response.json()

Key rules:

• Terminate TLS at the gateway; use mTLS between gateway and backends
• Validate JWTs at the gateway, but re-validate authorization at the service
• Log all requests at the gateway for audit and anomaly detection
• Use WAF rules to block known attack patterns (SQLi, XSS, path traversal)
• Version all public APIs; communicate deprecation with standard headers
• Implement circuit breakers with sensible timeouts for all downstream calls
• Never expose internal service topology in error messages or headers

---

6. CORS and Cross-Origin Security

Configure CORS to allow only known origins. Misconfigured CORS can bypass same-origin protections entirely.

CORS Configuration

Anti-pattern — Wildcard with credentials:

# DANGEROUS: Allows any origin to send credentialed requests
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
)

Correct — Explicit origin allowlist:

from fastapi.middleware.cors import CORSMiddleware

ALLOWED_ORIGINS = [
    "https://app.example.com",
    "https://admin.example.com",
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=ALLOWED_ORIGINS,
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["Authorization", "Content-Type"],
    max_age=3600,  # Cache preflight for 1 hour
)

Correct — Express CORS:

const cors = require("cors");

const corsOptions = {
  origin: ["https://app.example.com", "https://admin.example.com"],
  credentials: true,
  methods: ["GET", "POST", "PUT", "DELETE"],
  allowedHeaders: ["Authorization", "Content-Type"],
  maxAge: 3600,
};

app.use(cors(corsOptions));

CSRF Protection for Cookie-Based APIs

If using cookies for authentication, enforce CSRF tokens:

const csrf = require("csurf");

app.use(csrf({ cookie: { httpOnly: true, secure: true, sameSite: "strict" } }));

app.get("/api/csrf-token", (req, res) => {
  res.json({ csrfToken: req.csrfToken() });
});

Key rules:

• Never use allow_origins=["*"] with allow_credentials=True
• Maintain an explicit allowlist of origins — do not reflect the Origin header
• Restrict allow_methods and allow_headers to what is actually needed
• Set Access-Control-Max-Age (3600s recommended) to reduce preflight requests
• For cookie-based auth: set SameSite=Strict or Lax, Secure, HttpOnly
• Implement CSRF token validation for any state-changing cookie-authenticated request

---

7. Error Handling and Information Disclosure

Return consistent, generic error responses. Never leak internal implementation details, stack traces, or database information to clients.

RFC 7807 Problem Details

Anti-pattern — Leaking internals:

@app.exception_handler(Exception)
async def error_handler(request, exc):
    return JSONResponse(
        status_code=500,
        content={
            "error": str(exc),  # "psycopg2.OperationalError: connection to 10.0.1.5:5432 refused"
            "stack": traceback.format_exc(),
            "query": "SELECT * FROM users WHERE id = 42",
        },
    )

Correct — RFC 7807 structured error, no internals:

import uuid
import logging

logger = logging.getLogger(__name__)

class ProblemDetail(BaseModel):
    type: str = "about:blank"
    title: str
    status: int
    detail: str | None = None
    instance: str | None = None

@app.exception_handler(Exception)
async def error_handler(request: Request, exc: Exception):
    error_id = str(uuid.uuid4())
    logger.error("Unhandled error %s: %s", error_id, exc, exc_info=True)  # Full details in logs only
    return JSONResponse(
        status_code=500,
        content=ProblemDetail(
            type="https://api.example.com/errors/internal",
            title="Internal Server Error",
            detail="An unexpected error occurred. Reference: " + error_id,
            instance=str(request.url.path),
            status=500,
        ).model_dump(),
        media_type="application/problem+json",
    )

Correct — Express centralized error handler:

const { v4: uuidv4 } = require("uuid");

app.use((err, req, res, _next) => {
  const errorId = uuidv4();
  console.error(`Error ${errorId}:`, err); // Full details to logs only

  const status = err.statusCode || 500;
  res.status(status).type("application/problem+json").json({
    type: `https://api.example.com/errors/${status === 500 ? "internal" : "client"}`,
    title: status === 500 ? "Internal Server Error" : err.message,
    status,
    detail: `Reference: ${errorId}`,
    instance: req.originalUrl,
  });
});

Status Code Usage

Key rules:

• Return application/problem+json content type for all error responses
• Generate a unique error ID per incident; return the ID to the client, log the details server-side
• Never include stack traces, SQL queries, file paths, or server versions in responses
• Use 404 instead of 403 to prevent resource enumeration when appropriate
• Implement fail-closed: on authorization errors or service failures, deny access by default
• Use consistent error response schema across all endpoints

---

8. API Documentation and OpenAPI Security

Treat API documentation as a security-sensitive asset. Define security schemes explicitly and control documentation access.

Security Scheme Definitions

# OpenAPI 3.1 security schemes
components:
  securitySchemes:
    bearerAuth:
      type: http
      scheme: bearer
      bearerFormat: JWT
    apiKeyAuth:
      type: apiKey
      in: header
      name: X-API-Key
    oauth2:
      type: oauth2
      flows:
        authorizationCode:
          authorizationUrl: https://auth.example.com/authorize
          tokenUrl: https://auth.example.com/token
          scopes:
            read:users: Read user data
            write:users: Modify user data

security:
  - bearerAuth: []

paths:
  /api/users:
    get:
      security:
        - oauth2: [read:users]
      responses:
        "401":
          description: Authentication required
        "403":
          description: Insufficient scope

Marking Sensitive Fields

components:
  schemas:
    User:
      properties:
        id:
          type: integer
        email:
          type: string
          format: email
          x-sensitive: true  # Custom extension for PII marking
        ssn:
          type: string
          writeOnly: true  # Never returned in responses
          x-sensitive: true

Key rules:

• Define security schemes in OpenAPI spec for every endpoint
• Disable Swagger UI and OpenAPI spec access in production for internal APIs
• Mark PII and sensitive fields with custom extensions (x-sensitive)
• Use writeOnly for fields that should never appear in responses
• Maintain an API inventory; track all published endpoints and their security posture
• Set deprecation timelines and remove deprecated endpoints on schedule
• Auto-generate documentation from code to prevent spec drift

---

9. Transport and Data Security

Encrypt data in transit and at rest. Validate the integrity of requests and responses across the wire.

TLS Enforcement

Anti-pattern — Accepting HTTP:

# No redirect — allows plaintext traffic
server {
    listen 80;
    listen 443 ssl;
}

Correct — Enforce HTTPS with HSTS:

server {
    listen 80;
    return 301 https://$host$request_uri;
}

server {
    listen 443 ssl;
    ssl_protocols TLSv1.2 TLSv1.3;
    ssl_ciphers HIGH:!aNULL:!MD5:!RC4;
    add_header Strict-Transport-Security "max-age=63072000; includeSubDomains; preload" always;
}

Request Signing (Webhook Delivery)

Anti-pattern — No signature verification:

@app.post("/webhooks/payment")
async def handle_webhook(request: Request):
    data = await request.json()  # No verification — anyone can send fake events
    process_payment(data)

Correct — HMAC signature verification:

import hmac
import hashlib

WEBHOOK_SECRET = os.environ["WEBHOOK_SECRET"]

@app.post("/webhooks/payment")
async def handle_webhook(request: Request):
    body = await request.body()
    signature = request.headers.get("X-Signature-256")
    if not signature:
        raise HTTPException(status_code=401, detail="Missing signature")

    expected = "sha256=" + hmac.new(
        WEBHOOK_SECRET.encode(), body, hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(signature, expected):
        raise HTTPException(status_code=401, detail="Invalid signature")

    data = json.loads(body)
    process_payment(data)

PII Handling

from cryptography.fernet import Fernet

FIELD_KEY = Fernet(os.environ["FIELD_ENCRYPTION_KEY"])

class UserRecord:
    def encrypt_pii(self, ssn: str) -> bytes:
        return FIELD_KEY.encrypt(ssn.encode())

    def decrypt_pii(self, encrypted: bytes) -> str:
        return FIELD_KEY.decrypt(encrypted).decode()

Key rules:

• Enforce TLS 1.2+ on all endpoints; disable TLS 1.0/1.1
• Set HSTS with max-age ≥ 63072000 (2 years), includeSubDomains, and preload
• Implement certificate pinning for mobile API clients
• Sign webhooks with HMAC-SHA256; use hmac.compare_digest for timing-safe comparison
• Apply field-level encryption for PII (SSN, payment data) at rest
• Follow data minimization: collect and return only the fields necessary for the operation
• Use AWS Signature V4 or similar for signed requests in cloud-to-cloud scenarios

---

10. Logging, Monitoring, and Incident Response

Log every API interaction with sufficient detail for security investigation. Never log secrets or PII.

Structured Request Logging

Anti-pattern — Logging sensitive data:

logger.info(f"Login: user={username} password={password} token={token}")
# Credentials and tokens in logs — catastrophic if logs are breached

Correct — Structured logging with redaction:

import structlog
import time

logger = structlog.get_logger()

@app.middleware("http")
async def log_requests(request: Request, call_next):
    correlation_id = request.headers.get("X-Correlation-ID", str(uuid.uuid4()))
    start = time.perf_counter()

    response = await call_next(request)
    duration_ms = (time.perf_counter() - start) * 1000

    logger.info(
        "api_request",
        method=request.method,
        path=request.url.path,
        status=response.status_code,
        duration_ms=round(duration_ms, 2),
        user_id=getattr(request.state, "user_id", None),
        correlation_id=correlation_id,
        ip=request.client.host,
        user_agent=request.headers.get("user-agent"),
        # NEVER log: Authorization header, request body, cookies
    )
    response.headers["X-Correlation-ID"] = correlation_id
    return response

Correct — Express structured logging:

const pino = require("pino");
const logger = pino({ redact: ["req.headers.authorization", "req.headers.cookie"] });

app.use((req, res, next) => {
  const correlationId = req.headers["x-correlation-id"] || uuidv4();
  req.correlationId = correlationId;
  res.setHeader("X-Correlation-ID", correlationId);

  const start = process.hrtime.bigint();
  res.on("finish", () => {
    const durationMs = Number(process.hrtime.bigint() - start) / 1e6;
    logger.info({
      method: req.method,
      path: req.originalUrl,
      status: res.statusCode,
      durationMs: Math.round(durationMs * 100) / 100,
      userId: req.user?.id,
      correlationId,
      ip: req.ip,
    });
  });
  next();
});

Audit Trail

Log security-relevant events with immutable, append-only storage:

def audit_log(event: str, user_id: str, resource: str, details: dict):
    logger.info(
        "audit_event",
        event=event,           # "data_access", "role_change", "export_pii"
        user_id=user_id,
        resource=resource,
        details=details,
        timestamp=datetime.utcnow().isoformat(),
    )

Anomaly Detection Signals

Monitor for and alert on:

Key rules:

• Use structured logging (JSON) — never unstructured text
• Never log: passwords, tokens, API keys, session IDs, PII, or full request/response bodies with sensitive content
• Attach correlation IDs (propagated via X-Correlation-ID header) across all services
• Implement separate audit logs for data access, modifications, and authentication events
• Set up real-time alerting for auth failure spikes, error rate increases, and unusual access patterns
• Store logs in tamper-evident, append-only storage with retention policies
• Use distributed tracing (OpenTelemetry) for cross-service request tracking
• Review and rotate logging configurations regularly to prevent log injection