Chapter 52: API Security Framework¶

Overview¶

APIs are the connective tissue of modern software. Every mobile application, every single-page frontend, every microservice mesh, every IoT device, and every third-party integration communicates through APIs. When organizations migrated from monolithic server-rendered applications to distributed architectures, they moved the attack surface from a handful of web forms to hundreds or thousands of programmatic endpoints — each accepting structured input, returning structured output, and often trusting that callers are who they claim to be. The result is an attack surface that is broader, deeper, and harder to secure than traditional web applications.

This chapter provides a comprehensive, offense-and-defense treatment of API security across four protocol families: REST (HTTP/JSON), GraphQL, gRPC (Protocol Buffers over HTTP/2), and WebSocket. We begin with the attack surface taxonomy, move through the OWASP API Security Top 10 (2023 edition), then dive deep into authentication and authorization attacks, GraphQL-specific exploitation, rate limiting evasion, mass assignment, and SSRF. Every offensive technique is paired with detection queries (KQL and SPL), architectural mitigations, and mapping to the MITRE ATT&CK framework. The chapter closes with API security architecture patterns — API gateways, mutual TLS, schema validation — and cloud-native considerations including service mesh security and zero-trust API design.

This is the chapter you reference when you need to attack-test, defend, or architect API security for any modern application.

Learning Objectives¶

By the end of this chapter, students SHALL be able to:

1. Classify API attack surfaces across REST, GraphQL, gRPC, and WebSocket protocols, identifying unique risk profiles for each (Analysis)
2. Evaluate all ten OWASP API Security Top 10 (2023) categories and map organizational API vulnerabilities to the relevant category (Evaluation)
3. Demonstrate Broken Object Level Authorization (BOLA/IDOR) exploitation techniques and construct corresponding detection queries (Application)
4. Analyze JWT-based authentication attacks including algorithm confusion, none algorithm bypass, key confusion, and claim tampering (Analysis)
5. Design GraphQL security controls that mitigate introspection abuse, depth attacks, batching attacks, and injection through variables (Synthesis)
6. Construct rate limiting architectures that resist bypass techniques including header spoofing, distributed requests, and pagination abuse (Synthesis)
7. Implement detection engineering pipelines for API attacks using KQL and SPL query pairs mapped to MITRE ATT&CK techniques (Application)
8. Assess API gateway architectures (Kong, Apigee, AWS API Gateway) for security posture gaps and misconfigurations (Evaluation)
9. Create a continuous API security testing methodology integrated into CI/CD pipelines with automated schema validation (Synthesis)
10. Develop zero-trust API architecture patterns using mutual TLS, service mesh, and policy-as-code enforcement for cloud-native environments (Synthesis)

Prerequisites¶

• Completion of Chapter 20: Cloud Attack & Defense — cloud security fundamentals, shared responsibility model, cloud-native service exploitation
• Completion of Chapter 44: Web Application Penetration Testing — HTTP fundamentals, injection attacks, session management, OWASP Top 10 web vulnerabilities
• Familiarity with Chapter 35: DevSecOps Pipeline — CI/CD security, shift-left principles, automated security testing integration
• Familiarity with Chapter 5: Detection Engineering at Scale — detection query methodology, KQL/SPL fundamentals, alert tuning
• Working knowledge of HTTP/HTTPS, JSON, REST conventions, and basic authentication mechanisms (OAuth2, JWT, API keys)

MITRE ATT&CK API Security Mapping¶

52.1 API Attack Surface Taxonomy¶

APIs are not a single technology — they are a family of protocols, each with distinct security characteristics. Understanding the attack surface requires understanding the protocol.

52.1.1 Protocol Comparison¶

52.1.2 Authentication vs Authorization Boundaries¶

The most common confusion in API security — and the root cause of the OWASP API #1 vulnerability — is conflating authentication with authorization.

Authentication: "Who are you?"  → Token validation, credential verification
Authorization:  "What can you do?" → Object-level, function-level, field-level access control

Example failure:
  User A authenticates successfully (valid JWT)
  User A requests GET /api/v2/orders/10847  (Order belongs to User B)
  API validates JWT ✓  — authentication passes
  API does NOT check order ownership — authorization fails OPEN
  User A receives User B's order data

Key principle: Authentication is necessary but never sufficient. Every API endpoint must enforce authorization at the object level, function level, and field level independently.

52.1.3 API Gateway Architectures¶

graph LR
    subgraph External
        C1[Mobile App] --> GW
        C2[Web SPA] --> GW
        C3[Partner API] --> GW
    end

    subgraph API Gateway - 198.51.100.10
        GW[API Gateway<br/>Rate Limiting<br/>Auth Validation<br/>Schema Validation]
    end

    subgraph Internal Services - 10.0.0.0/16
        GW --> S1[User Service<br/>10.0.1.10]
        GW --> S2[Order Service<br/>10.0.1.20]
        GW --> S3[Payment Service<br/>10.0.1.30]
        GW --> S4[Notification Service<br/>10.0.1.40]
    end

    style GW fill:#1e3a5f,color:#e6edf3
    style S1 fill:#2a4a6f,color:#e6edf3
    style S2 fill:#2a4a6f,color:#e6edf3
    style S3 fill:#36597f,color:#e6edf3
    style S4 fill:#36597f,color:#e6edf3

Security-critical gateway functions:

• Authentication termination — Validate tokens at the edge; internal services receive pre-validated identity
• Rate limiting — Enforce per-client, per-endpoint, and global rate limits
• Schema validation — Reject malformed requests before they reach application logic
• TLS termination — Enforce TLS 1.2+ externally; optionally mutual TLS internally
• Request/response transformation — Strip sensitive headers, mask fields
• Logging and telemetry — Centralized API access logs for detection engineering

Gateway misconfiguration risks:

52.2 OWASP API Security Top 10 (2023)¶

The OWASP API Security Top 10 (2023 revision) provides the canonical vulnerability taxonomy for API-specific risks. Each category below includes the vulnerability mechanism, an attack example with synthetic data, and a detection approach.

52.2.1 API1:2023 — Broken Object Level Authorization (BOLA)¶

The most prevalent and dangerous API vulnerability. The API exposes endpoints that handle object identifiers, but fails to validate that the authenticated user has permission to access the requested object.

Mechanism: The API uses client-supplied identifiers (IDs in URL path, query parameters, or request body) to retrieve objects. Authorization checks are missing or applied inconsistently.

# Legitimate request — User 1042 retrieves their own profile
GET /api/v2/users/1042/profile HTTP/1.1
Host: api.megacorp-example.com
Authorization: Bearer eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9...
# Response: 200 OK — returns user 1042's profile

# BOLA attack — User 1042 requests another user's profile
GET /api/v2/users/1043/profile HTTP/1.1
Host: api.megacorp-example.com
Authorization: Bearer eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9...
# Response: 200 OK — returns user 1043's profile (VULNERABILITY)

Detection (KQL):

// Detect potential BOLA — user accessing multiple different user IDs
ApiAccessLogs
| where Endpoint matches regex @"/api/v\d+/users/\d+/"
| extend TargetUserId = extract(@"/users/(\d+)/", 1, Endpoint)
| where TargetUserId != AuthenticatedUserId
| summarize DistinctTargets = dcount(TargetUserId),
            Attempts = count() by AuthenticatedUserId, bin(TimeGenerated, 5m)
| where DistinctTargets > 5
| project TimeGenerated, AuthenticatedUserId, DistinctTargets, Attempts

Detection (SPL):

index=api_logs endpoint="/api/v*/users/*/profile"
| rex field=endpoint "/users/(?<target_user_id>\d+)/"
| where target_user_id != authenticated_user_id
| stats dc(target_user_id) as distinct_targets, count as attempts by authenticated_user_id, _time span=5m
| where distinct_targets > 5

Mitigation: Implement authorization checks at the data access layer. Use the authenticated user's identity (from the validated token) to scope database queries — never trust client-supplied user IDs for authorization.

52.2.2 API2:2023 — Broken Authentication¶

Authentication mechanisms are implemented incorrectly, allowing attackers to compromise authentication tokens or exploit implementation flaws.

Common manifestations:

• Weak password requirements on API registration endpoints
• No rate limiting on /auth/login or /auth/token endpoints
• JWT tokens without expiration (exp claim missing)
• Credentials transmitted in URL query parameters (logged in server logs, proxies)
• Token refresh without rotation (old refresh tokens remain valid)
• Password reset tokens that are predictable or never expire

# Credential stuffing — no rate limiting on login endpoint
POST /api/v1/auth/login HTTP/1.1
Host: api.megacorp-example.com
Content-Type: application/json

{"email": "testuser@example.com", "password": "REDACTED"}

# Attacker sends 10,000 requests/minute — no 429 response, no lockout

Mitigation: Enforce rate limiting on all authentication endpoints. Use strong token generation (256-bit entropy minimum). Implement token expiration and rotation. Never accept credentials in query parameters.

52.2.3 API3:2023 — Broken Object Property Level Authorization¶

The API exposes object properties that the user should not be able to read (excessive data exposure) or write (mass assignment).

# User retrieves their own profile — API returns excessive data
GET /api/v2/users/me HTTP/1.1
Authorization: Bearer <token>

# Response includes internal fields the user should not see:
{
  "id": 1042,
  "email": "testuser@example.com",
  "name": "Test User",
  "role": "user",
  "internal_credit_score": 742,
  "account_flags": ["verified", "premium_trial"],
  "created_by_admin": "admin-198",
  "cost_center": "CC-4400"
}

Mitigation: Use explicit response schemas. Never return raw database objects. Whitelist response fields per endpoint and per role. Use DTOs (Data Transfer Objects) or serializers that enforce field-level visibility.

52.2.4 API4:2023 — Unrestricted Resource Consumption¶

The API does not restrict the size or number of resources that can be requested, enabling denial of service and cost amplification.

Examples: Missing pagination limits, no maximum query complexity, no upload size limits, no rate limits on expensive operations.

# Requesting all records with no pagination limit
GET /api/v2/transactions?limit=999999999&offset=0 HTTP/1.1
Authorization: Bearer <token>

# GraphQL complexity attack
POST /graphql HTTP/1.1
Content-Type: application/json

{
  "query": "{ users { orders { items { reviews { author { orders { items { reviews { text }}}}}}}}}}"
}

Mitigation: Enforce maximum page sizes, query complexity limits, request rate limits, upload size limits, and execution timeouts on all endpoints.

52.2.5 API5:2023 — Broken Function Level Authorization¶

The API does not properly enforce function-level authorization, allowing regular users to access administrative functions.

# Regular user discovers admin endpoint
DELETE /api/v2/admin/users/1043 HTTP/1.1
Authorization: Bearer <regular_user_token>

# Response: 200 OK — user deleted (VULNERABILITY)
# The API validated the token but did not check the user's role

Detection (KQL):

// Detect non-admin users accessing admin API endpoints
ApiAccessLogs
| where Endpoint contains "/admin/"
| where UserRole != "admin" and UserRole != "superadmin"
| where StatusCode in (200, 201, 204)
| summarize AttemptCount = count() by UserId, UserRole, Endpoint, bin(TimeGenerated, 1h)
| project TimeGenerated, UserId, UserRole, Endpoint, AttemptCount

Detection (SPL):

index=api_logs endpoint="*/admin/*" user_role!=admin user_role!=superadmin status_code IN (200, 201, 204)
| stats count as attempt_count by user_id, user_role, endpoint, _time span=1h
| table _time user_id user_role endpoint attempt_count

Mitigation: Implement role-based access control (RBAC) with explicit deny-by-default. Administrative endpoints should be on separate routes with middleware that validates role claims before reaching business logic.

52.2.6 API6:2023 — Unrestricted Access to Sensitive Business Flows¶

Attackers exploit legitimate business flows by automating them at scale — purchasing limited-stock items, creating fake accounts, scraping proprietary data.

Mitigation: Implement business-logic rate limiting (not just technical rate limiting). Use CAPTCHAs for sensitive flows, device fingerprinting, and anomaly detection on usage patterns.

52.2.7 API7:2023 — Server Side Request Forgery (SSRF)¶

The API fetches remote resources based on user-supplied URLs without validation, enabling attackers to access internal services.

POST /api/v2/documents/import HTTP/1.1
Authorization: Bearer <token>
Content-Type: application/json

{
  "source_url": "http://169.254.169.254/latest/meta-data/iam/security-credentials/api-role"
}
# API fetches the URL server-side → returns cloud metadata credentials

Mitigation: Validate and whitelist allowed URL schemes and domains. Block requests to private IP ranges (10.0.0.0/8, 172.16.0.0/12, 192.168.0.0/16, 169.254.169.254). Use allowlists, not denylists.

52.2.8 API8:2023 — Security Misconfiguration¶

Missing security hardening, overly permissive CORS, unnecessary HTTP methods enabled, verbose error messages, missing security headers.

Common misconfigurations:

# Overly permissive CORS
Access-Control-Allow-Origin: *
Access-Control-Allow-Credentials: true

# Debug mode in production
GET /api/v2/debug/routes HTTP/1.1
# Returns full route table with internal documentation

# Stack trace in error response
{
  "error": "NullPointerException at com.example.api.UserService.getUser(UserService.java:142)",
  "stack": "...",
  "database": "postgresql://db.internal.example.com:5432/users_prod"
}

Mitigation: Harden all API configurations. Disable debug endpoints in production. Configure restrictive CORS policies. Return generic error messages. Remove unnecessary HTTP methods. Apply security headers.

52.2.9 API9:2023 — Improper Inventory Management¶

Organizations lose track of old API versions, shadow APIs, deprecated endpoints that still function, and undocumented internal APIs exposed externally.

# Old API version still active — lacks newer security controls
GET /api/v1/users/1043/profile HTTP/1.1
# v1 has no authorization checks (fixed in v2)
# v1 was never decommissioned

Mitigation: Maintain a complete API inventory. Decommission old versions with hard deadlines. Use API gateway routing to block deprecated versions. Scan for shadow APIs with traffic analysis.

52.2.10 API10:2023 — Unsafe Consumption of APIs¶

The API blindly trusts data from third-party APIs or integrations without validation, treating external input as trusted.

Application → calls Partner API → receives JSON response → inserts into database without sanitization
Partner API is compromised → returns malicious payload → SQL injection via trusted channel

Mitigation: Validate and sanitize all input from third-party APIs. Apply the same input validation to external API responses as you would to user input. Use schema validation on all API consumption.

52.3 Authentication & Authorization Attacks¶

52.3.1 Broken Object Level Authorization (BOLA/IDOR) — Deep Dive¶

BOLA (also known as Insecure Direct Object Reference — IDOR) is the #1 API vulnerability for a reason: it is trivially exploitable, frequently missed by automated scanners, and can expose entire databases.

Attack patterns:

Exploitation methodology:

# Step 1: Authenticate and identify your own user ID
curl -s https://api.megacorp-example.com/api/v2/users/me \
  -H "Authorization: Bearer $TOKEN" | jq '.id'
# Returns: 1042

# Step 2: Attempt to access adjacent user IDs
for id in $(seq 1040 1050); do
  status=$(curl -s -o /dev/null -w "%{http_code}" \
    "https://api.megacorp-example.com/api/v2/users/$id/profile" \
    -H "Authorization: Bearer $TOKEN")
  echo "User $id: HTTP $status"
done

# Step 3: Check horizontal access across resource types
curl -s "https://api.megacorp-example.com/api/v2/users/1043/documents" \
  -H "Authorization: Bearer $TOKEN"
# If 200 OK with data — BOLA confirmed

Defensive architecture:

# VULNERABLE — direct ID from URL, no ownership check
@app.route('/api/v2/orders/<int:order_id>')
def get_order(order_id):
    order = db.query(Order).get(order_id)
    return jsonify(order.to_dict())

# SECURE — scoped to authenticated user's ownership
@app.route('/api/v2/orders/<int:order_id>')
@require_auth
def get_order(order_id):
    order = db.query(Order).filter(
        Order.id == order_id,
        Order.user_id == current_user.id  # Authorization at query level
    ).first()
    if not order:
        return jsonify({"error": "Not found"}), 404  # Don't reveal existence
    return jsonify(order.to_dict())

52.3.2 JWT Authentication Attacks¶

JSON Web Tokens (JWTs) are the dominant API authentication mechanism — and a rich attack surface when implemented incorrectly.

Algorithm Confusion Attack¶

JWT header: {"alg": "RS256", "typ": "JWT"}
Server expects RS256 (asymmetric — public key verifies, private key signs)

Attack: Attacker changes header to {"alg": "HS256", "typ": "JWT"}
Attacker signs the token with the PUBLIC key (which is publicly available)
Vulnerable server uses HS256 verification with the public key
Token validates — attacker has forged a valid token

Exploitation example:

# Algorithm confusion attack (synthetic demonstration)
import jwt
import base64

# Step 1: Obtain the server's public key (often in JWKS or .well-known)
# curl https://auth.megacorp-example.com/.well-known/jwks.json
public_key = open('public_key.pem', 'r').read()

# Step 2: Create a forged token using HS256 with the public key as secret
forged_payload = {
    "sub": "admin@example.com",
    "role": "admin",
    "iat": 1700000000,
    "exp": 1700003600
}

# Step 3: Sign with HS256 using the RSA public key as the HMAC secret
forged_token = jwt.encode(forged_payload, public_key, algorithm='HS256')
# If the server accepts HS256 and verifies with the public key → forged admin token

Detection (KQL):

// Detect JWT algorithm switching attempts
ApiAccessLogs
| where AuthHeaderPresent == true
| extend JwtHeader = base64_decode_tostring(split(AuthToken, ".")[0])
| extend Algorithm = parse_json(JwtHeader).alg
| where Algorithm in ("HS256", "none", "HS384", "HS512")
    and ExpectedAlgorithm == "RS256"
| project TimeGenerated, SourceIP, Endpoint, Algorithm, UserAgent

Detection (SPL):

index=api_logs auth_header=*
| eval jwt_header=base64decode(mvindex(split(auth_token, "."), 0))
| spath input=jwt_header output=algorithm path=alg
| where algorithm IN ("HS256", "none", "HS384", "HS512") AND expected_algorithm="RS256"
| table _time src_ip endpoint algorithm user_agent

None Algorithm Attack¶

# Original valid token
Header: {"alg": "RS256", "typ": "JWT"}
Payload: {"sub": "user@example.com", "role": "user"}
Signature: <valid RSA signature>

# Forged token — algorithm set to "none"
Header: {"alg": "none", "typ": "JWT"}
Payload: {"sub": "user@example.com", "role": "admin"}
Signature: <empty>

# Encoded: eyJhbGciOiJub25lIiwidHlwIjoiSldUIn0.eyJzdWIiOiJ1c2VyQGV4YW1wbGUuY29tIiwicm9sZSI6ImFkbWluIn0.
# Trailing dot with no signature — vulnerable servers accept this

Mitigation: Explicitly whitelist allowed algorithms on the server side. Never allow the token to dictate which algorithm to use. Reject tokens with alg: "none". Use a well-maintained JWT library.

Key Confusion (HMAC/RSA)¶

When a server is configured for RSA but also accepts HMAC, an attacker can use the RSA public key as the HMAC secret. Since the public key is public, anyone can forge tokens.

Mitigation: Configure the JWT library to only accept the expected algorithm. Use separate key stores for different algorithm types. Modern libraries (e.g., jose, python-jose) support algorithm whitelisting.

Claim Tampering¶

// Original claims
{
  "sub": "1042",
  "email": "testuser@example.com",
  "role": "user",
  "tenant_id": "tenant-a",
  "exp": 1700003600
}

// Tampered claims (requires signature bypass via algorithm confusion or weak key)
{
  "sub": "1",
  "email": "admin@example.com",
  "role": "admin",
  "tenant_id": "tenant-a",
  "exp": 1900000000
}

Critical claims to protect: sub (subject/user ID), role/roles, scope, tenant_id (multi-tenant), exp (expiration), iss (issuer), aud (audience).

52.3.3 OAuth2 Flows and Pitfalls¶

OAuth2 is the standard authorization framework for API access delegation. Each flow has distinct security characteristics.

Common OAuth2 vulnerabilities:

# 1. Open redirect via redirect_uri manipulation
GET /oauth/authorize?client_id=legit-app&redirect_uri=https://evil.example.com/callback&response_type=code

# 2. Missing state parameter — CSRF attack
GET /oauth/authorize?client_id=legit-app&redirect_uri=https://app.example.com/callback&response_type=code
# No state parameter → attacker can forge authorization requests

# 3. Scope escalation
POST /oauth/token
grant_type=authorization_code&code=AUTH_CODE&scope=read+write+admin
# Original authorization was for "read" scope only — server fails to validate

Mitigation: Validate redirect URIs against a strict whitelist (exact match, not substring). Always require the state parameter. Use PKCE for all public clients. Enforce scope downscoping — tokens can only have scopes equal to or less than what was authorized.

52.3.4 API Key Management¶

API keys are the simplest authentication mechanism — and the most frequently leaked.

Common exposure vectors:

• Hardcoded in frontend JavaScript (visible in browser DevTools)
• Committed to Git repositories (even after deletion — visible in history)
• Logged in server access logs (passed as query parameters)
• Shared in documentation, Slack messages, email
• Embedded in mobile app binaries (extractable via decompilation)

Secure API key practices:

52.4 GraphQL-Specific Attacks¶

GraphQL's flexibility — its single endpoint, client-defined queries, and rich type system — creates a unique attack surface that differs fundamentally from REST.

52.4.1 Introspection Abuse¶

GraphQL introspection allows clients to query the schema itself — every type, field, argument, and relationship. In development, this enables tooling. In production, this is reconnaissance gold.

# Full introspection query — reveals the entire API schema
{
  __schema {
    types {
      name
      fields {
        name
        type {
          name
          kind
          ofType {
            name
          }
        }
        args {
          name
          type {
            name
          }
        }
      }
    }
    mutationType {
      fields {
        name
        args {
          name
          type {
            name
          }
        }
      }
    }
  }
}

What attackers learn from introspection:

• All available queries and mutations (including admin-only operations)
• All types and their fields (including sensitive internal fields)
• Argument names and types (input for fuzzing and injection)
• Relationships between types (navigation paths to sensitive data)
• Deprecated fields that may lack security controls

Detection (KQL):

// Detect GraphQL introspection queries
ApiAccessLogs
| where Endpoint == "/graphql"
| where RequestBody contains "__schema" or RequestBody contains "__type"
| project TimeGenerated, SourceIP, UserAgent, RequestBody
| summarize IntrospectionAttempts = count() by SourceIP, bin(TimeGenerated, 1h)
| where IntrospectionAttempts > 2

Detection (SPL):

index=api_logs endpoint="/graphql" (request_body="*__schema*" OR request_body="*__type*")
| stats count as introspection_attempts by src_ip, _time span=1h
| where introspection_attempts > 2

Mitigation: Disable introspection in production. If introspection is required for internal tooling, restrict it to authenticated admin users or internal network ranges. Use field-level authorization regardless.

52.4.2 Depth and Complexity Attacks¶

GraphQL allows deeply nested queries. Without limits, an attacker can craft a query that causes exponential resource consumption on the server.

# Depth attack — nested relationships cause N+1 query explosion
{
  users(first: 100) {
    orders(first: 100) {
      items(first: 100) {
        product {
          reviews(first: 100) {
            author {
              orders(first: 100) {
                items(first: 100) {
                  product {
                    name
                  }
                }
              }
            }
          }
        }
      }
    }
  }
}
# Depth: 8 levels
# Potential DB queries: 100 * 100 * 100 * 100 * 100 * 100 = 10^12
# Result: Server CPU and memory exhaustion → denial of service

Complexity scoring model:

Complexity = SUM(field_cost * child_count * depth_multiplier)

Example:
  users(first:100)     → cost=1, children=100, depth=1 → 100
    orders(first:50)   → cost=2, children=50, depth=2 → 200 per user → 20,000
      items(first:20)  → cost=1, children=20, depth=3 → 60 per order → 1,200,000

Total complexity: 1,220,100
Maximum allowed: 10,000
Result: Query rejected

Mitigation:

# GraphQL depth and complexity limiting (Python/Graphene example)
from graphql import validate
from graphql.validation import NoSchemaIntrospectionCustomRule

# Configure query depth limit
DEPTH_LIMIT = 5

# Configure complexity limit
COMPLEXITY_LIMIT = 10000

# Configure query timeout
QUERY_TIMEOUT_MS = 5000

# Rate limit per client
QUERIES_PER_MINUTE = 60

52.4.3 Batching Attacks¶

GraphQL supports sending multiple queries in a single HTTP request. Attackers exploit this to bypass rate limiting that counts HTTP requests rather than GraphQL operations.

// Batch query — 100 login attempts in a single HTTP request
[
  {"query": "mutation { login(email: \"user1@example.com\", password: \"REDACTED\") { token }}"},
  {"query": "mutation { login(email: \"user2@example.com\", password: \"REDACTED\") { token }}"},
  {"query": "mutation { login(email: \"user3@example.com\", password: \"REDACTED\") { token }}"},
  // ... 97 more mutations
]
// Rate limiter sees: 1 HTTP request
// Server processes: 100 login attempts

Also exploitable via aliases within a single query:

{
  attempt1: login(email: "user1@example.com", password: "REDACTED") { token }
  attempt2: login(email: "user2@example.com", password: "REDACTED") { token }
  attempt3: login(email: "user3@example.com", password: "REDACTED") { token }
  # ... hundreds of aliases
}

Mitigation: Limit batch query size (maximum operations per request). Count GraphQL operations for rate limiting, not HTTP requests. Limit the number of aliases per query. Apply per-operation rate limits on sensitive mutations.

52.4.4 Field Suggestion Exploitation¶

Many GraphQL servers return helpful error messages that suggest valid field names when a query contains a typo. Attackers exploit this for schema enumeration even when introspection is disabled.

# Attacker sends intentional typos
{ users { emai } }

# Server response:
{
  "errors": [
    {
      "message": "Cannot query field 'emai' on type 'User'. Did you mean 'email', 'emailVerified', 'emailPreferences'?"
    }
  ]
}
# Attacker now knows three valid field names without introspection

Mitigation: Disable field suggestions in production (suggest: false in most GraphQL server configurations). Return generic "field not found" errors without suggestions.

52.4.5 Injection Through Variables¶

GraphQL variables can carry injection payloads that reach backend resolvers if input validation is missing.

# SQL injection through GraphQL variable
mutation {
  searchUsers(filter: {
    name: "testuser' OR '1'='1' --"
    role: "admin"
  }) {
    id
    email
    role
  }
}

# If the resolver constructs SQL from the filter variable:
# SELECT * FROM users WHERE name = 'testuser' OR '1'='1' --' AND role = 'admin'
# Result: Returns all users

Mitigation: Use parameterized queries in all resolvers. Validate GraphQL input types strictly. Apply input validation at the resolver level, not just the schema level.

52.5 Rate Limiting & Resource Exhaustion¶

52.5.1 Rate Limiting Bypass Techniques¶

Rate limiting is the primary defense against brute force, enumeration, and denial-of-service attacks on APIs. Attackers have developed numerous bypass techniques.

Robust rate limiting architecture:

# Multi-layer rate limiting configuration (Kong API Gateway - synthetic)
plugins:
  - name: rate-limiting
    config:
      # Layer 1: Global rate limit
      second: 100
      minute: 3000
      hour: 50000
      policy: redis
      redis_host: 10.0.2.50
      redis_port: 6379

  - name: rate-limiting-advanced
    config:
      # Layer 2: Per-endpoint limits
      limits:
        - path: "/api/v2/auth/login"
          second: 3
          minute: 20
          identifier: consumer  # Per authenticated user
        - path: "/api/v2/auth/reset-password"
          minute: 5
          hour: 10
          identifier: ip
        - path: "/api/v2/users/*/profile"
          second: 10
          minute: 100
          identifier: consumer

  - name: ip-restriction
    config:
      # Block known rate-limit bypass headers from untrusted sources
      deny_xff_override: true  # Ignore X-Forwarded-For from non-trusted proxies

Detection (KQL):

// Detect rate limit bypass via X-Forwarded-For header spoofing
ApiAccessLogs
| where isnotempty(XForwardedFor)
| extend TrueSourceIP = SourceIP
| extend ClaimedIP = split(XForwardedFor, ",")[0]
| where ClaimedIP != TrueSourceIP
| summarize SpoofedRequests = count(),
            DistinctClaimedIPs = dcount(ClaimedIP)
  by TrueSourceIP, Endpoint, bin(TimeGenerated, 5m)
| where DistinctClaimedIPs > 10
| project TimeGenerated, TrueSourceIP, Endpoint, SpoofedRequests, DistinctClaimedIPs

Detection (SPL):

index=api_logs x_forwarded_for=*
| eval true_src=src_ip, claimed_ip=mvindex(split(x_forwarded_for, ","), 0)
| where claimed_ip != true_src
| stats count as spoofed_requests, dc(claimed_ip) as distinct_claimed_ips by true_src, endpoint, _time span=5m
| where distinct_claimed_ips > 10

52.5.2 Pagination Abuse¶

APIs that support pagination can be exploited for data exfiltration and resource exhaustion.

# Normal usage
GET /api/v2/products?page=1&per_page=20

# Abuse pattern 1: Excessive page size
GET /api/v2/products?page=1&per_page=1000000

# Abuse pattern 2: Rapid sequential pagination (data scraping)
GET /api/v2/products?page=1&per_page=100
GET /api/v2/products?page=2&per_page=100
GET /api/v2/products?page=3&per_page=100
# ... automated to scrape entire dataset

# Abuse pattern 3: Cursor manipulation
GET /api/v2/products?cursor=eyJpZCI6MH0=  # Base64: {"id":0} — start from beginning

Mitigation:

• Enforce maximum per_page values (e.g., 100)
• Use cursor-based pagination instead of offset-based (prevents arbitrary position access)
• Rate limit pagination requests per session
• Monitor for sequential full-dataset crawls

52.5.3 Regex Denial of Service (ReDoS) via API Input¶

APIs that apply regex validation to user input are vulnerable to ReDoS if the regex has exponential backtracking.

POST /api/v2/users/validate-email HTTP/1.1
Content-Type: application/json

{
  "email": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa!"
}

# If the server uses a vulnerable regex for email validation:
# ^([a-zA-Z0-9_\-\.]+)+\@([a-zA-Z0-9_\-\.]+)+\.([a-zA-Z]{2,5})$
# The nested quantifiers (+)+ cause catastrophic backtracking
# Processing time: exponential with input length → CPU exhaustion

Mitigation: Use non-backtracking regex engines (RE2). Set regex execution timeouts. Use simple string validation for common patterns (email, URL) instead of complex regex. Test regex patterns against ReDoS payloads before deployment.

52.5.4 API Endpoint Enumeration¶

Attackers systematically discover API endpoints through wordlist-based brute forcing, documentation scraping, and traffic analysis.

# Endpoint enumeration using common API path patterns
# (Synthetic demonstration — authorized testing only)
wordlist=(
  "users" "admin" "auth" "login" "register" "reset" "verify"
  "orders" "payments" "invoices" "reports" "export" "import"
  "config" "settings" "health" "status" "debug" "test"
  "graphql" "swagger" "docs" "openapi" "api-docs"
  "internal" "private" "hidden" "backup" "legacy"
)

for word in "${wordlist[@]}"; do
  for version in v1 v2 v3; do
    status=$(curl -s -o /dev/null -w "%{http_code}" \
      "https://api.megacorp-example.com/api/$version/$word" \
      -H "Authorization: Bearer $TOKEN")
    if [ "$status" != "404" ]; then
      echo "[FOUND] /api/$version/$word → HTTP $status"
    fi
  done
done

Detection (KQL):

// Detect API endpoint enumeration (high 404 rate from single source)
ApiAccessLogs
| where StatusCode == 404
| summarize NotFoundCount = count(),
            DistinctPaths = dcount(Endpoint)
  by SourceIP, bin(TimeGenerated, 10m)
| where NotFoundCount > 50 and DistinctPaths > 20
| project TimeGenerated, SourceIP, NotFoundCount, DistinctPaths

Detection (SPL):

index=api_logs status_code=404
| stats count as not_found_count, dc(endpoint) as distinct_paths by src_ip, _time span=10m
| where not_found_count > 50 AND distinct_paths > 20

52.6 Mass Assignment & SSRF¶

52.6.1 Mass Assignment (Auto-Binding) Attacks¶

Mass assignment occurs when an API automatically binds client-supplied JSON properties to internal object fields without filtering. Attackers add unexpected properties to requests to modify fields they should not have access to.

# Normal profile update request
PUT /api/v2/users/me HTTP/1.1
Authorization: Bearer <token>
Content-Type: application/json

{
  "name": "Test User",
  "email": "testuser@example.com"
}

# Mass assignment attack — adding unauthorized fields
PUT /api/v2/users/me HTTP/1.1
Authorization: Bearer <token>
Content-Type: application/json

{
  "name": "Test User",
  "email": "testuser@example.com",
  "role": "admin",
  "is_verified": true,
  "credit_balance": 999999,
  "subscription_tier": "enterprise",
  "is_staff": true,
  "approved": true
}
# If the API binds all JSON properties to the User model without filtering,
# the attacker has elevated their role to admin and given themselves enterprise access

Detection methodology:

# Compare submitted fields against allowed fields
allowed_update_fields = {"name", "email", "phone", "timezone"}
submitted_fields = set(request.json.keys())
unauthorized_fields = submitted_fields - allowed_update_fields

if unauthorized_fields:
    # Log security event
    log_security_event(
        event="mass_assignment_attempt",
        user_id=current_user.id,
        ip=request.remote_addr,
        unauthorized_fields=list(unauthorized_fields),
        endpoint=request.path
    )
    # Return only allowed fields (fail safe — do not reveal which fields exist)
    return jsonify({"error": "Invalid request fields"}), 400

Defensive code patterns:

# VULNERABLE — binds all request fields to model
@app.route('/api/v2/users/me', methods=['PUT'])
@require_auth
def update_user():
    user = current_user
    for key, value in request.json.items():
        setattr(user, key, value)  # Mass assignment vulnerability
    db.session.commit()
    return jsonify(user.to_dict())

# SECURE — explicit field whitelist
@app.route('/api/v2/users/me', methods=['PUT'])
@require_auth
def update_user():
    user = current_user
    ALLOWED_FIELDS = {'name', 'email', 'phone', 'timezone'}
    for key, value in request.json.items():
        if key in ALLOWED_FIELDS:
            setattr(user, key, value)
    db.session.commit()
    return jsonify(user.to_dict())

52.6.2 Server-Side Request Forgery (SSRF) Through API Parameters¶

SSRF in APIs is particularly dangerous because APIs often have access to internal networks, cloud metadata services, and other backend systems.

Attack scenarios:

# Scenario 1: Cloud metadata access
POST /api/v2/webhooks HTTP/1.1
Authorization: Bearer <token>
Content-Type: application/json

{
  "name": "Order Notifications",
  "callback_url": "http://169.254.169.254/latest/meta-data/iam/security-credentials/"
}
# Server-side: API validates webhook by making a GET request to callback_url
# Result: Returns AWS IAM credentials for the instance role

# Scenario 2: Internal service access
POST /api/v2/documents/fetch HTTP/1.1
Authorization: Bearer <token>
Content-Type: application/json

{
  "url": "http://10.0.1.30:8080/admin/reset-database"
}
# Server fetches URL from internal network — reaches admin interface not exposed externally

# Scenario 3: Port scanning via response timing
POST /api/v2/url-preview HTTP/1.1
Content-Type: application/json

{"url": "http://10.0.1.30:22"}    # SSH — connection refused quickly
{"url": "http://10.0.1.30:8080"}  # HTTP — connection accepted, longer response
{"url": "http://10.0.1.30:3306"}  # MySQL — connection accepted, protocol mismatch
# Response time differences reveal which ports are open on internal hosts

SSRF prevention architecture:

import ipaddress
import urllib.parse

BLOCKED_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('169.254.0.0/16'),     # Link-local / cloud metadata
    ipaddress.ip_network('127.0.0.0/8'),         # Loopback
    ipaddress.ip_network('0.0.0.0/8'),           # Current network
    ipaddress.ip_network('100.64.0.0/10'),       # Shared address space
]

ALLOWED_SCHEMES = {'https'}  # HTTP-only for outbound requests

def validate_url(url: str) -> bool:
    """Validate URL is safe for server-side fetching."""
    parsed = urllib.parse.urlparse(url)

    # Check scheme
    if parsed.scheme not in ALLOWED_SCHEMES:
        return False

    # Resolve hostname to IP
    try:
        resolved_ip = ipaddress.ip_address(socket.gethostbyname(parsed.hostname))
    except (socket.gaierror, ValueError):
        return False

    # Check against blocked ranges
    for blocked in BLOCKED_RANGES:
        if resolved_ip in blocked:
            return False

    # Check for DNS rebinding (resolve again and compare)
    second_resolve = ipaddress.ip_address(socket.gethostbyname(parsed.hostname))
    if second_resolve != resolved_ip:
        return False  # DNS rebinding detected

    return True

Detection (KQL):

// Detect SSRF attempts — API making requests to internal/metadata IPs
ApiOutboundLogs
| where DestinationIP startswith "10." or DestinationIP startswith "172.16."
    or DestinationIP startswith "192.168." or DestinationIP startswith "169.254."
    or DestinationIP == "127.0.0.1"
| where InitiatingEndpoint contains "/api/"
| project TimeGenerated, InitiatingEndpoint, DestinationIP, DestinationPort,
          RequestedURL, SourceUserID
| summarize SSRFAttempts = count() by InitiatingEndpoint, DestinationIP, bin(TimeGenerated, 5m)

Detection (SPL):

index=api_outbound_logs
  (dest_ip="10.*" OR dest_ip="172.16.*" OR dest_ip="192.168.*"
   OR dest_ip="169.254.*" OR dest_ip="127.0.0.1")
  initiating_endpoint="/api/*"
| stats count as ssrf_attempts by initiating_endpoint, dest_ip, _time span=5m
| table _time initiating_endpoint dest_ip ssrf_attempts

52.6.3 Property-Level Authorization¶

Beyond object-level authorization (BOLA), APIs must enforce field-level authorization. Different users may access the same object but see different fields.

// Same GET /api/v2/users/1042 endpoint — different field visibility by role

// Response for the user themselves (role: owner)
{
  "id": 1042,
  "name": "Test User",
  "email": "testuser@example.com",
  "phone": "+1-555-0142",
  "ssn_last4": "4832",
  "account_balance": 1500.00,
  "internal_notes": null  // Should NOT be visible even to owner
}

// Response for another authenticated user (role: peer)
{
  "id": 1042,
  "name": "Test User"
  // email, phone, ssn_last4, balance — all hidden
}

// Response for an admin (role: admin)
{
  "id": 1042,
  "name": "Test User",
  "email": "testuser@example.com",
  "phone": "+1-555-0142",
  "ssn_last4": "4832",
  "account_balance": 1500.00,
  "internal_notes": "Account flagged for review — CS ticket #4401"
}

Implementation pattern:

FIELD_VISIBILITY = {
    'owner': ['id', 'name', 'email', 'phone', 'ssn_last4', 'account_balance'],
    'peer': ['id', 'name'],
    'admin': ['id', 'name', 'email', 'phone', 'ssn_last4', 'account_balance', 'internal_notes'],
    'public': ['id', 'name'],
}

def serialize_user(user, viewer_role):
    allowed_fields = FIELD_VISIBILITY.get(viewer_role, FIELD_VISIBILITY['public'])
    return {field: getattr(user, field) for field in allowed_fields}

52.7 API Security Testing Methodology¶

52.7.1 Phase 1 — Reconnaissance and API Discovery¶

The first step in API security testing is discovering what APIs exist and how they are documented.

API documentation discovery:

# Common API documentation paths
endpoints=(
  "/swagger.json" "/swagger/v1/swagger.json"
  "/api-docs" "/api/docs" "/api/v1/docs" "/api/v2/docs"
  "/openapi.json" "/openapi/v3/api-docs"
  "/.well-known/openapi" "/.well-known/openapi.yaml"
  "/graphql" "/graphiql" "/playground"
  "/api/swagger-resources"
  "/v2/api-docs" "/v3/api-docs"
  "/redoc" "/api/redoc"
  "/_catalog" "/api/health" "/api/status" "/api/version"
  "/actuator" "/actuator/info" "/actuator/env"
)

for endpoint in "${endpoints[@]}"; do
  status=$(curl -s -o /dev/null -w "%{http_code}" \
    "https://api.megacorp-example.com$endpoint")
  if [ "$status" != "404" ] && [ "$status" != "000" ]; then
    echo "[FOUND] $endpoint → HTTP $status"
  fi
done

OpenAPI/Swagger analysis:

# Download and analyze OpenAPI spec
curl -s "https://api.megacorp-example.com/openapi.json" | python3 -c "
import json, sys
spec = json.load(sys.stdin)
print(f'API Version: {spec.get(\"info\", {}).get(\"version\", \"unknown\")}')
print(f'Total Paths: {len(spec.get(\"paths\", {}))}')
for path, methods in spec.get('paths', {}).items():
    for method in methods:
        if method.upper() in ('GET','POST','PUT','DELETE','PATCH'):
            security = methods[method].get('security', [])
            auth = 'AUTH' if security else 'NO-AUTH'
            print(f'  {method.upper():6} {path:50} [{auth}]')
"

52.7.2 Phase 2 — Authentication Testing¶

Test checklist:
[ ] Token generation entropy (minimum 256-bit)
[ ] Token expiration enforcement (exp claim)
[ ] Token revocation (logout invalidates token)
[ ] Refresh token rotation (old refresh tokens invalidated)
[ ] Algorithm confusion (RS256 → HS256)
[ ] None algorithm acceptance
[ ] JWT claim tampering (role, sub, tenant_id)
[ ] API key exposure (client-side, logs, repos)
[ ] OAuth redirect_uri validation (open redirect)
[ ] OAuth state parameter (CSRF protection)
[ ] PKCE enforcement for public clients
[ ] Credential stuffing rate limits
[ ] Account lockout after failed attempts
[ ] MFA bypass testing

52.7.3 Phase 3 — Authorization Testing¶

Test checklist:
[ ] BOLA — horizontal access (user A accessing user B's resources)
[ ] BOLA — vertical access (user accessing admin resources)
[ ] Function-level auth (regular user → admin endpoints)
[ ] Field-level auth (user sees fields beyond their role)
[ ] Mass assignment (user writes to protected fields)
[ ] HTTP method override (GET blocked, PUT/PATCH allowed)
[ ] Path traversal in resource IDs (/api/users/../admin/config)
[ ] GraphQL alias-based enumeration
[ ] Nested resource authorization (parent-child ownership validation)
[ ] Multi-tenant isolation (tenant A accessing tenant B's data)

52.7.4 Phase 4 — Input Validation and Injection Testing¶

# Injection test payloads for API parameters (synthetic)

# SQL Injection
{"search": "testuser' OR '1'='1' --"}
{"filter": {"name": {"$gt": ""}}}  # NoSQL injection (MongoDB)

# Command Injection
{"hostname": "api.example.com; cat /etc/passwd"}
{"filename": "report.pdf|id"}

# SSTI (Server-Side Template Injection)
{"template_name": "{{7*7}}"}
{"greeting": "${7*7}"}

# Path Traversal
{"file": "../../../etc/passwd"}
{"document_id": "..%2f..%2f..%2fetc%2fpasswd"}

# SSRF
{"webhook_url": "http://169.254.169.254/latest/meta-data/"}
{"callback": "http://10.0.1.30:8080/admin"}

# XXE (XML APIs)
{"data": "<?xml version='1.0'?><!DOCTYPE foo [<!ENTITY xxe SYSTEM 'file:///etc/passwd'>]><root>&xxe;</root>"}

52.7.5 Phase 5 — Business Logic Testing¶

Business logic flaws are the most impactful and hardest to detect automatically. They require understanding the application's intended behavior.

Business logic test cases:
[ ] Price manipulation (client-supplied prices in requests)
[ ] Quantity manipulation (negative quantities for refunds)
[ ] Currency confusion (mixing currencies in multi-currency APIs)
[ ] Race conditions (concurrent requests for limited resources)
[ ] Workflow bypass (skip required steps in multi-step processes)
[ ] State manipulation (modify order status directly)
[ ] Coupon/discount stacking beyond intended limits
[ ] Free tier abuse (create multiple accounts for free allocations)
[ ] Referral program gaming (self-referrals)
[ ] Data export limits bypass (export more than allowed)

Race condition exploitation example:

# Race condition test — redeeming a single-use coupon multiple times
# (Synthetic demonstration — authorized testing only)
import asyncio
import aiohttp

async def redeem_coupon(session, token, coupon_code):
    async with session.post(
        'https://api.megacorp-example.com/api/v2/coupons/redeem',
        json={'code': coupon_code},
        headers={'Authorization': f'Bearer {token}'}
    ) as resp:
        return await resp.json()

async def race_test():
    async with aiohttp.ClientSession() as session:
        token = "REDACTED"
        coupon = "SINGLE-USE-COUPON-2024"

        # Send 50 concurrent redemption requests
        tasks = [redeem_coupon(session, token, coupon) for _ in range(50)]
        results = await asyncio.gather(*tasks)

        successes = sum(1 for r in results if r.get('status') == 'redeemed')
        print(f"Successful redemptions: {successes}")
        # If > 1: race condition vulnerability confirmed

52.7.6 Phase 6 — Continuous API Security Testing in CI/CD¶

# GitHub Actions — API Security Testing Pipeline (synthetic)
name: API Security Tests
on:
  pull_request:
    paths:
      - 'src/api/**'
      - 'openapi/**'

jobs:
  schema-validation:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Validate OpenAPI Schema
        run: |
          npm install -g @stoplight/spectral-cli
          spectral lint openapi/api-v2.yaml --ruleset .spectral.yaml

      - name: Schema Breaking Change Detection
        run: |
          # Compare current schema against main branch
          git fetch origin main
          git show origin/main:openapi/api-v2.yaml > /tmp/old-spec.yaml
          npx oasdiff breaking /tmp/old-spec.yaml openapi/api-v2.yaml

  security-scan:
    runs-on: ubuntu-latest
    needs: schema-validation
    steps:
      - name: OWASP ZAP API Scan
        run: |
          docker run --rm \
            -v $(pwd)/openapi:/zap/openapi \
            zaproxy/zap-stable zap-api-scan.py \
            -t openapi/api-v2.yaml \
            -f openapi \
            -r report.html

      - name: Custom BOLA Test Suite
        run: |
          python tests/security/test_bola.py \
            --api-url $API_TEST_URL \
            --spec openapi/api-v2.yaml

      - name: JWT Security Tests
        run: |
          python tests/security/test_jwt.py \
            --api-url $API_TEST_URL \
            --test-algorithms \
            --test-expiration \
            --test-claims

52.8 Detection Engineering for API Attacks¶

This section provides KQL and SPL detection query pairs for the major API attack categories. Each query is designed for production deployment with tuning guidance.

52.8.1 BOLA/IDOR Detection¶

// KQL — Detect BOLA: user accessing multiple object IDs they don't own
ApiAccessLogs
| where HttpMethod in ("GET", "DELETE", "PUT", "PATCH")
| extend ResourceId = extract(@"/api/v\d+/\w+/(\d+)", 1, Endpoint)
| where isnotempty(ResourceId)
| where ResourceId != AuthenticatedUserId
| summarize DistinctResources = dcount(ResourceId),
            TotalRequests = count(),
            Endpoints = make_set(Endpoint, 10)
  by AuthenticatedUserId, SourceIP, bin(TimeGenerated, 15m)
| where DistinctResources > 10
| extend AlertSeverity = iff(DistinctResources > 50, "High", "Medium")
| project TimeGenerated, AuthenticatedUserId, SourceIP, DistinctResources,
          TotalRequests, Endpoints, AlertSeverity

// SPL — Detect BOLA: user accessing multiple object IDs they don't own
index=api_logs http_method IN ("GET", "DELETE", "PUT", "PATCH")
| rex field=endpoint "/api/v\d+/\w+/(?<resource_id>\d+)"
| where isnotnull(resource_id) AND resource_id != authenticated_user_id
| stats dc(resource_id) as distinct_resources, count as total_requests,
        values(endpoint) as endpoints by authenticated_user_id, src_ip, _time span=15m
| where distinct_resources > 10
| eval alert_severity=if(distinct_resources > 50, "High", "Medium")

52.8.2 JWT Attack Detection¶

// KQL — Detect JWT attacks: algorithm manipulation, expired tokens, invalid signatures
ApiAuthLogs
| where AuthMethod == "JWT"
| where AuthResult == "failure"
| extend FailureReason = tostring(parse_json(AuthDetails).reason)
| where FailureReason in ("invalid_algorithm", "signature_mismatch",
                           "token_expired", "none_algorithm", "invalid_issuer")
| summarize FailureCount = count(),
            FailureReasons = make_set(FailureReason)
  by SourceIP, bin(TimeGenerated, 10m)
| where FailureCount > 5
| project TimeGenerated, SourceIP, FailureCount, FailureReasons

// SPL — Detect JWT attacks
index=api_auth_logs auth_method=JWT auth_result=failure
  failure_reason IN ("invalid_algorithm", "signature_mismatch",
                     "token_expired", "none_algorithm", "invalid_issuer")
| stats count as failure_count, values(failure_reason) as failure_reasons by src_ip, _time span=10m
| where failure_count > 5

52.8.3 GraphQL Abuse Detection¶

// KQL — Detect GraphQL abuse: introspection, deep queries, batching
ApiAccessLogs
| where Endpoint == "/graphql"
| extend QueryComplexity = toint(ResponseHeaders.["X-Query-Complexity"])
| extend QueryDepth = toint(ResponseHeaders.["X-Query-Depth"])
| extend IsBatch = RequestBody startswith "["
| extend IsIntrospection = RequestBody contains "__schema"
                        or RequestBody contains "__type"
| where QueryComplexity > 5000 or QueryDepth > 7
    or IsBatch == true or IsIntrospection == true
| summarize AbuseEvents = count(),
            MaxComplexity = max(QueryComplexity),
            MaxDepth = max(QueryDepth),
            BatchCount = countif(IsBatch),
            IntrospectionCount = countif(IsIntrospection)
  by SourceIP, AuthenticatedUserId, bin(TimeGenerated, 10m)
| where AbuseEvents > 3

// SPL — Detect GraphQL abuse
index=api_logs endpoint="/graphql"
| eval is_batch=if(match(request_body, "^\["), 1, 0),
       is_introspection=if(match(request_body, "__schema|__type"), 1, 0)
| stats count as abuse_events,
        max(query_complexity) as max_complexity,
        max(query_depth) as max_depth,
        sum(is_batch) as batch_count,
        sum(is_introspection) as introspection_count
  by src_ip, authenticated_user_id, _time span=10m
| where abuse_events > 3

52.8.4 Mass Assignment Detection¶

// KQL — Detect mass assignment attempts: unexpected fields in PUT/PATCH requests
ApiAccessLogs
| where HttpMethod in ("PUT", "PATCH", "POST")
| extend SubmittedFields = extract_all(@"\"(\w+)\":", RequestBody)
| mv-expand SubmittedField = SubmittedFields
| where SubmittedField in ("role", "is_admin", "is_staff", "permissions",
                            "verified", "approved", "credit_balance",
                            "subscription_tier", "internal_id")
| summarize SuspiciousFields = make_set(SubmittedField),
            AttemptCount = count()
  by AuthenticatedUserId, SourceIP, Endpoint, bin(TimeGenerated, 1h)
| project TimeGenerated, AuthenticatedUserId, SourceIP, Endpoint,
          SuspiciousFields, AttemptCount

// SPL — Detect mass assignment attempts
index=api_logs http_method IN ("PUT", "PATCH", "POST")
| rex field=request_body max_match=20 "\"(?<submitted_field>\w+)\":"
| mvexpand submitted_field
| where submitted_field IN ("role", "is_admin", "is_staff", "permissions",
                             "verified", "approved", "credit_balance",
                             "subscription_tier", "internal_id")
| stats values(submitted_field) as suspicious_fields, count as attempt_count
  by authenticated_user_id, src_ip, endpoint, _time span=1h

52.8.5 SSRF Detection¶

// KQL — Detect SSRF: outbound requests to internal/metadata IPs from API context
ApiOutboundRequests
| where InitiatedBy contains "/api/"
| where ipv4_is_private(DestinationIP)
    or DestinationIP startswith "169.254."
    or DestinationIP == "127.0.0.1"
    or DestinationIP startswith "0."
| project TimeGenerated, InitiatedBy, DestinationIP, DestinationPort,
          DestinationURL, TriggeringUserID, TriggeringRequestBody
| summarize SSRFAttempts = count() by InitiatedBy, DestinationIP, bin(TimeGenerated, 5m)
| where SSRFAttempts >= 1

// SPL — Detect SSRF attempts
index=api_outbound_requests initiated_by="/api/*"
| where cidrmatch("10.0.0.0/8", dest_ip) OR cidrmatch("172.16.0.0/12", dest_ip)
    OR cidrmatch("192.168.0.0/16", dest_ip) OR cidrmatch("169.254.0.0/16", dest_ip)
    OR dest_ip="127.0.0.1"
| stats count as ssrf_attempts by initiated_by, dest_ip, _time span=5m
| where ssrf_attempts >= 1

52.8.6 API Credential Stuffing Detection¶

// KQL — Detect credential stuffing on API auth endpoints
ApiAccessLogs
| where Endpoint in ("/api/v1/auth/login", "/api/v2/auth/login",
                      "/api/v1/auth/token", "/api/v2/auth/token")
| where StatusCode in (401, 403)
| summarize FailedLogins = count(),
            DistinctUsernames = dcount(RequestBody_Username),
            DistinctPasswords = dcount(RequestBody_PasswordHash)
  by SourceIP, bin(TimeGenerated, 5m)
| where FailedLogins > 20 and DistinctUsernames > 5
| extend AttackType = iff(DistinctUsernames > DistinctPasswords,
                          "Credential Stuffing", "Password Spraying")
| project TimeGenerated, SourceIP, FailedLogins, DistinctUsernames, AttackType

// SPL — Detect credential stuffing
index=api_logs endpoint IN ("/api/v1/auth/login", "/api/v2/auth/login",
                             "/api/v1/auth/token", "/api/v2/auth/token")
  status_code IN (401, 403)
| stats count as failed_logins, dc(username) as distinct_usernames,
        dc(password_hash) as distinct_passwords by src_ip, _time span=5m
| where failed_logins > 20 AND distinct_usernames > 5
| eval attack_type=if(distinct_usernames > distinct_passwords,
                      "Credential Stuffing", "Password Spraying")

52.8.7 WebSocket Attack Detection¶

// KQL — Detect WebSocket abuse: rapid messages, large payloads, injection attempts
WebSocketLogs
| where Direction == "inbound"
| extend MessageSize = strlen(MessagePayload)
| extend HasInjection = MessagePayload matches regex @"['\";]|<script|union\s+select|\.\./"
| summarize MessagesPerSecond = count() / 60.0,
            MaxMessageSize = max(MessageSize),
            InjectionAttempts = countif(HasInjection),
            TotalMessages = count()
  by ConnectionId, SourceIP, bin(TimeGenerated, 1m)
| where MessagesPerSecond > 100 or MaxMessageSize > 1048576
    or InjectionAttempts > 0

// SPL — Detect WebSocket abuse
index=websocket_logs direction=inbound
| eval message_size=len(message_payload),
       has_injection=if(match(message_payload, "['\";\<]|script|union\s+select|\.\.\/"), 1, 0)
| stats count as total_messages, max(message_size) as max_message_size,
        sum(has_injection) as injection_attempts by connection_id, src_ip, _time span=1m
| eval messages_per_second=total_messages/60
| where messages_per_second > 100 OR max_message_size > 1048576 OR injection_attempts > 0

52.8.8 API Key Abuse Detection¶

// KQL — Detect API key abuse: key used from multiple geographies, excessive rate
ApiAccessLogs
| where AuthMethod == "API_KEY"
| extend GeoLocation = geo_info_from_ip_address(SourceIP).country
| summarize DistinctCountries = dcount(GeoLocation),
            DistinctIPs = dcount(SourceIP),
            TotalRequests = count(),
            Countries = make_set(GeoLocation, 10)
  by ApiKeyId, bin(TimeGenerated, 1h)
| where DistinctCountries > 3 or DistinctIPs > 10 or TotalRequests > 10000
| project TimeGenerated, ApiKeyId, DistinctCountries, DistinctIPs,
          TotalRequests, Countries

// SPL — Detect API key abuse
index=api_logs auth_method=API_KEY
| iplocation src_ip
| stats dc(Country) as distinct_countries, dc(src_ip) as distinct_ips,
        count as total_requests, values(Country) as countries
  by api_key_id, _time span=1h
| where distinct_countries > 3 OR distinct_ips > 10 OR total_requests > 10000

52.9 API Security Architecture¶

52.9.1 API Gateway Security Configurations¶

API gateways are the first line of defense for API security. Proper configuration is critical.

Kong API Gateway security configuration (synthetic):

# Kong declarative configuration — security plugins
_format_version: "3.0"

services:
  - name: user-service
    url: http://10.0.1.10:8080
    routes:
      - name: user-routes
        paths:
          - /api/v2/users
        protocols:
          - https
    plugins:
      - name: jwt
        config:
          claims_to_verify:
            - exp
            - iss
          key_claim_name: iss
          maximum_expiration: 3600

      - name: acl
        config:
          allow:
            - users-group
            - admin-group

      - name: rate-limiting
        config:
          second: 10
          minute: 100
          hour: 5000
          policy: redis
          redis_host: 10.0.2.50

      - name: request-size-limiting
        config:
          allowed_payload_size: 1  # 1 MB max

      - name: cors
        config:
          origins:
            - https://app.megacorp-example.com
          methods:
            - GET
            - POST
            - PUT
            - DELETE
          headers:
            - Authorization
            - Content-Type
          credentials: true
          max_age: 3600

      - name: bot-detection
        config:
          deny:
            - "python-requests"
            - "curl"
            - "wget"
          # Note: User-Agent filtering is easily bypassed — defense in depth only

      - name: response-transformer
        config:
          remove:
            headers:
              - X-Powered-By
              - Server
          add:
            headers:
              - "X-Content-Type-Options: nosniff"
              - "X-Frame-Options: DENY"
              - "Strict-Transport-Security: max-age=31536000; includeSubDomains"