diff --git a/.env.example b/.env.example index 8d2ce90..6e2e418 100644 --- a/.env.example +++ b/.env.example @@ -57,8 +57,53 @@ PORT=8000 CONCURRENCY_LIMIT=20 # Seconds per HTTP request before timeout. FETCH_TIMEOUT=20.0 -# Shannon-entropy floor (bits/char). Raise to 4.0 to cut Gemini cost / false matches. +# Shannon-entropy floor (bits/char) for the GENERIC keyword=value catch-all. +# Raise to 4.0 to cut Gemini cost / false matches. MIN_ENTROPY_THRESHOLD=3.5 +# Low anti-degenerate entropy floor for STRUCTURAL/provider detectors (AKIA…, ghp_…, +# sk_live_…). Only rejects obvious junk (e.g. all-identical chars); keep it well below +# MIN_ENTROPY_THRESHOLD so genuinely modest-entropy live keys are not dropped. +MIN_STRUCTURAL_ENTROPY=2.5 + +# ── Passive reconnaissance (subdomain enumeration) ────────────────────────── +# Certificate Transparency front-end used to enumerate subdomains passively. +# Discovery never contacts the target — only this CT source. +CRTSH_URL=https://crt.sh +# Seconds to wait for the CT-log query before failing closed (empty result). +SUBDOMAIN_ENUM_TIMEOUT=30 +# Safety cap on how many discovered subdomains to return. +MAX_SUBDOMAINS=500 +# ── Multi-target deep scan (domain → enumerate → probe → scan each host) ───── +# Max live hosts scanned in one --deep-scan run. +MAX_TARGETS=25 +# Parallel liveness probes when checking which discovered hosts are up. +PROBE_CONCURRENCY=10 +# Seconds per liveness probe before a host is treated as unreachable. +PROBE_TIMEOUT=10 +# How many hosts a deep scan scans in parallel. Each host scan is itself +# concurrent, so keep this modest on a Raspberry Pi (3 is a good default). +HOST_SCAN_CONCURRENCY=3 +# ── Historical path discovery (passive, from public web archives) ─────────── +# Recover historically-exposed URLs (the passive alternative to brute-forcing). +WAYBACK_CDX_URL=http://web.archive.org/cdx/search/cdx +COMMONCRAWL_COLLINFO=https://index.commoncrawl.org/collinfo.json +# Set false to skip CommonCrawl and use the Wayback Machine only (faster). +ENABLE_COMMONCRAWL=true +HISTORICAL_TIMEOUT=30 +HISTORICAL_RETRIES=2 +# Safety cap on how many historical URLs to return. +MAX_HISTORICAL_URLS=2000 +# Cap on externally-supplied seed assets fetched per scan (e.g. historical JS +# bundles fed into a --deep-scan --with-historical run). +MAX_SEED_URLS=200 +# ── Surface intelligence (endpoints referenced in code + associated hosts) ── +# Mine fetched JS/HTML for referenced endpoints and external hosts, and fetch +# same-site .js endpoints one level deeper. Set false to disable. +EXTRACT_SURFACE=true +# Same-site .js endpoints (referenced in code) to fetch for the deeper crawl. +MAX_ENDPOINT_SEEDS=50 +# Cap on how many discovered endpoints to store/report. +MAX_DISCOVERED_ENDPOINTS=300 # Skip JS assets larger than this many bytes (default 5 MB). MAX_ASSET_BYTES=5242880 # Only surface AI-confirmed findings at or above this confidence (0-100). diff --git a/.gitignore b/.gitignore index 5767c5b..55463dc 100644 --- a/.gitignore +++ b/.gitignore @@ -18,6 +18,7 @@ venv/ backend/data/*.db backend/data/*.db-wal backend/data/*.db-shm +backend/data/*.db-journal *.sqlite3 # OS / editor diff --git a/CHANGELOG.md b/CHANGELOG.md index d42657b..32da708 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,102 @@ All notable changes to SecretNode are documented here. This project adheres to [Semantic Versioning](https://semver.org/). +## [2.7.0] — Deep attack-surface platform (passive) + +SecretNode grows from a single-URL secret scanner into a **passive attack-surface platform**: give it +a domain and it enumerates the subdomain surface, recovers historically-exposed URLs from public +archives, mines JavaScript for referenced endpoints and third-party hosts, probes which hosts are +live, checks each for **subdomain-takeover risk**, and scans them all — live pages *and* archived +bundles — for exposed credentials and misconfigurations, aggregated into one reviewable report and +drivable from the CLI **or** the dashboard. Every layer stays passive and authorized-scope only. +Test suite **187 → 270**, all green; ruff clean. See `docs/TECHNICAL-AUDIT-AND-ROADMAP.md`. + +### Changed +- **Concurrent host orchestration (deep-dive slice D5).** A domain deep scan now scans its hosts in + parallel with a bounded semaphore (`HOST_SCAN_CONCURRENCY`, default 3) instead of one at a time — + a large multi-host domain finishes far faster. Results are collected in target order, per-host + error isolation is preserved (one host failing never sinks the run), and progress is emitted as + `[k/N] host — done` events. A test proves the parallelism is real *and* stays within the bound. + +### Added +- **Subdomain-takeover detection (deep-dive slice D1).** New `backend/takeover.py` flags hosts whose + DNS still points (via CNAME) at an **unclaimed third-party service** (S3, GitHub Pages, Heroku, + Netlify, Shopify, Fastly, Zendesk, …) — a hijackable subdomain an attacker can claim to serve + content from the target's domain. High-precision by design: a host is flagged only when the + response carries a service's *specific* unclaimed-resource signature (generic 404s excluded), with + the CNAME recorded as corroborating evidence. The deep scan runs a concurrent takeover pass over + every in-scope host; results surface as CRITICAL/HIGH findings with a "Subdomain Takeover Risks" + section + KPI in the combined report. Passive (DNS + one GET), stdlib-only, ReDoS-free. +- **Surface intelligence: endpoints + associated-host graph (deep-ASM slices 5 & 4).** New + `backend/surface.py` mines every fetched asset (passively, no new target requests) for two things: + **(5)** URLs/paths referenced in the JavaScript — `fetch()`/`axios` targets, `/api/…` routes a live + page crawl never links to — and then fetches same-site `.js` endpoints **one level deeper** so + code-referenced bundles get secret-scanned too; and **(4)** the external hosts each asset talks to + (CDNs, APIs, third parties), aggregated into an **associated-asset graph**. `run_scan` now returns + `discovered_endpoints` + `associated_hosts`; both reports gain an "Attack Surface Intelligence" + section. Extractor regexes are bounded/ReDoS-safe. Config: `EXTRACT_SURFACE`, `MAX_ENDPOINT_SEEDS`, + `MAX_DISCOVERED_ENDPOINTS`. +- **Dashboard domain-mode + deep-scan API (deep-ASM slice 6).** The whole deep-ASM pipeline is now + drivable from the web UI, not just the CLI. New `POST /api/deep-scans` runs a domain-wide deep scan + as a streaming background task — per-host progress flows over the existing `/ws/logs/{scan_id}` + WebSocket (`run_deep_scan` gained a `broadcast` hook and emits enumerate/probe/per-host/complete + events), and the report endpoint serves the combined multi-target report for deep results. + Frontend: a **DEEP toggle** turns the target box into a whole-domain scan (bare domain in → + enumerate + historical + probe + scan-all), finalising on `deep_scan_complete` rather than + per-host. API tests added (route, auth, input caps, start). Passive; authorized-scope only. +- **Historical bundles fed into the scan (deep-ASM slice 3.5).** `run_scan()` gains a `seed_urls` + parameter — externally-supplied asset URLs are fetched and scanned alongside the live crawl, + deduped against it (capped by `MAX_SEED_URLS`). `run_deep_scan(include_historical=True)` now + recovers the domain's historical JS bundles (Wayback/CommonCrawl) and routes each host its own + archived bundles as seeds, so a secret in a forgotten bundle **no live page links to** still gets + fetched and confirmed. CLI: `python cli.py --deep-scan --with-historical`; the combined + report gains a "Historical URLs" metric. This turns discovery into findings — the payoff of the + whole passive discovery chain. +- **Historical path discovery (deep-ASM slice 3).** New `backend/historical.py` recovers a domain's + historically-exposed URLs from **public web archives (Wayback Machine + CommonCrawl)** — the + passive alternative to directory/content brute-forcing, so no request ever touches the target. Two + sources merged with backoff retries and fail-closed handling (matching the subdomain layer); + surfaces forgotten endpoints, stale JS bundles and old admin paths a live crawl would never link + to. `HistoricalResult` exposes the raw URLs, the unique-path view ("hidden directories"), and a + `js_urls()` helper (highest-value scan seeds). CLI: `python cli.py --historical`. Config: + `WAYBACK_CDX_URL`, `COMMONCRAWL_COLLINFO`, `ENABLE_COMMONCRAWL`, `HISTORICAL_TIMEOUT`, + `HISTORICAL_RETRIES`, `MAX_HISTORICAL_URLS`. +- **Multi-target orchestration (deep-ASM slice 2).** New `backend/orchestrator.py` closes the loop + from discovery to findings: a single domain → passive subdomain enumeration → liveness probe of + each host → the existing passive secret+posture scan per live host → one aggregated + `DeepScanResult`. Includes a per-host **SSRF guard** (a discovered host that resolves to a + private/internal address is skipped unless `ALLOW_PRIVATE_TARGETS=true`), a `MAX_TARGETS` cap, + concurrent probing, and per-host error isolation (one host failing never sinks the run). New + combined client report `report.generate_deep_scan_html()` (subdomain surface + live hosts + + per-host confirmed/needs-review/posture). CLI: `python cli.py --deep-scan -o report.html`. + Config: `MAX_TARGETS`, `PROBE_CONCURRENCY`, `PROBE_TIMEOUT`. +- **Passive subdomain enumeration (deep-ASM slice 1).** New `backend/recon.py` expands a domain + into its known subdomain surface from **Certificate Transparency** — fully passive, it never + contacts the target, so it runs before a client engagement is signed. Queries **two independent CT + sources (crt.sh + Certspotter)** with backoff retries and merges them, so a single flaky/rate- + limited source (crt.sh 502s often) no longer zeroes out a good result; the result lists which + sources succeeded and only reports an error if *all* fail. `extract_registrable_domain()` + normalises URL/host/IP inputs (two-label public-suffix table incl. `.bd`). Exposed via the CLI: + `python cli.py --subdomains`. First layer of the passive attack-surface pipeline + (subdomains → historical paths → associated assets → existing secret/posture scan). + +### Fixed +- **False-negative: structural keys wrongly entropy-gated.** The Shannon-entropy floor + (`MIN_ENTROPY_THRESHOLD=3.5`) was applied uniformly to every detector, silently dropping + genuinely low-entropy but well-formed provider keys (e.g. an AWS key ID at ~3.27 bits) before + they ever reached AI validation — the worst failure mode for a scanner. Entropy is now + class-aware: the *generic* keyword=value catch-all keeps the full 3.5 bar, while + *structural/provider* detectors (AKIA…, ghp_…, sk_live_…, PEM, fixed-format tokens) only clear a + low anti-degenerate floor (`MIN_STRUCTURAL_ENTROPY=2.5`) that still rejects obvious junk like + `AKIAAAAAAAAAAAAAAAAA`. Precision/recall stays 1.000/1.000. +- **False-negative: AI-dismissed structural matches silently dropped.** A finding was routed to + manual review only when AI validation was *unavailable*; a structural/provider match the AI + *actively* rejected with a real confidence matched no bucket and was discarded — so a live key the + AI merely under-called on (e.g. lacking page context) vanished with no trace. New + `classify_validated()` sends any structural match the AI does **not confidently dismiss** to + manual review instead of dropping it; the generic catch-all keeps aggressive filtering, so the + "no false positives in Confirmed" promise holds. Suite **187 → 197**. + ## [2.6.0] — Detection quality, safety & attack-surface breadth A measured capability pass grounded in a fresh audit vs 2026 secret-scanning SOTA diff --git a/Makefile b/Makefile index 58f5182..11f0173 100644 --- a/Makefile +++ b/Makefile @@ -20,7 +20,10 @@ bench: ## Measure detection-layer precision/recall on the labelled corpus (R2) cd backend && SECRETNODE_API_KEY=bench python -m bench.run_bench run: ## Start the server (requires .env with SECRETNODE_API_KEY) - cd backend && uvicorn main:app --host 0.0.0.0 --port 8000 --loop uvloop + # `python -m uvicorn` uses the same interpreter that runs the CLI, so it works + # even when the `uvicorn` console script isn't on PATH; --loop auto uses uvloop + # when available and falls back to asyncio otherwise (no hard uvloop dependency). + cd backend && python -m uvicorn main:app --host 0.0.0.0 --port 8000 --loop auto docker: ## Build and run via docker compose docker compose up --build diff --git a/backend/cli.py b/backend/cli.py index ed2b500..fcca0ae 100644 --- a/backend/cli.py +++ b/backend/cli.py @@ -27,6 +27,9 @@ import sys from urllib.parse import urlparse +import historical +import orchestrator +import recon import report import scanner @@ -78,11 +81,120 @@ def build_parser() -> argparse.ArgumentParser: ap.add_argument("-o", "--output", help="Write report to this file (default: stdout)") ap.add_argument("--fail-on-findings", action="store_true", help="Exit non-zero if any confirmed findings (use as a CI gate)") + ap.add_argument("--subdomains", action="store_true", + help="Passive attack-surface discovery: enumerate the target's subdomains " + "via Certificate Transparency (crt.sh) and print them. Never contacts " + "the target. Authorized use only.") + ap.add_argument("--historical", action="store_true", + help="Passive path discovery: recover historically-exposed URLs for the domain " + "from public web archives (Wayback Machine + CommonCrawl) and print them. " + "Never contacts the target. Authorized use only.") + ap.add_argument("--deep-scan", dest="deep_scan", action="store_true", + help="Domain-wide deep scan: enumerate subdomains, probe which hosts are live, " + "scan each, and write a combined report. Passive; authorized use only.") + ap.add_argument("--max-targets", dest="max_targets", type=int, default=orchestrator.MAX_TARGETS, + help=f"Max live hosts to scan in a --deep-scan run (default: {orchestrator.MAX_TARGETS})") + ap.add_argument("--with-historical", dest="with_historical", action="store_true", + help="In a --deep-scan, also recover historical JS bundles from public archives " + "(Wayback/CommonCrawl) and scan them as seeds — catches secrets in forgotten " + "bundles no live page links to. Slower; passive.") return ap +async def _run_deep_scan(args) -> int: + """Domain-wide deep scan: enumerate → probe live hosts → scan each → combined report.""" + result = await orchestrator.run_deep_scan( + args.target, + max_crawl_pages=max(1, args.crawl), + verify=args.verify, + only_verified=args.only_verified, + max_targets=max(1, args.max_targets), + include_historical=args.with_historical, + ) + if args.output: + report_fmt = (args.format if args.format in ("json",) else "html") + body = (json.dumps(result.to_dict(), indent=2) if report_fmt == "json" + else report.generate_deep_scan_html(result.to_dict())) + with open(args.output, "w", encoding="utf-8") as fh: + fh.write(body) + print(f"Deep-scan report written to {args.output}", file=sys.stderr) + else: + sys.stdout.write(json.dumps(result.to_dict(), indent=2) + "\n") + + t = result.to_dict()["totals"] + hist = f", {t['historical_urls']} historical URL(s)" if t.get("historical_urls") else "" + print( + f"SecretNode deep scan of {result.domain}: {t['subdomains']} subdomain(s), " + f"{t['live_hosts']} live, {t['hosts_scanned']} scanned{hist} — " + f"{t['confirmed']} confirmed, {t['needs_review']} needs-review, " + f"{t['posture_issues']} posture issue(s).", + file=sys.stderr, + ) + if args.fail_on_findings and t["confirmed"]: + return 1 + return 0 + + +async def _run_subdomain_enum(target: str) -> int: + """Passive subdomain discovery mode: expand a domain into its known subdomain + surface from Certificate Transparency and print the results as JSON.""" + domain = recon.extract_registrable_domain(target) + if domain is None: + raise SystemExit( + f"Cannot enumerate subdomains for {target!r} — need a domain " + "(subdomain enumeration does not apply to bare IP addresses)." + ) + async with scanner.build_client() as client: + result = await recon.enumerate_subdomains(client, domain) + print(json.dumps(result.to_dict(), indent=2)) + sources = ", ".join(result.sources) if result.sources else "none" + print( + f"SecretNode: discovered {result.count} subdomain(s) for {domain} " + f"(sources: {sources})." + (f" [error: {result.error}]" if result.error else ""), + file=sys.stderr, + ) + return 0 + + +async def _run_historical(target: str) -> int: + """Passive path discovery: recover historically-exposed URLs from public + archives (Wayback + CommonCrawl) — the passive alternative to brute-forcing.""" + domain = recon.extract_registrable_domain(target) + if domain is None: + raise SystemExit( + f"Cannot run historical discovery for {target!r} — need a domain " + "(does not apply to bare IP addresses)." + ) + async with scanner.build_client() as client: + result = await historical.discover_historical_urls(client, domain) + print(json.dumps(result.to_dict(), indent=2)) + sources = ", ".join(result.sources) if result.sources else "none" + print( + f"SecretNode: recovered {result.count} historical URL(s) " + f"({len(result.paths)} unique path(s), {len(result.js_urls())} JS) for {domain} " + f"(sources: {sources})." + (f" [error: {result.error}]" if result.error else ""), + file=sys.stderr, + ) + return 0 + + def main(argv: list[str] | None = None) -> int: args = build_parser().parse_args(argv) + + # Passive discovery mode: enumerate subdomains and exit. Accepts a bare domain + # (no scheme) since it never fetches the target — only Certificate Transparency. + if args.subdomains: + return asyncio.run(_run_subdomain_enum(args.target)) + + # Passive historical path discovery from public archives; accepts a bare domain. + if args.historical: + return asyncio.run(_run_historical(args.target)) + + # Domain-wide deep scan: enumerate → probe → scan each live host. Accepts a + # bare domain; the orchestrator applies the SSRF guard per discovered host. + if args.deep_scan: + return asyncio.run(_run_deep_scan(args)) + if not args.target.startswith(("http://", "https://")): raise SystemExit("Target must start with http:// or https://") assert_public_target(args.target) diff --git a/backend/historical.py b/backend/historical.py new file mode 100644 index 0000000..995452d --- /dev/null +++ b/backend/historical.py @@ -0,0 +1,262 @@ +#!/usr/bin/env python3 +""" +SecretNode historical path discovery — deep-ASM slice 3. + +The passive answer to directory/content brute-forcing. Instead of hammering a +target with guessed paths (active, noisy, needs a signed RoE), we recover the +URLs it has *already exposed* from public web archives: + + • the Wayback Machine (Internet Archive) CDX index, and + • CommonCrawl's URL index. + +Both are third-party archives — no request is ever sent to the target. This +surfaces forgotten endpoints, old admin panels, stale JS bundles and API paths +that a live crawl of the current site would never link to, which is exactly +where credentials tend to linger. + +Posture rules mirror the rest of the scanner: passive only, fails closed (any +source erroring yields an empty contribution, never an exception), and +authorized-scope use only. +""" + +from __future__ import annotations + +import asyncio +import json +import logging +import os +from dataclasses import dataclass, field +from urllib.parse import urlparse + +import httpx + +logger = logging.getLogger("secretnode.historical") + + +def _env_int(name: str, default: int) -> int: + try: + return int(os.environ.get(name, "").strip() or default) + except ValueError: + return default + + +WAYBACK_CDX_URL = os.environ.get("WAYBACK_CDX_URL", "http://web.archive.org/cdx/search/cdx") +COMMONCRAWL_COLLINFO = os.environ.get("COMMONCRAWL_COLLINFO", "https://index.commoncrawl.org/collinfo.json") +HISTORICAL_TIMEOUT = _env_int("HISTORICAL_TIMEOUT", 30) +HISTORICAL_RETRIES = _env_int("HISTORICAL_RETRIES", 2) +MAX_HISTORICAL_URLS = _env_int("MAX_HISTORICAL_URLS", 2000) +ENABLE_COMMONCRAWL = os.environ.get("ENABLE_COMMONCRAWL", "true").lower() == "true" + +_TRANSIENT_STATUS: frozenset[int] = frozenset({429, 500, 502, 503, 504}) + + +@dataclass +class HistoricalResult: + """Historical URLs recovered for a domain from public archives.""" + domain: str + urls: list[str] = field(default_factory=list) + sources: list[str] = field(default_factory=list) + error: str | None = None + + @property + def count(self) -> int: + return len(self.urls) + + @property + def paths(self) -> list[str]: + """Unique URL paths across all discovered URLs — the 'hidden directories' + view (each distinct path an archive has seen for this domain).""" + seen: set[str] = set() + for u in self.urls: + p = urlparse(u).path or "/" + seen.add(p) + return sorted(seen) + + def js_urls(self) -> list[str]: + """Discovered URLs that look like JavaScript — the highest-value scan + seeds, since bundled JS is where secrets most often leak. + + Deduplicated by (host, path): archives store the same file under many + cache-buster query strings (app.js?v=1, app.js?v=2, …); scanning each + variant re-finds the same secrets, so we keep only one URL per unique + file. This is what collapses e.g. 11 api_data.js?v=… variants to one.""" + seen: set[tuple[str, str]] = set() + out: list[str] = [] + for u in self.urls: + p = urlparse(u) + if not p.path.lower().endswith(".js"): + continue + key = ((p.hostname or "").lower(), p.path) + if key not in seen: + seen.add(key) + out.append(u) + return out + + def to_dict(self) -> dict: + return { + "domain": self.domain, + "urls": self.urls, + "count": self.count, + "paths": self.paths, + "js_urls": self.js_urls(), + "sources": self.sources, + "error": self.error, + } + + +def _in_scope(url: str, domain: str) -> bool: + """True if `url`'s host is `domain` or a subdomain of it.""" + try: + host = (urlparse(url).hostname or "").lower() + except ValueError: + return False + return bool(host) and (host == domain or host.endswith("." + domain)) + + +def parse_wayback_cdx(payload: object, domain: str) -> list[str]: + """Parse a Wayback CDX `output=json&fl=original` response into sorted, in-scope + URLs. The response is a JSON array whose first row is the header ['original']. + Pure function.""" + domain = (domain or "").strip().lower().rstrip(".") + if isinstance(payload, (str, bytes)): + try: + payload = json.loads(payload) + except (ValueError, TypeError): + return [] + if not isinstance(payload, list): + return [] + + found: set[str] = set() + for row in payload: + if not isinstance(row, list) or not row: + continue + url = str(row[0]).strip() + if not url or url == "original": # skip the CDX header row + continue + if _in_scope(url, domain): + found.add(url) + return sorted(found) + + +def parse_commoncrawl_jsonl(text: str, domain: str) -> list[str]: + """Parse a CommonCrawl CDX JSONL response (one JSON object per line, each with + a `url` field) into sorted, in-scope URLs. Pure function.""" + domain = (domain or "").strip().lower().rstrip(".") + found: set[str] = set() + for line in (text or "").splitlines(): + line = line.strip() + if not line: + continue + try: + record = json.loads(line) + except (ValueError, TypeError): + continue + url = str(record.get("url", "")).strip() if isinstance(record, dict) else "" + if url and _in_scope(url, domain): + found.add(url) + return sorted(found) + + +async def _get_with_retries( + client: httpx.AsyncClient, url: str, +) -> tuple[httpx.Response | None, str | None]: + """GET with backoff retries on transient statuses/timeouts. Returns + (response, None) or (None, error-with-type-name).""" + last_err = "no attempt made" + for attempt in range(HISTORICAL_RETRIES + 1): + try: + resp = await client.get( + url, timeout=httpx.Timeout(HISTORICAL_TIMEOUT, connect=10.0), + ) + if resp.status_code in _TRANSIENT_STATUS and attempt < HISTORICAL_RETRIES: + last_err = f"HTTP {resp.status_code}" + await asyncio.sleep(2 ** attempt) + continue + return resp, None + except (httpx.HTTPError, httpx.InvalidURL) as exc: + last_err = f"{type(exc).__name__}: {exc}".rstrip(": ").strip() + if attempt < HISTORICAL_RETRIES: + await asyncio.sleep(2 ** attempt) + continue + return None, last_err + + +async def fetch_wayback( + client: httpx.AsyncClient, domain: str, *, limit: int, +) -> tuple[set[str], str | None]: + # matchType=domain covers the apex AND every subdomain in the archive. + url = (f"{WAYBACK_CDX_URL}?url={domain}&matchType=domain&output=json" + f"&fl=original&collapse=urlkey&limit={limit}") + resp, err = await _get_with_retries(client, url) + if resp is None: + return set(), err + if resp.status_code != 200: + return set(), f"HTTP {resp.status_code}" + return set(parse_wayback_cdx(resp.text, domain)), None + + +async def fetch_commoncrawl( + client: httpx.AsyncClient, domain: str, *, limit: int, +) -> tuple[set[str], str | None]: + """CommonCrawl needs two hops: discover the newest index's CDX API, then query + it. Both fail closed.""" + resp, err = await _get_with_retries(client, COMMONCRAWL_COLLINFO) + if resp is None: + return set(), err + if resp.status_code != 200: + return set(), f"collinfo HTTP {resp.status_code}" + try: + indexes = json.loads(resp.text) + cdx_api = indexes[0]["cdx-api"] # newest index is first + except (ValueError, TypeError, KeyError, IndexError) as exc: + return set(), f"collinfo parse: {type(exc).__name__}" + + resp2, err2 = await _get_with_retries( + client, f"{cdx_api}?url={domain}&matchType=domain&output=json&limit={limit}") + if resp2 is None: + return set(), err2 + if resp2.status_code != 200: + return set(), f"HTTP {resp2.status_code}" + return set(parse_commoncrawl_jsonl(resp2.text, domain)), None + + +async def discover_historical_urls( + client: httpx.AsyncClient, + domain: str, + *, + limit: int = MAX_HISTORICAL_URLS, + enable_commoncrawl: bool = ENABLE_COMMONCRAWL, +) -> HistoricalResult: + """Recover a domain's historically-exposed URLs from public archives + (Wayback + optionally CommonCrawl), merged and deduplicated. + + Never contacts the target. Fails closed: `error` is set only when every + enabled source fails; if any source returns data the result is usable.""" + domain = (domain or "").strip().lower().rstrip(".") + if not domain: + return HistoricalResult(domain=domain, error="empty domain") + + sources: list[tuple[str, object]] = [("wayback", fetch_wayback)] + if enable_commoncrawl: + sources.append(("commoncrawl", fetch_commoncrawl)) + + all_urls: set[str] = set() + ok_sources: list[str] = [] + errors: list[str] = [] + for name, fetch in sources: + try: + urls, err = await fetch(client, domain, limit=limit) + except Exception as exc: # a source must never crash the run + urls, err = set(), f"{type(exc).__name__}: {exc}".rstrip(": ").strip() + if urls: + all_urls |= urls + ok_sources.append(name) + if err: + errors.append(f"{name}: {err}") + + urls = sorted(all_urls) + if len(urls) > limit: + urls = urls[:limit] + + error = None if ok_sources else ("; ".join(errors) or "no sources returned data") + return HistoricalResult(domain=domain, urls=urls, sources=ok_sources, error=error) diff --git a/backend/main.py b/backend/main.py index 3c3f9cd..5f643e3 100644 --- a/backend/main.py +++ b/backend/main.py @@ -43,6 +43,7 @@ # so the speed-up is preserved without a module-level global side effect. from scanner import run_scan, ScanState +import orchestrator from storage import ( init_db, save_scan, load_scans, load_scan, get_previous_scan_for_target, mark_false_positive, unmark_false_positive, get_suppressed_fingerprints, @@ -374,6 +375,95 @@ async def _run() -> None: } +class DeepScanRequest(BaseModel): + """Domain-wide deep scan: enumerate → probe → scan each live host.""" + domain: str + crawl_pages: int = DEFAULT_CRAWL_PAGES + verify: bool = False + only_verified: bool = False + include_historical: bool = False + max_targets: int = orchestrator.MAX_TARGETS + + @field_validator("domain") + @classmethod + def validate_domain(cls, v: str) -> str: + v = v.strip() + if not v or len(v) > 253: + raise ValueError("Provide a domain (or host/URL) up to 253 characters.") + return v + + @field_validator("crawl_pages") + @classmethod + def _cap_crawl(cls, v: int) -> int: + return max(1, min(v, MAX_CRAWL_PAGES_CAP)) + + @field_validator("max_targets") + @classmethod + def _cap_targets(cls, v: int) -> int: + return max(1, min(v, 100)) + + +@app.post("/api/deep-scans", status_code=202, dependencies=[Depends(require_api_key)]) +async def start_deep_scan(request: DeepScanRequest, http_request: Request) -> dict[str, Any]: + """Start a background domain-wide deep scan (enumerate → probe → scan each + live host). Returns a scan_id; connect to /ws/logs/{scan_id} to stream + progress. Discovery is passive; authorized-scope use only.""" + client_ip = http_request.client.host if http_request.client else "unknown" + logger.info("AUDIT deep_scan_request client=%s domain=%s", client_ip, request.domain) + + active_count = sum(1 for e in _registry.values() if not e["task"].done()) + if active_count >= MAX_CONCURRENT_SCANS: + raise HTTPException( + status_code=429, + detail=f"{active_count} scan(s) already running (limit: {MAX_CONCURRENT_SCANS}).", + ) + + scan_id = str(uuid.uuid4()) + + async def broadcaster(event: dict[str, Any]) -> None: + await manager.broadcast_scan(scan_id, event) + + async def _run() -> None: + try: + deep = await orchestrator.run_deep_scan( + request.domain, + max_crawl_pages=request.crawl_pages, + verify=request.verify, + only_verified=request.only_verified, + include_historical=request.include_historical, + max_targets=request.max_targets, + broadcast=broadcaster, + ) + result = deep.to_dict() + result["scan_id"] = scan_id + result["target_url"] = deep.domain + result["deep_scan"] = True + totals = result.get("totals", {}) + result["status"] = ("complete" + if (totals.get("confirmed") or totals.get("needs_review") + or totals.get("posture_issues")) else "clean") + _registry[scan_id]["meta"] = result + await save_scan(scan_id, result) + except asyncio.CancelledError: + await manager.broadcast_scan(scan_id, {"type": "scan_cancelled", "scan_id": scan_id}) + except Exception as exc: + logger.exception("Deep scan %s crashed", scan_id) + await manager.broadcast_scan(scan_id, { + "type": "scan_error", "scan_id": scan_id, "error": str(exc)}) + + task = asyncio.create_task(_run(), name=f"deep-scan-{scan_id}") + _registry[scan_id] = { + "state": ScanState(), + "task": task, + "meta": {"scan_id": scan_id, "target_url": request.domain, + "status": "running", "deep_scan": True}, + } + logger.info("Deep scan %s started for %s", scan_id, request.domain) + return {"scan_id": scan_id, "domain": request.domain, "status": "started", + "ws_url": f"/ws/logs/{scan_id}", + "message": "Connect to ws_url to receive live events."} + + @app.post("/api/scans/{scan_id}/stop", status_code=200, dependencies=[Depends(require_api_key)]) async def stop_scan(scan_id: str) -> dict[str, Any]: """Signal a running scan to stop cooperatively.""" @@ -481,7 +571,10 @@ async def get_scan_report( ) if format == "html": - body = report_gen.generate_html_report(scan, agency_name=agency_name) + # A domain-wide deep scan aggregates many hosts — render the combined + # multi-target report instead of the single-target one. + body = (report_gen.generate_deep_scan_html(scan) if scan.get("deep_scan") + else report_gen.generate_html_report(scan, agency_name=agency_name)) return HTMLResponse(content=body, headers={ "Content-Disposition": f'inline; filename="secretnode_report_{scan_id[:8]}.html"' }) diff --git a/backend/orchestrator.py b/backend/orchestrator.py new file mode 100644 index 0000000..43224f3 --- /dev/null +++ b/backend/orchestrator.py @@ -0,0 +1,352 @@ +#!/usr/bin/env python3 +""" +SecretNode multi-target orchestration — deep-ASM slice 2. + +Closes the loop from discovery to findings: given a single domain, expand it to +its subdomain surface (passive CT enumeration), probe which of those hosts are +actually live, then run the existing passive secret+posture scan against each +one and aggregate the results into a single deliverable. + +Everything here stays within SecretNode's passive/authorized posture: + • Enumeration is passive (Certificate Transparency, never the target). + • Liveness probing and scanning are the same passive fetches the single-target + scanner already performs — no exploitation, no brute-force, read-only. + • Authorized use only. Scanning a whole domain's host list at once magnifies + the responsibility: run it only against assets you own or are explicitly + permitted to assess (owned / signed RoE / in-scope program). + +Collaborators (enumeration, client factory, per-host scan) are injected so the +orchestration is unit-testable without any network. +""" + +from __future__ import annotations + +import asyncio +import ipaddress +import logging +import os +import socket +from dataclasses import dataclass, field +from typing import Awaitable, Callable + +import httpx + +import historical +import recon +import scanner +import takeover + +logger = logging.getLogger("secretnode.orchestrator") + + +def _env_int(name: str, default: int) -> int: + try: + return int(os.environ.get(name, "").strip() or default) + except ValueError: + return default + + +MAX_TARGETS = _env_int("MAX_TARGETS", 25) # cap hosts scanned per run +PROBE_CONCURRENCY = _env_int("PROBE_CONCURRENCY", 10) # parallel liveness probes +PROBE_TIMEOUT = _env_int("PROBE_TIMEOUT", 10) # seconds per liveness probe +HOST_SCAN_CONCURRENCY = _env_int("HOST_SCAN_CONCURRENCY", 3) # hosts scanned in parallel — each + # host scan is itself concurrent, so keep this modest on a Pi + +# Type aliases for the injected collaborators. +EnumerateFn = Callable[..., Awaitable["recon.SubdomainResult"]] +ScanFn = Callable[..., Awaitable[dict]] +ClientFactory = Callable[[], httpx.AsyncClient] + + +@dataclass +class HostScan: + """Per-host outcome inside a deep scan.""" + host: str + url: str + confirmed: int = 0 + needs_review: int = 0 + posture_issues: int = 0 + assets: int = 0 + error: str | None = None + + def to_dict(self) -> dict: + return { + "host": self.host, "url": self.url, "confirmed": self.confirmed, + "needs_review": self.needs_review, "posture_issues": self.posture_issues, + "assets": self.assets, "error": self.error, + } + + +@dataclass +class DeepScanResult: + """Aggregate of a domain-wide deep scan.""" + domain: str + subdomains: list[str] = field(default_factory=list) + enum_sources: list[str] = field(default_factory=list) + live_hosts: list[str] = field(default_factory=list) + hosts: list[HostScan] = field(default_factory=list) + scans: list[dict] = field(default_factory=list) # raw per-host scan dicts + historical_urls: int = 0 # historical URLs discovered (0 if not requested) + takeover_findings: list[dict] = field(default_factory=list) # dangling-CNAME hijack risks + error: str | None = None + + @property + def total_confirmed(self) -> int: + return sum(h.confirmed for h in self.hosts) + + @property + def total_needs_review(self) -> int: + return sum(h.needs_review for h in self.hosts) + + @property + def total_posture(self) -> int: + return sum(h.posture_issues for h in self.hosts) + + def _aggregate(self, key: str) -> list[dict]: + """Flatten a per-host finding list across all scans, tagging each finding + with the host it came from so the combined report can show provenance.""" + out: list[dict] = [] + for scan in self.scans: + host = recon._host_of(scan.get("target_url", "")) or scan.get("target_url", "") + for f in scan.get(key, []): + out.append({**f, "_host": host}) + return out + + def to_dict(self) -> dict: + return { + "domain": self.domain, + "subdomains": self.subdomains, + "enum_sources": self.enum_sources, + "live_hosts": self.live_hosts, + "hosts": [h.to_dict() for h in self.hosts], + "historical_urls": self.historical_urls, + "confirmed_findings": self._aggregate("confirmed_findings"), + "needs_review_findings": self._aggregate("needs_review_findings"), + "associated_hosts": sorted({h for s in self.scans + for h in s.get("associated_hosts", [])}), + "takeover_findings": self.takeover_findings, + "totals": { + "subdomains": len(self.subdomains), + "live_hosts": len(self.live_hosts), + "hosts_scanned": len(self.hosts), + "historical_urls": self.historical_urls, + "confirmed": self.total_confirmed, + "needs_review": self.total_needs_review, + "posture_issues": self.total_posture, + "takeover_risks": len(self.takeover_findings), + }, + "error": self.error, + } + + +def _target_ip_class(host: str) -> str: + """Classify a host by the address it resolves to: 'public', 'private' + (loopback/private/link-local/reserved/multicast — an SSRF risk), or + 'unresolved'. Used to keep a discovered host list from ever pointing the + scanner at internal infrastructure.""" + try: + infos = socket.getaddrinfo(host, None) + except socket.gaierror: + return "unresolved" + for info in infos: + ip = ipaddress.ip_address(info[4][0]) + if (ip.is_private or ip.is_loopback or ip.is_link_local + or ip.is_reserved or ip.is_multicast): + return "private" + return "public" + + +async def _probe_one(client: httpx.AsyncClient, host: str) -> str | None: + """Return the reachable base URL for a host (preferring https), or None if it + is unreachable. ANY HTTP response — including 401/403/5xx — means the host is + live and worth scanning; only a transport error (DNS/connect/timeout) is dead.""" + for scheme in ("https", "http"): + url = f"{scheme}://{host}" + try: + await client.get(url, timeout=httpx.Timeout(PROBE_TIMEOUT, connect=10.0)) + return url + except httpx.HTTPError: + continue + return None + + +async def probe_live_hosts( + client: httpx.AsyncClient, hosts: list[str], *, concurrency: int = PROBE_CONCURRENCY, +) -> list[str]: + """Concurrently probe hosts; return the reachable base URLs, order preserved.""" + sem = asyncio.Semaphore(max(1, concurrency)) + + async def _guarded(h: str) -> str | None: + async with sem: + return await _probe_one(client, h) + + results = await asyncio.gather(*(_guarded(h) for h in hosts)) + return [u for u in results if u] + + +def _summarise_scan(host: str, url: str, scan: dict) -> HostScan: + return HostScan( + host=host, + url=url, + confirmed=len(scan.get("confirmed_findings", [])), + needs_review=len(scan.get("needs_review_findings", [])), + posture_issues=len(scan.get("posture_findings", [])), + assets=int(scan.get("assets_fetched", 0) or 0), + error=scan.get("error"), + ) + + +async def run_deep_scan( + target: str, + *, + max_crawl_pages: int = 1, + verify: bool = False, + only_verified: bool = False, + max_targets: int = MAX_TARGETS, + include_historical: bool = False, + broadcast: Callable[[dict], Awaitable[None]] | None = None, + enumerate_fn: EnumerateFn = recon.enumerate_subdomains, + scan_fn: ScanFn = scanner.run_scan, + client_factory: ClientFactory = scanner.build_client, + discover_historical_fn: Callable[..., Awaitable["historical.HistoricalResult"]] + = historical.discover_historical_urls, + takeover_fn: Callable[..., Awaitable[list]] = takeover.scan_hosts_for_takeover, +) -> DeepScanResult: + """Domain → enumerate → probe → scan each live host → aggregate. + + Falls back to scanning the bare target itself when the input is an IP or has + no enumerable domain, so a deep scan always does *something* useful.""" + domain = recon.extract_registrable_domain(target) + if domain is None: + # No enumerable domain (e.g. an IP): degrade to a single passive scan of + # the given target so the deep-scan entry point is still usable. + host = recon._host_of(target) or target + url = target if "://" in target else f"https://{host}" + result = DeepScanResult(domain=host) + scan = await scan_fn(target_url=url, max_crawl_pages=max_crawl_pages, + verify=verify, only_verified=only_verified, broadcast=broadcast) + result.live_hosts = [url] + result.hosts = [_summarise_scan(host, url, scan)] + result.scans = [scan] + return result + + async def emit(event: dict) -> None: + if broadcast: + await broadcast(event) + + def log(msg: str, level: str = "INFO") -> dict: + return {"type": "log", "level": level, "message": msg} + + original_host = recon._host_of(target) + result = DeepScanResult(domain=domain) + async with client_factory() as client: + await emit(log(f"Deep scan of {domain} — enumerating subdomains (Certificate Transparency)…")) + enum = await enumerate_fn(client, domain) + result.subdomains = enum.subdomains + result.enum_sources = enum.sources + await emit(log(f"Enumerated {len(enum.subdomains)} subdomain(s)" + + (f" via {', '.join(enum.sources)}" if enum.sources else ""))) + + # Optional: recover historical URLs (Wayback/CommonCrawl) once for the + # domain. They enrich BOTH host discovery (hostnames seen in the archive) + # and per-host scan seeds (archived JS bundles) — so a flaky CT source no + # longer zeroes out the run, and forgotten bundles still get scanned. + js_by_host: dict[str, list[str]] = {} + hist_hosts: list[str] = [] + if include_historical: + await emit(log("Recovering historical URLs from public archives (Wayback/CommonCrawl)…")) + hist = await discover_historical_fn(client, domain) + result.historical_urls = hist.count + hist_hosts = [recon._host_of(u) for u in hist.urls] + for u in hist.js_urls(): + js_by_host.setdefault(recon._host_of(u), []).append(u) + await emit(log(f"Recovered {hist.count} historical URL(s); " + f"{sum(len(v) for v in js_by_host.values())} archived JS seed(s)")) + + # Candidate hosts, in-scope and deduped. The host the caller actually + # typed is ALWAYS included first — enumeration must never be able to drop + # the specified target — followed by the apex, CT subdomains, and any + # hosts seen in the archive. + candidates = [ + h for h in dict.fromkeys( + [original_host, domain, *enum.subdomains, *hist_hosts]) + if h and (h == domain or h.endswith("." + domain)) + ] + + # SSRF guard: never probe/scan a discovered host that resolves to an + # internal address (a wildcard/misissued cert can name one). Bypassed + # only by the same ALLOW_PRIVATE_TARGETS opt-in the single-target path uses. + allow_private = os.environ.get("ALLOW_PRIVATE_TARGETS", "false").lower() == "true" + safe_hosts: list[str] = [] + for host in candidates: + cls = "public" if allow_private else await asyncio.to_thread(_target_ip_class, host) + if cls == "public": + safe_hosts.append(host) + elif cls == "private": + result.hosts.append(HostScan( + host=host, url=f"https://{host}", + error="skipped: resolves to a private/internal address (SSRF guard)")) + # 'unresolved' hosts are silently dropped — nothing to scan. + + # Subdomain-takeover pass (D1): check every in-scope host for a dangling + # CNAME pointing at an unclaimed third-party service — a hijackable + # subdomain. Runs on the full candidate set (a takeover target often is + # not a "normal" live host). Best-effort; never sinks the run. + if safe_hosts: + await emit(log(f"Checking {len(safe_hosts)} host(s) for subdomain-takeover risk…")) + try: + tos = await takeover_fn(client, safe_hosts) + except Exception as exc: # defensive + logger.debug("takeover pass failed: %s", exc) + tos = [] + result.takeover_findings = [t.to_dict() if hasattr(t, "to_dict") else t for t in tos] + if result.takeover_findings: + await emit(log(f"⚠ {len(result.takeover_findings)} potential subdomain " + f"takeover(s) found", "WARN")) + + await emit(log(f"Probing {len(safe_hosts)} candidate host(s) for liveness…")) + result.live_hosts = await probe_live_hosts(client, safe_hosts) + + if not result.live_hosts: + result.error = enum.error or "no live hosts found" + await emit(log("No live hosts to scan.", "WARN")) + await emit({"type": "deep_scan_complete", "totals": result.to_dict()["totals"]}) + return result + + targets = result.live_hosts[:max(1, max_targets)] + n = len(targets) + await emit(log(f"{len(result.live_hosts)} host(s) live — scanning {n} " + f"(concurrency {HOST_SCAN_CONCURRENCY})")) + + # Scan hosts concurrently with a bounded semaphore instead of one-at-a-time. + # Results are collected in target order; a single host failing is isolated + # to that host and never sinks the run. + sem = asyncio.Semaphore(max(1, HOST_SCAN_CONCURRENCY)) + done = {"n": 0} + + async def _scan_host(url: str) -> tuple[HostScan, dict | None]: + host = recon._host_of(url) + async with sem: + try: + scan = await scan_fn(target_url=url, max_crawl_pages=max_crawl_pages, + verify=verify, only_verified=only_verified, + seed_urls=js_by_host.get(host, []), broadcast=broadcast) + except Exception as exc: # a single host must not sink the whole run + logger.debug("deep scan: host %s failed: %s", url, exc) + done["n"] += 1 + await emit(log(f"[{done['n']}/{n}] {host} — error: {type(exc).__name__}", "WARN")) + return HostScan(host=host, url=url, + error=f"{type(exc).__name__}: {exc}".strip(": ")), None + done["n"] += 1 + await emit(log(f"[{done['n']}/{n}] {host} — done " + f"({len(scan.get('confirmed_findings', []))} confirmed)")) + return _summarise_scan(host, url, scan), scan + + pairs = await asyncio.gather(*(_scan_host(u) for u in targets)) + for host_scan, scan in pairs: + result.hosts.append(host_scan) + if scan is not None: + result.scans.append(scan) + + await emit({"type": "deep_scan_complete", "totals": result.to_dict()["totals"]}) + return result diff --git a/backend/recon.py b/backend/recon.py new file mode 100644 index 0000000..a45fb60 --- /dev/null +++ b/backend/recon.py @@ -0,0 +1,297 @@ +#!/usr/bin/env python3 +""" +SecretNode passive reconnaissance — attack-surface discovery. + +The first discovery layer of the deep-ASM pipeline: given a domain, expand it +into its full known subdomain surface using **passive public data only**, so no +packet is ever sent to the target itself. This slice sources subdomains from +Certificate Transparency (crt.sh) — every publicly-trusted TLS certificate is +logged there, which makes CT the single richest passive source of a domain's +hostnames. + +Design rules (mirror the scanner's posture): + • Passive only — we query third-party CT logs, never the target. Nothing here + resolves DNS, connects to, or probes the target host. + • Fails closed — any network/parse error yields an empty result, never an + exception that would abort a scan. Discovery is best-effort by nature. + • Authorized use only — enumerating a domain's surface is reconnaissance; only + run it against assets you own or are explicitly permitted to assess. +""" + +from __future__ import annotations + +import asyncio +import ipaddress +import json +import logging +import os +from dataclasses import dataclass, field +from urllib.parse import urlparse + +import httpx + +logger = logging.getLogger("secretnode.recon") + + +def _env_int(name: str, default: int) -> int: + try: + return int(os.environ.get(name, "").strip() or default) + except ValueError: + return default + + +# crt.sh is the CT-log front-end; %25 is a URL-encoded '%' wildcard so the query +# "%.example.com" returns every logged hostname under the domain. +CRTSH_URL = os.environ.get("CRTSH_URL", "https://crt.sh").rstrip("/") +# Certspotter is a second, independent CT aggregator. Querying more than one +# passive source is how industrial enumerators (subfinder/amass) stay reliable +# when any single source is rate-limited or down — crt.sh in particular returns +# 502/timeout often enough that a lone source is not production-grade. +CERTSPOTTER_URL = os.environ.get("CERTSPOTTER_URL", "https://api.certspotter.com").rstrip("/") +SUBDOMAIN_TIMEOUT = _env_int("SUBDOMAIN_ENUM_TIMEOUT", 30) +MAX_SUBDOMAINS = _env_int("MAX_SUBDOMAINS", 500) # safety cap on result size +ENUM_RETRIES = _env_int("SUBDOMAIN_ENUM_RETRIES", 2) # retries on transient CT errors + +# Transient HTTP statuses worth retrying — crt.sh/Certspotter throw these under +# load; a retry with backoff usually clears them. +_TRANSIENT_STATUS: frozenset[int] = frozenset({429, 500, 502, 503, 504}) + +# A small public-suffix table so we can find the *registrable* domain of a host +# without a heavyweight PSL dependency. Covers the common two-label suffixes, +# with Bangladesh (Cindrasec's home market) explicitly included. Anything not +# listed falls back to the last two labels, which is correct for gTLDs. +_TWO_LEVEL_SUFFIXES: frozenset[str] = frozenset({ + "co.uk", "org.uk", "gov.uk", "ac.uk", "me.uk", "net.uk", "sch.uk", + "com.au", "net.au", "org.au", "edu.au", "gov.au", + "co.nz", "net.nz", "org.nz", "govt.nz", + "com.bd", "net.bd", "org.bd", "edu.bd", "gov.bd", "ac.bd", "mil.bd", + "co.in", "net.in", "org.in", "gen.in", "firm.in", "ind.in", + "com.br", "net.br", "org.br", "gov.br", + "com.sg", "com.my", "com.pk", "com.np", "com.lk", "com.cn", "com.hk", + "co.jp", "or.jp", "ne.jp", "co.kr", "co.za", "co.id", "co.th", +}) + + +@dataclass +class SubdomainResult: + """Outcome of a passive subdomain enumeration for one registrable domain. + + `sources` lists the passive sources that actually returned data; `error` is + set only when *every* source failed (so a caller can distinguish "the domain + has no subdomains" from "enumeration could not run").""" + domain: str + subdomains: list[str] = field(default_factory=list) + sources: list[str] = field(default_factory=list) + error: str | None = None + + @property + def count(self) -> int: + return len(self.subdomains) + + def to_dict(self) -> dict: + return { + "domain": self.domain, + "subdomains": self.subdomains, + "count": self.count, + "sources": self.sources, + "error": self.error, + } + + +def _host_of(target: str) -> str: + """Extract the bare hostname from a URL, host[:port], or bare domain.""" + target = (target or "").strip() + if not target: + return "" + if "://" in target: + target = urlparse(target).hostname or "" + else: + # Strip a trailing path and any :port, but keep bracketed IPv6 intact. + target = target.split("/", 1)[0] + if target.count(":") == 1: # host:port (not IPv6, which has many) + target = target.split(":", 1)[0] + return target.strip().strip(".").lower() + + +def is_ip_literal(target: str) -> bool: + """True if the target's host is an IP address (subdomain enum does not apply).""" + host = _host_of(target).strip("[]") + try: + ipaddress.ip_address(host) + return True + except ValueError: + return False + + +def extract_registrable_domain(target: str) -> str | None: + """Reduce a URL / host / bare domain to its registrable domain + (e.g. https://api.blog.example.co.uk/x -> example.co.uk). Returns None for IP + literals and inputs with no usable hostname, since neither can be enumerated.""" + host = _host_of(target) + if not host or is_ip_literal(target): + return None + labels = [x for x in host.split(".") if x] + if len(labels) < 2: + return None + last_two = ".".join(labels[-2:]) + if last_two in _TWO_LEVEL_SUFFIXES and len(labels) >= 3: + return ".".join(labels[-3:]) + return last_two + + +def _clean_name(name: str, domain: str) -> str | None: + """Normalise one CT 'name_value' entry into a single in-scope hostname, or + None if it is not a usable subdomain of `domain`.""" + n = (name or "").strip().lower().rstrip(".") + if n.startswith("*."): # wildcard cert — the base host is what matters + n = n[2:] + if not n or "@" in n or " " in n: # skip emails / malformed entries + return None + if n == domain or n.endswith("." + domain): + return n + return None + + +def parse_crtsh_json(payload: object, domain: str) -> list[str]: + """Parse a crt.sh JSON response into a sorted, deduplicated list of in-scope + hostnames. Pure function (no I/O) so it is cheap and deterministic to test. + + Accepts either an already-decoded list or a raw JSON string. crt.sh packs one + or more newline-separated names into each record's `name_value`, plus a + `common_name`; we harvest every name from both fields.""" + domain = (domain or "").strip().lower().rstrip(".") + if isinstance(payload, (str, bytes)): + try: + payload = json.loads(payload) + except (ValueError, TypeError): + return [] + if not isinstance(payload, list): + return [] + + found: set[str] = set() + for record in payload: + if not isinstance(record, dict): + continue + raw_names = str(record.get("name_value", "")).split("\n") + raw_names.append(str(record.get("common_name", ""))) + for raw in raw_names: + host = _clean_name(raw, domain) + if host: + found.add(host) + return sorted(found) + + +def parse_certspotter_json(payload: object, domain: str) -> list[str]: + """Parse a Certspotter `/v1/issuances` response into sorted, in-scope + hostnames. Each issuance carries a `dns_names` list. Pure function.""" + domain = (domain or "").strip().lower().rstrip(".") + if isinstance(payload, (str, bytes)): + try: + payload = json.loads(payload) + except (ValueError, TypeError): + return [] + if not isinstance(payload, list): + return [] + + found: set[str] = set() + for record in payload: + if not isinstance(record, dict): + continue + for raw in record.get("dns_names", []) or []: + host = _clean_name(str(raw), domain) + if host: + found.add(host) + return sorted(found) + + +async def _get_with_retries( + client: httpx.AsyncClient, url: str, *, headers: dict | None = None, +) -> tuple[httpx.Response | None, str | None]: + """GET with backoff retries on transient statuses/timeouts. Returns + (response, None) on a final response, or (None, error) if every attempt + failed to produce one. Always names the failure type so errors are never + blank (the crt.sh timeout showed up as an empty string before).""" + last_err = "no attempt made" + for attempt in range(ENUM_RETRIES + 1): + try: + resp = await client.get( + url, headers=headers or {}, + timeout=httpx.Timeout(SUBDOMAIN_TIMEOUT, connect=10.0), + ) + if resp.status_code in _TRANSIENT_STATUS and attempt < ENUM_RETRIES: + last_err = f"HTTP {resp.status_code}" + await asyncio.sleep(2 ** attempt) + continue + return resp, None + except (httpx.HTTPError, httpx.InvalidURL) as exc: + last_err = f"{type(exc).__name__}: {exc}".rstrip(": ").strip() + if attempt < ENUM_RETRIES: + await asyncio.sleep(2 ** attempt) + continue + return None, last_err + + +async def _fetch_crtsh(client: httpx.AsyncClient, domain: str) -> tuple[set[str], str | None]: + resp, err = await _get_with_retries(client, f"{CRTSH_URL}/?q=%25.{domain}&output=json") + if resp is None: + return set(), err + if resp.status_code != 200: + return set(), f"HTTP {resp.status_code}" + return set(parse_crtsh_json(resp.text, domain)), None + + +async def _fetch_certspotter(client: httpx.AsyncClient, domain: str) -> tuple[set[str], str | None]: + url = (f"{CERTSPOTTER_URL}/v1/issuances?domain={domain}" + "&include_subdomains=true&expand=dns_names") + resp, err = await _get_with_retries(client, url) + if resp is None: + return set(), err + if resp.status_code != 200: + return set(), f"HTTP {resp.status_code}" + return set(parse_certspotter_json(resp.text, domain)), None + + +async def enumerate_subdomains( + client: httpx.AsyncClient, + domain: str, + *, + limit: int = MAX_SUBDOMAINS, +) -> SubdomainResult: + """Passively enumerate subdomains of `domain` from multiple Certificate + Transparency sources (crt.sh + Certspotter), merged and deduplicated. + + Never contacts the target. Fails closed: `error` is set only if *every* + source failed; if any source returns data the result is usable and lists the + sources that succeeded. Querying several sources is what keeps enumeration + reliable when one is rate-limited or down.""" + domain = (domain or "").strip().lower().rstrip(".") + if not domain: + return SubdomainResult(domain=domain, error="empty domain") + + all_hosts: set[str] = set() + ok_sources: list[str] = [] + errors: list[str] = [] + for name, fetch in (("crt.sh", _fetch_crtsh), ("certspotter", _fetch_certspotter)): + try: + hosts, err = await fetch(client, domain) + except Exception as exc: # defensive: a source must never crash the run + hosts, err = set(), f"{type(exc).__name__}: {exc}".rstrip(": ").strip() + if hosts: + all_hosts |= hosts + ok_sources.append(name) + if err: + errors.append(f"{name}: {err}") + + subs = sorted(all_hosts) + if len(subs) > limit: + logger.debug("enumeration returned %d subdomains for %s; capping at %d", + len(subs), domain, limit) + subs = subs[:limit] + + error = None if ok_sources else ("; ".join(errors) or "no sources returned data") + return SubdomainResult(domain=domain, subdomains=subs, sources=ok_sources, error=error) + + +# Backward-compatible alias: the original single-source name now maps to the +# multi-source enumerator. +enumerate_subdomains_ct = enumerate_subdomains diff --git a/backend/report.py b/backend/report.py index fde435a..ee96b9a 100644 --- a/backend/report.py +++ b/backend/report.py @@ -45,7 +45,7 @@ def _tool_version() -> str: return s.split("=", 1)[1].strip().strip('"').strip("'") except Exception: pass - return "2.6.0" + return "2.7.0" _TOOL_VERSION = _tool_version() @@ -226,6 +226,17 @@ def _posture_row(p: dict[str, Any]) -> str: posture_html = "\n".join(_posture_row(p) for p in sorted(posture, key=_sort_key)) or \ 'No header/misconfiguration issues detected.' + # Attack-surface intelligence (slices 5 & 4): endpoints referenced in code and + # the external hosts this asset talks to. Capped for readability. + endpoints = scan.get("discovered_endpoints", []) or [] + assoc_hosts = scan.get("associated_hosts", []) or [] + endpoints_html = "".join( + f'
  • {html.escape(e)}
    • ' for e in endpoints[:80] + ) or '
  • None discovered.
    • ' + if len(endpoints) > 80: + endpoints_html += f'
  • … and {len(endpoints) - 80} more.
    • ' + hosts_html = ", ".join(html.escape(h) for h in assoc_hosts) or "—" + return f""" @@ -349,12 +360,19 @@ def _posture_row(p: dict[str, Any]) -> str: {posture_html} -

    Flagged for Manual Review (AI validation unavailable)

    +

    Flagged for Manual Review (AI unavailable, or a structural match it could not confidently clear)

    {review_html}
    SeverityTypeLocationNote
    +

    Attack Surface Intelligence

    +
    Associated hosts (external services this + asset references — CDNs, APIs, third parties): {hosts_html}
    +
    Endpoints referenced in code ({len(endpoints)} discovered — URLs/paths + the JavaScript calls that a page crawl would miss):
    +
      {endpoints_html}
    +