diff --git a/.env.example b/.env.example index 759077c..8d2ce90 100644 --- a/.env.example +++ b/.env.example @@ -87,3 +87,14 @@ VERIFY_SECRETS=false FOLLOW_SOURCE_MAPS=true # Cap on source maps fetched per scan. MAX_SOURCE_MAPS=40 +# Decode a source map's embedded original source (sourcesContent) and scan it as +# real code — better per-file attribution, catches secrets escaped in the raw +# .map JSON, and avoids the map's high-entropy "mappings" blob. (R5) +SCAN_SOURCEMAP_CONTENT=true +# Cap on original source files decoded per source map. +MAX_SOURCEMAP_SOURCES=200 +# +# Passive HTTP security-posture check (R8): analyse the target root's own +# response for missing/weak security headers, version disclosure, and insecure +# cookies. Pure analysis of a response the target already serves. +SCAN_HTTP_POSTURE=true diff --git a/Makefile b/Makefile index 9d7c46c..58f5182 100644 --- a/Makefile +++ b/Makefile @@ -1,5 +1,5 @@ # SecretNode — developer & operator shortcuts -.PHONY: help setup test lint run docker clean +.PHONY: help setup test lint bench run docker clean help: ## Show this help @grep -E '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) | awk 'BEGIN{FS=":.*?## "}{printf " \033[36m%-10s\033[0m %s\n", $$1, $$2}' @@ -16,6 +16,9 @@ test: ## Run the full test suite lint: ## Run the ruff correctness lint ruff check backend/ +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 diff --git a/README.md b/README.md index 477be55..5c3ed18 100644 --- a/README.md +++ b/README.md @@ -54,7 +54,7 @@ Pipeline: **browser-like spider (+ source-map mining) → regex (54 patterns) + > **v2.3.0 — ASM-industry alignment: verification-first & CI-native** > Grounded in 2025–2026 ASM / secret-scanning practice (verification-first detection + CI-native gating). -> - **Optional live verification** (`VERIFY_SECRETS` / `?verify=true`) — read-only "is this key still active?" checks against each secret's own provider (GitHub, GitLab, Stripe, SendGrid, OpenAI, Slack, npm, Mailgun, Telegram), à la TruffleHog `--only-verified`. **Off by default**, fails closed, never touches the scan target. Each finding gains a `verified` status, and `only_verified` mode drops dead-key noise. +> - **Optional live verification** (`VERIFY_SECRETS` / `?verify=true`) — read-only "is this key still active?" checks against each secret's own provider (GitHub, GitLab, Stripe, SendGrid, OpenAI, Slack, npm, Mailgun, Telegram, Cloudflare, DigitalOcean, Datadog, Notion, Linear, Figma, Postman, Doppler), à la TruffleHog `--only-verified`. **Off by default**, fails closed, never touches the scan target. Each finding gains a `verified` status, and `only_verified` mode drops dead-key noise. > - **Base64 decoding** of encoded blobs + **example/placeholder allowlisting** (e.g. AWS's `AKIAIOSFODNN7EXAMPLE`) — fewer false positives, more real catches. > - **CLI (`backend/cli.py`) + GitHub Action (`action.yml`)** — scan and emit SARIF from CI with `--fail-on-findings` as a build gate. > - **Registry now 44 patterns** (added Slack app-level, GitHub server/refresh, OpenAI service-account, New Relic, Grafana, HCP Terraform). @@ -245,7 +245,7 @@ Square · Postman · Databricks · Telegram Bot · Discord Bot · Datadog · Fir **MEDIUM** — Stripe Publishable Key · Bearer Token · Generic High-Entropy Secret -Matches are also checked against **base64-decoded** content and filtered through an **example/placeholder allowlist**. Many types (GitHub, GitLab, Stripe, SendGrid, OpenAI, Slack, npm, Mailgun, Telegram) can be **live-verified** (see below). New patterns land with a `severity`, `cwe`, and `remediation` — see [`CONTRIBUTING.md`](CONTRIBUTING.md). +Matches are also checked against **base64-decoded** content and filtered through an **example/placeholder allowlist**. Many types (GitHub, GitLab, Stripe, SendGrid, OpenAI, Slack, npm, Mailgun, Telegram, Cloudflare, DigitalOcean, Datadog, Notion, Linear, Figma, Postman, Doppler) can be **live-verified** (see below). New patterns land with a `severity`, `cwe`, and `remediation` — see [`CONTRIBUTING.md`](CONTRIBUTING.md). --- @@ -311,7 +311,7 @@ false-positive fatigue. - **Off by default.** Enable per scan (`{"verify": true}` / `--verify`) or globally (`VERIFY_SECRETS=true`). - **Read-only.** One "whoami"-style call to the secret's **own provider** (never the scan target): - GitHub, GitLab, Stripe, SendGrid, OpenAI, Slack, npm, Mailgun, Telegram. Fails closed on any error. + GitHub, GitLab, Stripe, SendGrid, OpenAI, Slack, npm, Mailgun, Telegram, Cloudflare, DigitalOcean, Datadog, Notion, Linear, Figma, Postman, Doppler. Fails closed on any error. - Each finding gets a `verified` status: `verified` (active), `unverified` (dead / unconfirmed), `unsupported` (no safe auto-check — verify manually). - **`only_verified`** drops confirmed-inactive findings so a pipeline only fails on live secrets. diff --git a/backend/bench/__init__.py b/backend/bench/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/bench/corpus.py b/backend/bench/corpus.py new file mode 100644 index 0000000..01b9fb9 --- /dev/null +++ b/backend/bench/corpus.py @@ -0,0 +1,116 @@ +""" +SecretNode — bench/corpus.py (R2: FP/FN benchmark corpus) + +A small, labelled corpus for measuring the deterministic detection layer +(regex + placeholder allowlist + entropy gate + base64 pass — everything in +extract_secrets(), no network, no AI). It answers, reproducibly: + + • RECALL — of real secrets planted in code, how many does the layer catch? + • PRECISION — of everything it flags, how much is a real secret vs noise/placeholder? + +> IMPORTANT: every "secret" here is SYNTHETIC — deterministically generated, +> correctly *shaped* and high-entropy, but not a real credential. Nothing in this +> file is live. It exists only to hold the detector's precision/recall steady as +> the code changes. + +Each sample: (id, text, expect) where `expect` is the secret_type that SHOULD be +detected, or None for a negative (nothing should be detected). +""" + +from __future__ import annotations + +import hashlib +from dataclasses import dataclass + +_ALNUM = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789" +_UPPER = "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789" + + +def _rand(n: int, seed: str, alphabet: str = _ALNUM) -> str: + """Deterministic, high-entropy string of length n over `alphabet`. + Deterministic so the corpus is stable across runs; high-entropy so valid + positives clear the entropy gate (>= 3.5 bits/char).""" + out: list[str] = [] + h = hashlib.sha256(seed.encode()).digest() + while len(out) < n: + h = hashlib.sha256(h).digest() + for b in h: + out.append(alphabet[b % len(alphabet)]) + if len(out) >= n: + break + return "".join(out) + + +@dataclass(frozen=True) +class Sample: + id: str + text: str + expect: str | None # secret_type that must be detected, or None for a clean negative + + @property + def is_positive(self) -> bool: + return self.expect is not None + + +# ── POSITIVES — a real (synthetic) secret is planted; the layer must catch it ── +_POSITIVES: list[Sample] = [ + Sample("aws-access-key", + f'const AWS_KEY = "AKIA{_rand(16, "aws", _UPPER)}";', + "AWS Access Key"), + Sample("github-pat", + f'GITHUB_TOKEN=ghp_{_rand(36, "ghpat")}', + "GitHub Personal Access Token"), + Sample("stripe-secret", + f'stripe.setApiKey("sk_live_{_rand(24, "stripe")}");', + "Stripe Secret Key"), + Sample("google-api-key", + f'apiKey: "AIza{_rand(35, "goog")}",', + "Google Cloud API Key"), + Sample("slack-webhook", + f'https://hooks.slack.com/services/T{_rand(9, "sa", _UPPER)}/B{_rand(9, "sb", _UPPER)}/{_rand(24, "sc")}', + "Slack Webhook"), + Sample("jwt", + f'Authorization: Bearer eyJ{_rand(20, "j1")}.eyJ{_rand(30, "j2")}.{_rand(43, "j3")}', + "JWT Token"), + Sample("gitlab-pat", + f'GITLAB_TOKEN=glpat-{_rand(20, "glp")}', + "GitLab Personal Access Token"), + Sample("openai", + # real OpenAI keys carry the literal "T3BlbkFJ" infix; the detector + # requires it (rejects look-alikes) — so the corpus must include it. + f'OPENAI_API_KEY=sk-{_rand(20, "oai1")}T3BlbkFJ{_rand(20, "oai2")}', + "OpenAI API Key"), + Sample("sendgrid", + f'SG.{_rand(22, "sg1")}.{_rand(43, "sg2")}', + "SendGrid API Key"), + Sample("npm-token", + f'//registry.npmjs.org/:_authToken=npm_{_rand(36, "npm")}', + "npm Access Token"), + Sample("database-uri", + f'DATABASE_URL=postgres://dbuser:{_rand(16, "dbp")}@db.internal:5432/prod', + "Database Connection URI"), + Sample("basic-auth-url", + f'fetch("https://admin:{_rand(14, "bap")}@api.internal.example.com/v1/")', + "Basic-Auth URL Credentials"), +] + +# ── NEGATIVES — placeholders / examples / noise; the layer must stay silent ── +_NEGATIVES: list[Sample] = [ + Sample("aws-doc-example", 'key = "AKIAIOSFODNN7EXAMPLE"', None), + Sample("your-api-key", 'apiKey: "YOUR_API_KEY_HERE"', None), + Sample("stripe-placeholder", 'sk_live_your_secret_key_here', None), + Sample("github-xxxx", 'token = "ghp_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"', None), + Sample("google-zeros", 'apiKey: "AIzaSy00000000000000000000000000000000000"', None), + Sample("env-ref", 'const key = process.env.STRIPE_SECRET_KEY;', None), + Sample("angle-placeholder", 'Authorization: Bearer ', None), + Sample("changeme", 'password = "changeme_changeme_changeme"', None), + Sample("redacted", 'secret = "redacted_for_security_reasons"', None), + Sample("git-sha", "const commit = 'a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6e7f8a9b0';", None), + Sample("uuid", "id: '550e8400-e29b-41d4-a716-446655440000'", None), + Sample("low-entropy-aws", 'k = "AKIAAAAAAAAAAAAAAAAA"', None), + Sample("plain-prose", "This build ships no credentials; configure them via environment variables.", None), + Sample("semver-list", 'deps: ["1.2.3", "4.5.6", "7.8.9", "10.11.12"]', None), + Sample("css-color-hash", ".btn { color: #a3f2c1; background: #1b2e4d; border: #ff00aa; }", None), +] + +CORPUS: list[Sample] = _POSITIVES + _NEGATIVES diff --git a/backend/bench/run_bench.py b/backend/bench/run_bench.py new file mode 100644 index 0000000..ee484b4 --- /dev/null +++ b/backend/bench/run_bench.py @@ -0,0 +1,100 @@ +""" +SecretNode — bench/run_bench.py (R2: precision/recall harness) + +Runs the deterministic detection layer (extract_secrets: regex + placeholder +allowlist + entropy gate + base64 pass — no network, no AI) over the labelled +corpus and reports precision / recall / F1, plus every miss so a regression is +visible line-by-line. + +Usage: + python -m bench.run_bench # from backend/ (human-readable report) + make bench # from repo root + +Programmatic: + from bench.run_bench import evaluate + m = evaluate() # -> {"precision":…, "recall":…, "f1":…, "tp":…, …, "misses":[…]} + +Definitions (sample-level; each positive plants exactly one secret, each +negative plants none): + TP positive sample that produced >=1 finding (planted secret caught) + FN positive sample that produced 0 findings (missed a real secret) + FP negative sample that produced >=1 finding (flagged noise/placeholder) + TN negative sample that produced 0 findings + precision = TP / (TP + FP) recall = TP / (TP + FN) +""" + +from __future__ import annotations + +from bench.corpus import CORPUS, Sample +from scanner import extract_secrets + + +def _detect(sample: Sample) -> list[str]: + findings = extract_secrets("bench", "https://bench.local", "https://bench.local/x.js", sample.text) + return [f.secret_type for f in findings] + + +def evaluate() -> dict: + tp = fn = fp = tn = 0 + type_hits = 0 # positives that detected the EXACT expected type + misses: list[dict] = [] + + for s in CORPUS: + detected = _detect(s) + got = len(detected) > 0 + if s.is_positive: + if got: + tp += 1 + if s.expect in detected: + type_hits += 1 + else: + misses.append({"id": s.id, "kind": "WRONG_TYPE", + "expected": s.expect, "detected": detected}) + else: + fn += 1 + misses.append({"id": s.id, "kind": "FALSE_NEGATIVE", + "expected": s.expect, "detected": []}) + else: + if got: + fp += 1 + misses.append({"id": s.id, "kind": "FALSE_POSITIVE", + "expected": None, "detected": detected}) + else: + tn += 1 + + precision = tp / (tp + fp) if (tp + fp) else 1.0 + recall = tp / (tp + fn) if (tp + fn) else 1.0 + f1 = (2 * precision * recall / (precision + recall)) if (precision + recall) else 0.0 + positives = tp + fn + return { + "precision": round(precision, 4), + "recall": round(recall, 4), + "f1": round(f1, 4), + "type_accuracy": round(type_hits / positives, 4) if positives else 1.0, + "tp": tp, "fn": fn, "fp": fp, "tn": tn, + "positives": positives, "negatives": fp + tn, + "misses": misses, + } + + +def format_report(m: dict) -> str: + lines = [ + "SecretNode — detection-layer precision/recall (R2)", + "=" * 52, + f" corpus: {m['positives']} positives · {m['negatives']} negatives", + f" precision: {m['precision']:.3f} (TP={m['tp']} FP={m['fp']})", + f" recall: {m['recall']:.3f} (TP={m['tp']} FN={m['fn']})", + f" F1: {m['f1']:.3f}", + f" type accuracy: {m['type_accuracy']:.3f} (right type on caught positives)", + ] + if m["misses"]: + lines.append(" misses:") + for x in m["misses"]: + lines.append(f" - [{x['kind']}] {x['id']}: expected={x['expected']} detected={x['detected']}") + else: + lines.append(" misses: none — clean sweep") + return "\n".join(lines) + + +if __name__ == "__main__": # pragma: no cover + print(format_report(evaluate())) diff --git a/backend/posture.py b/backend/posture.py new file mode 100644 index 0000000..d36043c --- /dev/null +++ b/backend/posture.py @@ -0,0 +1,118 @@ +""" +SecretNode — posture.py (R8: passive attack-surface breadth) + +A first step from "secret scanner" toward the fuller attack-surface scanner the +brand promises: analyse the target's own HTTP response for missing/weak security +headers and simple misconfigurations. This is **pure passive analysis of a +response the target already serves** — no exploitation, no third-party calls, no +writes. Each issue is a `PostureFinding` with severity / CWE / remediation, so it +flows into the same reports as credential findings but stays in its own category +(it never enters the secret-validation or live-verification pipeline). + +Follow-ups (deferred, higher-risk / external network): CT-log subdomain +discovery (crt.sh), DNS resolution + dangling-CNAME takeover checks. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Any + +logger = logging.getLogger("secretnode.posture") + + +@dataclass +class PostureFinding: + name: str + severity: str # CRITICAL / HIGH / MEDIUM / LOW / INFO + cwe: str + evidence: str # what was observed on the response + remediation: str + category: str = "Security Posture" + found_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat()) + + def to_dict(self) -> dict[str, Any]: + return { + "name": self.name, "severity": self.severity, "cwe": self.cwe, + "evidence": self.evidence, "remediation": self.remediation, + "category": self.category, "found_at": self.found_at, + } + + +def analyze_security_headers(headers: dict[str, Any] | None, final_url: str) -> list[PostureFinding]: + """Inspect response headers for missing/weak security controls. Pure and + deterministic — unit-tested without any network.""" + h = {str(k).lower(): str(v) for k, v in (headers or {}).items()} + is_https = final_url.lower().startswith("https://") + out: list[PostureFinding] = [] + + def add(name: str, sev: str, cwe: str, evidence: str, remediation: str) -> None: + out.append(PostureFinding(name, sev, cwe, evidence, remediation)) + + if is_https and "strict-transport-security" not in h: + add("Missing HSTS", "MEDIUM", "CWE-319", + "No Strict-Transport-Security header on an HTTPS response.", + "Add 'Strict-Transport-Security: max-age=31536000; includeSubDomains'.") + + csp = h.get("content-security-policy", "") + if not csp: + add("Missing Content-Security-Policy", "MEDIUM", "CWE-693", + "No Content-Security-Policy header — weaker XSS / data-injection defence.", + "Define a restrictive Content-Security-Policy.") + + xcto = h.get("x-content-type-options", "") + if xcto.lower() != "nosniff": + add("Missing X-Content-Type-Options: nosniff", "LOW", "CWE-693", + f"X-Content-Type-Options is '{xcto or 'absent'}'.", + "Set 'X-Content-Type-Options: nosniff'.") + + if not h.get("x-frame-options", "") and "frame-ancestors" not in csp.lower(): + add("No clickjacking protection", "MEDIUM", "CWE-1021", + "Neither X-Frame-Options nor a CSP 'frame-ancestors' directive is present.", + "Set 'X-Frame-Options: DENY' or a CSP 'frame-ancestors' directive.") + + if "referrer-policy" not in h: + add("Missing Referrer-Policy", "LOW", "CWE-200", + "No Referrer-Policy header — the referrer URL may leak to third parties.", + "Set e.g. 'Referrer-Policy: strict-origin-when-cross-origin'.") + + if "permissions-policy" not in h: + add("Missing Permissions-Policy", "LOW", "CWE-693", + "No Permissions-Policy header — powerful browser features are not restricted.", + "Set a Permissions-Policy that disables features the site does not use.") + + for hdr in ("server", "x-powered-by", "x-aspnet-version"): + val = h.get(hdr, "") + if val and any(c.isdigit() for c in val): + add(f"Version disclosure via {hdr}", "LOW", "CWE-200", + f"{hdr}: {val}", + f"Remove or obscure the '{hdr}' header so it does not reveal software versions.") + + cookie = h.get("set-cookie", "") + if cookie: + low = cookie.lower() + if is_https and "secure" not in low: + add("Cookie without Secure flag", "MEDIUM", "CWE-614", + "A Set-Cookie on an HTTPS response lacks the 'Secure' attribute.", + "Add 'Secure' to cookies so they are never sent over plain HTTP.") + if "httponly" not in low: + add("Cookie without HttpOnly flag", "LOW", "CWE-1004", + "A Set-Cookie lacks the 'HttpOnly' attribute.", + "Add 'HttpOnly' to cookies that JavaScript does not need to read.") + + return out + + +async def fetch_posture(client: Any, target_url: str) -> list[PostureFinding]: + """One passive GET to the target root; analyse its response headers. Fully + defensive — any error yields no findings rather than failing the scan.""" + try: + r = await client.get(target_url) + headers = dict(getattr(r, "headers", {}) or {}) + final_url = str(getattr(r, "url", "") or target_url) + return analyze_security_headers(headers, final_url) + except Exception as exc: # noqa: BLE001 — best-effort, never break a scan + logger.warning("posture check failed for %s: %s", target_url, exc) + return [] diff --git a/backend/report.py b/backend/report.py index 2840d40..c978b0c 100644 --- a/backend/report.py +++ b/backend/report.py @@ -141,11 +141,15 @@ def finding_row(f: dict[str, Any]) -> str: badge = 'NEW' if f.get("is_new", True) else 'RECURRING' cwe = html.escape(str(f.get("cwe", ""))) impact = html.escape(f.get('impact', '') or '—') + vdetail = html.escape(f.get('verified_detail', '') or '') + # For a VERIFIED-active key, show WHO it belongs to / what it can reach — the + # concrete blast radius an attacker inherits (R1 verification enrichment). + vdetail_html = f'
🔓 live access: {vdetail}
' if vdetail else '' return f""" {sev_badge(_severity_of(f))} {html.escape(f.get('secret_type',''))}
{cwe}
- {impact} + {impact}{vdetail_html} {html.escape(f.get('source_url', f.get('target_url','')))} {f.get('confidence',0)}% {badge}{ver_badge(f)} @@ -182,6 +186,46 @@ def review_row(f: dict[str, Any]) -> str: if not remediation_blocks: remediation_blocks = '
No remediation items — scan is clean.
' + # R9 — verification-evidence callout: the single strongest client-facing signal. + # For credentials confirmed CURRENTLY ACTIVE (read-only check against the provider), + # list exactly what an attacker reaches. Only rendered when there is live proof. + if verified_active: + ve_rows = "" + for f in confirmed: + if str(f.get("verified", "")).lower() != "verified": + continue + detail = html.escape(f.get("verified_detail", "") or "live access confirmed") + loc = html.escape(f.get("source_url", f.get("target_url", ""))) + ve_rows += ( + f'
  • {html.escape(f.get("secret_type",""))} ' + f'{loc}' + f'
    🔓 confirmed live access: {detail}
    • ' + ) + verified_evidence = ( + '
    ' + f'
    ⚠ {verified_active} credential(s) confirmed CURRENTLY ACTIVE' + ' via a read-only check against the issuing provider
    ' + f'' + '
    These are not look-alikes — each was live at scan time. ' + 'Rotate/revoke them at the provider immediately.
    ' + '
    ' + ) + else: + verified_evidence = "" + + # R8 — passive security-posture findings (missing/weak headers, misconfig). + posture = scan.get("posture_findings", []) + + def _posture_row(p: dict[str, Any]) -> str: + return ( + f'{sev_badge(_severity_of(p))}' + f'{html.escape(p.get("name",""))}
    {html.escape(str(p.get("cwe","")))}
    ' + f'{html.escape(p.get("evidence",""))}' + f'{html.escape(p.get("remediation",""))}' + ) + posture_html = "\n".join(_posture_row(p) for p in sorted(posture, key=_sort_key)) or \ + 'No header/misconfiguration issues detected.' + return f""" @@ -228,6 +272,14 @@ def review_row(f: dict[str, Any]) -> str: .verdict-sub {{ font-size: 13px; color: #4a5568; margin-top: 8px; }} .scope p {{ font-size: 13px; color: #2d3748; }} td.impact {{ font-size: 12px; color: #742a2a; font-weight: 600; max-width: 260px; }} + .ver-evidence {{ background: #fff5f5; border: 1px solid #feb2b2; border-left: 6px solid #c53030; border-radius: 6px; padding: 14px 16px; margin: 16px 0; }} + .ver-evidence .ve-head {{ font-weight: 700; color: #c53030; font-size: 14px; }} + .ve-list {{ margin: 10px 0 6px; padding-left: 18px; font-size: 13px; }} + .ve-list li {{ margin: 6px 0; }} + .ver-ev-detail {{ font-family: 'Courier New', monospace; font-size: 12px; color: #742a2a; margin-top: 2px; }} + .stat.alert {{ border-color: #feb2b2; background: #fff5f5; }} + .stat.alert .num {{ color: #c53030; }} + .scope-quality {{ background: #f0fff4; border-left: 3px solid #276749; padding: 10px 12px; border-radius: 0 4px 4px 0; margin-top: 10px; }} @@ -243,6 +295,8 @@ def review_row(f: dict[str, Any]) -> str:
    {html.escape(verdict_sub)}
    + {verified_evidence} +
    Target {target}
    Prepared by {html.escape(agency_name)}
    @@ -258,9 +312,11 @@ def review_row(f: dict[str, Any]) -> str:
    {sev_counts['CRITICAL']}
    Critical
    {sev_counts['HIGH']}
    High
    {sev_counts['MEDIUM']}
    Medium
    +
    {verified_active}
    Verified Active
    {new_count}
    New
    {recurring_count}
    Recurring
    {len(needs_review)}
    Needs Review
    +
    {len(posture)}
    Posture Issues

    Scope & Methodology

    @@ -271,6 +327,11 @@ def review_row(f: dict[str, Any]) -> str: Shannon-entropy analysis. High-entropy candidates were then contextually validated to distinguish live secrets from mocks, placeholders and minified-code artefacts. No exploitation, authentication, data exfiltration, or write operations were performed against the target — testing is passive and authorized-scope only.

    +

    Detection quality. SecretNode is verification-first: where a credential type + supports it, a read-only check against the issuing provider confirms whether the key is currently active before + it is reported — so a "verified" finding is proven, not shape-matched. The deterministic detection layer is + continuously measured against a labelled benchmark corpus with a precision/recall gate in CI, so this report + favours confirmed impact over look-alikes and keeps false positives low.

    Confirmed Findings

    @@ -282,6 +343,12 @@ def review_row(f: dict[str, Any]) -> str:

    Remediation Guidance

    {remediation_blocks} +

    Security Posture & Misconfigurations

    + + + {posture_html} +
    SeverityIssue / CWEEvidenceRemediation
    +

    Flagged for Manual Review (AI validation unavailable)

    @@ -306,21 +373,23 @@ def generate_csv_report(scan: dict[str, Any]) -> str: writer = csv.writer(buf) writer.writerow([ "status", "severity", "cwe", "secret_type", "source_url", "confidence", - "is_new", "verified", "impact", "matched_value_partial", "reason", "found_at", + "is_new", "verified", "verified_detail", "impact", "matched_value_partial", + "reason", "found_at", ]) for f in sorted(scan.get("confirmed_findings", []), key=_sort_key): writer.writerow([ "CONFIRMED", _severity_of(f), f.get("cwe", ""), f.get("secret_type", ""), f.get("source_url", f.get("target_url", "")), f.get("confidence", 0), "NEW" if f.get("is_new", True) else "RECURRING", - f.get("verified", "disabled"), f.get("impact", ""), f.get("raw_match", ""), - f.get("reason", ""), f.get("found_at", ""), + f.get("verified", "disabled"), f.get("verified_detail", ""), f.get("impact", ""), + f.get("raw_match", ""), f.get("reason", ""), f.get("found_at", ""), ]) for f in scan.get("needs_review_findings", []): writer.writerow([ "NEEDS_REVIEW", _severity_of(f), f.get("cwe", ""), f.get("secret_type", ""), f.get("source_url", f.get("target_url", "")), - "", "", "", f.get("impact", ""), f.get("raw_match", ""), f.get("reason", ""), f.get("found_at", ""), + "", "", "", "", f.get("impact", ""), f.get("raw_match", ""), + f.get("reason", ""), f.get("found_at", ""), ]) return buf.getvalue() @@ -329,6 +398,49 @@ def generate_csv_report(scan: dict[str, Any]) -> str: # SARIF 2.1.0 # ───────────────────────────────────────────────────────────────────────────── +def _rule_id(secret_type: str) -> str: + """Deterministic SARIF ruleId for a secret type. Shared by the rule catalog + and each result so results always resolve to a described rule.""" + return "secretnode/" + secret_type.lower().replace(" ", "-").replace("/", "-") + + +def _catalog_rules() -> dict[str, dict[str, Any]]: + """Full catalog of every detector as a SARIF reportingDescriptor. + + A SARIF driver should advertise every rule it *can* apply — not only the ones + that happened to fire on this scan — so consumers (GitHub code scanning, CI + dashboards) always have the rule's help text, CWE, and default severity. Built + from the live pattern registry so it never drifts from what the scanner detects. + """ + try: + from scanner import SECRET_PATTERNS + except Exception: # pragma: no cover - packaged/alternate import path + try: + from backend.scanner import SECRET_PATTERNS # type: ignore + except Exception: + return {} + catalog: dict[str, dict[str, Any]] = {} + for p in SECRET_PATTERNS: + rid = _rule_id(p.name) + sev = str(getattr(p, "severity", "MEDIUM")).upper() + cwe = str(getattr(p, "cwe", "CWE-798")) + catalog[rid] = { + "id": rid, + "name": p.name.replace(" ", ""), + "shortDescription": {"text": f"Exposed {p.name}"}, + "fullDescription": {"text": str(getattr(p, "description", "") or f"Exposed {p.name}")}, + "help": {"text": str(getattr(p, "remediation", "") or "Rotate the exposed credential and remove it from client-side code.")}, + "helpUri": _TOOL_URI, + "defaultConfiguration": {"level": _SARIF_LEVEL.get(sev, "warning")}, + "properties": { + "tags": ["security", "secret", cwe], + "cwe": cwe, + "security-severity": _SARIF_SECURITY_SEVERITY.get(sev, "5.0"), + }, + } + return catalog + + def generate_sarif_report(scan: dict[str, Any]) -> str: """Emit findings as SARIF 2.1.0 — uploadable to GitHub code scanning or any SARIF-aware pipeline. Confirmed and needs-review findings are both included; @@ -340,13 +452,14 @@ def generate_sarif_report(scan: dict[str, Any]) -> str: (f, True) for f in scan.get("needs_review_findings", []) ] - # Build a rule per distinct secret type actually present. - rules: dict[str, dict[str, Any]] = {} + # Advertise the full detector catalog; add finding-specific rules for any + # unknown type below so every result still resolves to a described rule. + rules: dict[str, dict[str, Any]] = _catalog_rules() results: list[dict[str, Any]] = [] for f, is_review in all_findings: secret_type = f.get("secret_type", "Unknown") - rule_id = "secretnode/" + secret_type.lower().replace(" ", "-").replace("/", "-") + rule_id = _rule_id(secret_type) severity = _severity_of(f) cwe = str(f.get("cwe", "CWE-798")) @@ -368,7 +481,13 @@ def generate_sarif_report(scan: dict[str, Any]) -> str: level = "note" if is_review else _SARIF_LEVEL.get(severity, "warning") location_uri = f.get("source_url") or f.get("target_url") or "unknown" verified = str(f.get("verified", "disabled")) - vprefix = "[VERIFIED ACTIVE] " if verified == "verified" else "" + verified_detail = str(f.get("verified_detail", "") or "") + # Keep the literal "[VERIFIED ACTIVE]" token intact for downstream matchers, + # then append the identity/scope (blast radius) when we captured one. + vprefix = ( + (f"[VERIFIED ACTIVE] ({verified_detail}) " if verified_detail else "[VERIFIED ACTIVE] ") + if verified == "verified" else "" + ) impact = str(f.get("impact", "") or "") msg = ( vprefix + f"{secret_type} ({severity}) detected. " @@ -391,6 +510,7 @@ def generate_sarif_report(scan: dict[str, Any]) -> str: "cwe": cwe, "confidence": f.get("confidence", 0), "verified": verified, + "verified_detail": verified_detail, "impact": impact, "status": "needs_review" if is_review else "confirmed", }, diff --git a/backend/scanner.py b/backend/scanner.py index 8f73db0..1d98b4c 100644 --- a/backend/scanner.py +++ b/backend/scanner.py @@ -26,6 +26,7 @@ from google.genai import errors as genai_errors, types from pydantic import BaseModel, Field, ValidationError +import posture import verifier logger = logging.getLogger("secretnode.scanner") @@ -95,6 +96,9 @@ def _env_float(name: str, default: float) -> float: MAX_RAW_FINDINGS_PER_SCAN = _env_int("MAX_RAW_FINDINGS_PER_SCAN", 500) # safety cap: stop a runaway scan # (e.g. a minified bundle full of high-entropy noise) from # generating unbounded Gemini calls / RAM use on the Pi +MAX_MATCHES_PER_PATTERN = _env_int("MAX_MATCHES_PER_PATTERN", 100) # R3 defence-in-depth: bound the + # matches examined for ANY single pattern on ANY single text, so a + # crafted blob cannot spawn millions of matches for one detector. # ── Type alias for the broadcaster callback ──────────────────────────────────── Broadcaster = Callable[[dict[str, Any]], Coroutine[Any, Any, None]] @@ -156,6 +160,7 @@ class ValidatedFinding: ) is_new: bool = True # set False by run_scan if this fingerprint was seen in a prior scan verified: str = "disabled" # live-verification status: verified/unverified/unsupported/disabled + verified_detail: str = "" # identity/scope of a VERIFIED credential (R1) — never the secret itself impact: str = "" # AI blast-radius statement: what an attacker could actually do public_by_design: bool = False # True for identifiers meant to be public (Firebase web key, pk_ …) @@ -187,6 +192,7 @@ def to_dict(self) -> dict[str, Any]: "validated_at": self.validated_at, "is_new": self.is_new, "verified": self.verified, + "verified_detail": self.verified_detail, "severity": self.effective_severity(), "cwe": self._meta().cwe, "remediation": self._meta().remediation, @@ -875,6 +881,16 @@ async def fetch_url( # that are stripped from the shipped bundle — a well-established ASM technique. FOLLOW_SOURCE_MAPS = os.environ.get("FOLLOW_SOURCE_MAPS", "true").lower() == "true" MAX_SOURCE_MAPS = _env_int("MAX_SOURCE_MAPS", 40) +# R5 surface expansion: a source map embeds the ORIGINAL, un-minified source in +# its `sourcesContent` array (JSON-escaped). The raw .map is scanned as text, but +# secrets in the original source are frequently escaped/split there and missed — +# so we also decode each embedded source and scan it as real code. +SCAN_SOURCEMAP_CONTENT = os.environ.get("SCAN_SOURCEMAP_CONTENT", "true").lower() == "true" +MAX_SOURCEMAP_SOURCES = _env_int("MAX_SOURCEMAP_SOURCES", 200) +# R8: passive HTTP security-posture check (missing/weak security headers, +# version disclosure, insecure cookies) on the target root. Pure analysis of the +# response the target already serves — no exploitation, no third-party calls. +SCAN_HTTP_POSTURE = os.environ.get("SCAN_HTTP_POSTURE", "true").lower() == "true" def _same_scope(base_host: str, candidate_host: str) -> bool: @@ -944,6 +960,42 @@ def extract_source_map_urls(js_body: str, js_url: str) -> list[str]: return result +def looks_like_sourcemap(url: str, body: str) -> bool: + """Heuristic: is this asset a JS source map? By extension, or by the tell-tale + `sourcesContent` / (`version` + `mappings`) keys near the top of the body.""" + if url.split("?", 1)[0].split("#", 1)[0].endswith(".map"): + return True + head = body[:4000] + return '"sourcesContent"' in head or ('"mappings"' in head and '"version"' in head) + + +def extract_sourcemap_sources(map_body: str, map_url: str) -> list[tuple[str, str]]: + """R5 surface expansion: decode a source map's `sourcesContent` (the original, + un-minified source, JSON-escaped inside the .map) into scannable code, paired + with its `sources` name. Secrets stripped from the shipped bundle — or escaped + within the raw .map JSON so the regex pass misses them — surface here. + Bounded (MAX_SOURCEMAP_SOURCES) and fully defensive: any parse error → [].""" + try: + doc = json.loads(map_body) + except Exception: + return [] + if not isinstance(doc, dict): + return [] + contents = doc.get("sourcesContent") + if not isinstance(contents, list): + return [] + names = doc.get("sources") if isinstance(doc.get("sources"), list) else [] + out: list[tuple[str, str]] = [] + for i, content in enumerate(contents): + if len(out) >= MAX_SOURCEMAP_SOURCES: + break + if not isinstance(content, str) or not content: + continue + name = names[i] if i < len(names) and isinstance(names[i], str) else f"source[{i}]" + out.append((f"{map_url} → {name}", content)) + return out + + _ANCHOR_HREF_RE = re.compile(r']+href=["\']([^"\']+)["\']', re.IGNORECASE) _NON_PAGE_EXT = ( ".png", ".jpg", ".jpeg", ".gif", ".svg", ".webp", ".ico", ".css", @@ -1180,7 +1232,17 @@ def _scan_text( ) -> list[RawFinding]: findings: list[RawFinding] = [] for pattern in SECRET_PATTERNS: + examined = 0 for match in pattern.regex.finditer(text): + examined += 1 + if examined > MAX_MATCHES_PER_PATTERN: + # Defence-in-depth: a pathological blob shall not spawn unbounded + # matches for one detector. Bound the work and move on. + logger.debug( + "Match cap (%d) reached for %s; truncating further matches.", + MAX_MATCHES_PER_PATTERN, pattern.name, + ) + break raw_value = ( match.group(1) if match.lastindex and match.lastindex >= 1 @@ -1712,6 +1774,7 @@ async def emit(event: dict[str, Any]) -> None: "verified_count": 0, "unverified_count": 0, "filtered_unverified_count": 0, + "posture_findings": [], "errors": [], "duration_seconds": 0.0, } @@ -1740,11 +1803,44 @@ async def emit(event: dict[str, Any]) -> None: "message": f"Asset collection complete — {len(assets)} files to scan", }) + # ── 1b. Passive security-posture check (R8) ───────────────────────── + # Analyse the target root's own response headers for missing/weak + # security controls. Best-effort: never blocks or fails the scan. + if SCAN_HTTP_POSTURE: + state.check() + pfindings = await posture.fetch_posture(client, target_url) + result["posture_findings"] = [p.to_dict() for p in pfindings] + if pfindings: + await emit({ + "type": "log", "level": "INFO", + "message": f"Security posture: {len(pfindings)} header/misconfiguration issue(s) found", + }) + # ── 2. Regex Extraction ──────────────────────────────────────────── state.check() all_raw: list[RawFinding] = [] for source_url, body in assets: state.check() + + # ── R5: for a source map, scan its DECODED original source instead of + # the raw .map JSON. Better per-file attribution, catches secrets that + # are escaped/structured in the raw JSON, and avoids the map's own + # high-entropy "mappings" VLQ blob (a false-positive source). Falls + # back to scanning the raw body if there's no usable sourcesContent. + if SCAN_SOURCEMAP_CONTENT and looks_like_sourcemap(source_url, body): + srcs = extract_sourcemap_sources(body, source_url) + if srcs: + for vsrc_url, content in srcs: + state.check() + sfound = extract_secrets(scan_id, target_url, vsrc_url, content) + if sfound: + await emit({ + "type": "log", "level": "WARN", + "message": f"Found {len(sfound)} match(es) in source-map original {vsrc_url}", + }) + all_raw.extend(sfound) + continue + found = extract_secrets(scan_id, target_url, source_url, body) if found: await emit({ @@ -1874,9 +1970,11 @@ async def _validate_one(f: RawFinding) -> ValidatedFinding: }) for vf in confirmed: state.check() - vf.verified = await verifier.verify_finding( + _vres = await verifier.verify_finding_detailed( vf.raw.secret_type, vf.raw.raw_match, client ) + vf.verified = _vres.status + vf.verified_detail = _vres.detail result["verified_count"] = sum(1 for v in confirmed if v.verified == "verified") result["unverified_count"] = sum(1 for v in confirmed if v.verified == "unverified") await emit({ diff --git a/backend/tests/test_bench.py b/backend/tests/test_bench.py new file mode 100644 index 0000000..2a71787 --- /dev/null +++ b/backend/tests/test_bench.py @@ -0,0 +1,44 @@ +""" +R2 — regression gate for detection-layer precision/recall. + +Runs the labelled corpus (bench/) through extract_secrets() and fails the build +if precision or recall drops below threshold, so a change that starts missing +real secrets (FN) or flagging placeholders/noise (FP) is caught in CI — not in a +client's report. +""" + +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +os.environ.setdefault("SECRETNODE_API_KEY", "test-key-for-pytest") + +from bench.run_bench import evaluate + + +def test_precision_recall_gate(): + m = evaluate() + assert m["precision"] >= 0.95, f"precision regressed: {m['precision']} · {m['misses']}" + assert m["recall"] >= 0.95, f"recall regressed: {m['recall']} · {m['misses']}" + assert m["f1"] >= 0.95, f"F1 regressed: {m['f1']}" + + +def test_zero_false_positives_on_placeholders_and_noise(): + """Cindrasec's brand is 'no false positives'. Lock it in: placeholders, + documentation examples, and high-entropy non-secrets must never be flagged.""" + m = evaluate() + fps = [x for x in m["misses"] if x["kind"] == "FALSE_POSITIVE"] + assert not fps, f"false positives introduced: {fps}" + + +def test_type_accuracy_high(): + m = evaluate() + assert m["type_accuracy"] >= 0.9, f"type accuracy regressed: {m['type_accuracy']} · {m['misses']}" + + +def test_corpus_has_both_classes(): + # Guard the corpus itself: a benchmark with no negatives (or no positives) is + # meaningless. Ensure both classes are present and non-trivial. + m = evaluate() + assert m["positives"] >= 10 + assert m["negatives"] >= 10 diff --git a/backend/tests/test_posture.py b/backend/tests/test_posture.py new file mode 100644 index 0000000..dfce772 --- /dev/null +++ b/backend/tests/test_posture.py @@ -0,0 +1,94 @@ +""" +R8 — passive HTTP security-posture checks (headers / misconfig). +Pure analysis; a lightweight mock stands in for httpx for the fetch path. +""" + +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +os.environ.setdefault("SECRETNODE_API_KEY", "test-key-for-pytest") + +import pytest + +import posture + + +def _names(findings): + return {f.name for f in findings} + + +def test_bare_https_response_flags_the_core_headers(): + out = posture.analyze_security_headers({}, "https://x.com") + n = _names(out) + assert "Missing HSTS" in n + assert "Missing Content-Security-Policy" in n + assert "Missing X-Content-Type-Options: nosniff" in n + assert "No clickjacking protection" in n + assert "Missing Referrer-Policy" in n + assert "Missing Permissions-Policy" in n + + +def test_hardened_headers_produce_no_findings(): + hardened = { + "Strict-Transport-Security": "max-age=63072000; includeSubDomains; preload", + "Content-Security-Policy": "default-src 'self'; frame-ancestors 'none'", + "X-Content-Type-Options": "nosniff", + "X-Frame-Options": "DENY", + "Referrer-Policy": "strict-origin-when-cross-origin", + "Permissions-Policy": "geolocation=()", + } + assert posture.analyze_security_headers(hardened, "https://x.com") == [] + + +def test_hsts_not_flagged_on_plain_http(): + out = posture.analyze_security_headers({}, "http://x.com") + assert "Missing HSTS" not in _names(out) + + +def test_csp_frame_ancestors_satisfies_clickjacking(): + out = posture.analyze_security_headers( + {"Content-Security-Policy": "frame-ancestors 'none'"}, "https://x.com") + assert "No clickjacking protection" not in _names(out) + + +def test_version_disclosure_flagged(): + out = posture.analyze_security_headers({"Server": "nginx/1.18.0"}, "https://x.com") + hit = [f for f in out if f.name.startswith("Version disclosure")] + assert hit and hit[0].cwe == "CWE-200" + + +def test_insecure_cookie_flags(): + out = posture.analyze_security_headers({"Set-Cookie": "sid=abc"}, "https://x.com") + n = _names(out) + assert "Cookie without Secure flag" in n + assert "Cookie without HttpOnly flag" in n + secure_ok = posture.analyze_security_headers( + {"Set-Cookie": "sid=abc; Secure; HttpOnly"}, "https://x.com") + assert "Cookie without Secure flag" not in _names(secure_ok) + + +def test_findings_serialize(): + out = posture.analyze_security_headers({}, "https://x.com") + d = out[0].to_dict() + assert {"name", "severity", "cwe", "evidence", "remediation", "category"} <= set(d) + + +@pytest.mark.asyncio +async def test_fetch_posture_reads_headers(): + class _Resp: + headers = {"Server": "Apache/2.4.7"} + url = "https://x.com/" + class _Client: + async def get(self, url, **kw): + return _Resp() + out = await posture.fetch_posture(_Client(), "https://x.com") + assert any(f.name.startswith("Version disclosure") for f in out) + + +@pytest.mark.asyncio +async def test_fetch_posture_fails_closed(): + class _Boom: + async def get(self, *a, **k): + raise RuntimeError("network down") + assert await posture.fetch_posture(_Boom(), "https://x.com") == [] diff --git a/backend/tests/test_regex_safety.py b/backend/tests/test_regex_safety.py new file mode 100644 index 0000000..25f2f9f --- /dev/null +++ b/backend/tests/test_regex_safety.py @@ -0,0 +1,88 @@ +""" +R3 — regex safety / ReDoS audit (automated + permanent). + +Secret detectors run untrusted, attacker-influenceable input (a target's own +minified JS) through 54 regexes. A single catastrophic-backtracking pattern +would let a crafted asset hang a scan (denial of service). This suite proves, +and keeps proving as patterns are added, that: + + 1. no detector exhibits catastrophic backtracking on adversarial input + (empirical wall-clock bound — the real guarantee), and + 2. no detector source contains the classic nested-quantifier ReDoS construct + (static guard that gates future contributions), and + 3. the per-pattern match cap bounds work on a match-flood input. +""" + +import os +import re +import sys +import time + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +os.environ.setdefault("SECRETNODE_API_KEY", "test-key-for-pytest") + +import scanner + +# Per-pattern per-input wall-clock ceiling. A linear regex clears 50 KB in well +# under a millisecond; a backtracking one blows past this by orders of magnitude. +_TIME_BUDGET_S = 0.75 + +# Adversarial inputs: long single-char runs, near-miss prefixes that start a +# match then force a fail, and structures that target the ".{0,N}"-gap and +# credential-in-URL patterns specifically. +_N = 50_000 +_ADVERSARIAL = [ + "A" * _N, "a" * _N, "0" * _N, "=" * _N, "/" * _N, " " * _N, "!" * _N, + "AKIA" + "A" * _N, + "sk_live_" + "x" * _N, + "ghp_" + "a" * _N, + "eyJ" + "A" * _N, + "aws" + " " * 19 + "secret" + " " * 19 + "'" + "A" * _N, # AWS-secret .{0,20} gap + "twilio" + " " * 19 + "'" + "a" * _N, # Twilio .{0,20} gap + "heroku" + " " * 29 + "'" + "a" * _N, # Heroku .{0,30} gap + "https://" + "a" * _N + ":" + "b" * _N + "@" + "c" * _N, # basic-auth URL creds + "-----BEGIN " + "A" * _N, + "https://hooks.slack.com/services/T" + "A" * _N, + ("Bearer " + "a" * 500 + " ") * 100, + ("A" * 80 + "\n") * 600, # base64-ish lines +] + +# Classic ReDoS shape: a group that contains an unbounded quantifier and is +# ITSELF unbounded-quantified, e.g. (a+)+, (.*)*, (\w+)*. Char classes like +# [A-Za-z0-9/+=]+ (quantifier on a class, not a group) are linear and excluded. +_NESTED_QUANT_RE = re.compile(r"\((?![?]:)[^)]*[+*][^)]*\)[+*]") + + +def test_no_detector_catastrophic_backtracking(): + slow = [] + for p in scanner.SECRET_PATTERNS: + for inp in _ADVERSARIAL: + t0 = time.perf_counter() + list(p.regex.finditer(inp)) + dt = time.perf_counter() - t0 + if dt > _TIME_BUDGET_S: + slow.append((p.name, round(dt, 3), len(inp))) + assert not slow, f"potential ReDoS — patterns exceeded {_TIME_BUDGET_S}s: {slow}" + + +def test_no_nested_quantifier_construct_in_sources(): + offenders = [p.name for p in scanner.SECRET_PATTERNS + if _NESTED_QUANT_RE.search(p.regex.pattern)] + assert not offenders, f"nested-quantifier ReDoS construct in: {offenders}" + + +def test_match_cap_bounds_a_flood(): + # A blob with far more distinct, high-entropy AWS keys than the cap must not + # yield unbounded findings for that one pattern. + import hashlib + alpha = "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789" + parts = [] + for i in range(400): + h = hashlib.sha256(f"k{i}".encode()).digest() + body = "".join(alpha[b % len(alpha)] for b in h[:16]) + parts.append(f'k="AKIA{body}"') + text = "\n".join(parts) + findings = scanner._scan_text("s", "t", "u", text) + aws = [f for f in findings if f.secret_type == "AWS Access Key"] + assert len(aws) <= scanner.MAX_MATCHES_PER_PATTERN + assert len(aws) < 400 # proves the cap actually engaged diff --git a/backend/tests/test_report.py b/backend/tests/test_report.py index de747a9..ea79287 100644 --- a/backend/tests/test_report.py +++ b/backend/tests/test_report.py @@ -36,7 +36,8 @@ def _scan() -> dict: "source_url": "https://example.com/config.js", "confidence": 97, "raw_match": "AKIA***", "reason": "live key", "is_new": False, "severity": "CRITICAL", "cwe": "CWE-798", "remediation": "revoke now", - "verified": "verified", "found_at": "now", + "verified": "verified", "verified_detail": "account acme-bot · scopes: repo", + "found_at": "now", }, ], "needs_review_findings": [ @@ -128,3 +129,96 @@ def test_sarif_carries_verified_property_and_prefix(): verified = [r for r in results if r["properties"].get("verified") == "verified"] assert verified assert verified[0]["message"]["text"].startswith("[VERIFIED ACTIVE]") + + +def test_sarif_advertises_full_detector_catalog(): + """The SARIF driver must describe every detector it can apply (industrial + best practice), not only rules that fired. Catalog is built from the live + registry, so it stays in sync with the scanner.""" + import scanner + doc = json.loads(report.generate_sarif_report(_scan())) + rules = doc["runs"][0]["tool"]["driver"]["rules"] + rule_ids = {r["id"] for r in rules} + # Every registry pattern is advertised as a rule, regardless of findings. + assert len(rules) >= len(scanner.SECRET_PATTERNS) + for p in scanner.SECRET_PATTERNS: + assert report._rule_id(p.name) in rule_ids + # Rules carry the metadata CI consumers rely on. + sample = next(r for r in rules if r["id"] == report._rule_id("AWS Access Key")) + assert sample["properties"]["cwe"].startswith("CWE-") + assert "security-severity" in sample["properties"] + assert sample["defaultConfiguration"]["level"] in ("error", "warning", "note") + + +def test_sarif_clean_scan_still_lists_rules(): + """A clean scan (zero findings) must still advertise the rule catalog so the + report is a complete, self-describing artifact.""" + clean = {"scan_id": "s0", "target_url": "https://example.com", + "confirmed_findings": [], "needs_review_findings": [], "assets_fetched": 2} + doc = json.loads(report.generate_sarif_report(clean)) + run = doc["runs"][0] + assert run["results"] == [] + assert len(run["tool"]["driver"]["rules"]) > 0 + + +def test_html_surfaces_verified_identity_detail(): + """A VERIFIED-active key must show WHO it belongs to / what it reaches (R1).""" + out = report.generate_html_report(_scan()) + assert "live access" in out + assert "account acme-bot" in out + + +def test_csv_has_verified_detail_column(): + out = report.generate_csv_report(_scan()) + header = out.splitlines()[0] + assert "verified_detail" in header + assert "account acme-bot" in out + + +def test_sarif_message_carries_identity_and_keeps_token(): + doc = json.loads(report.generate_sarif_report(_scan())) + v = [r for r in doc["runs"][0]["results"] if r["properties"].get("verified") == "verified"][0] + # literal token preserved for downstream matchers, identity appended + assert v["message"]["text"].startswith("[VERIFIED ACTIVE]") + assert "account acme-bot" in v["message"]["text"] + assert v["properties"]["verified_detail"] == "account acme-bot · scopes: repo" + + +def test_html_r9_verification_evidence_callout(): + """R9: verified-active findings get a prominent live-access evidence block.""" + out = report.generate_html_report(_scan()) + assert "CURRENTLY ACTIVE" in out + assert "confirmed live access" in out + assert "account acme-bot" in out # the R1 identity detail surfaces in the callout + + +def test_html_r9_precision_statement_present(): + out = report.generate_html_report(_scan()) + assert "Detection quality" in out + assert "precision/recall" in out + assert "verification-first" in out + + +def test_html_r9_verified_active_tile(): + out = report.generate_html_report(_scan()) + assert "Verified Active" in out + + +def test_html_r9_no_evidence_callout_on_clean_scan(): + clean = {"scan_id": "s0", "target_url": "https://x.com", + "confirmed_findings": [], "needs_review_findings": [], "assets_fetched": 3} + out = report.generate_html_report(clean) + assert "CURRENTLY ACTIVE via" not in out # no callout when nothing is verified-active + assert "Detection quality" in out # methodology still present + + +def test_html_r8_posture_section_and_tile(): + scan = _scan() + scan["posture_findings"] = [ + {"name": "Missing HSTS", "severity": "MEDIUM", "cwe": "CWE-319", + "evidence": "No Strict-Transport-Security header.", "remediation": "Add HSTS.", "category": "Security Posture"}, + ] + out = report.generate_html_report(scan) + assert "Security Posture & Misconfigurations" in out + assert "Missing HSTS" in out + assert "Posture Issues" in out diff --git a/backend/tests/test_sourcemap.py b/backend/tests/test_sourcemap.py new file mode 100644 index 0000000..41e7769 --- /dev/null +++ b/backend/tests/test_sourcemap.py @@ -0,0 +1,78 @@ +""" +R5 — source-map surface expansion. + +A JS source map embeds the ORIGINAL, un-minified source in `sourcesContent`. +The raw .map is scanned as text, but secrets in the original source are often +escaped/structured there (or absent from the shipped bundle entirely). These +tests cover decoding that embedded source into scannable, per-file-attributed +code — and the defensive handling around it. +""" + +import hashlib +import json +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +os.environ.setdefault("SECRETNODE_API_KEY", "test-key-for-pytest") + +import scanner + +_A = "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789" + + +def _awskey(seed: str = "s") -> str: + h = hashlib.sha256(seed.encode()).digest() + return "AKIA" + "".join(_A[b % len(_A)] for b in h[:16]) + + +def _map(sources, contents) -> str: + return json.dumps({"version": 3, "sources": sources, "sourcesContent": contents, "mappings": "AAAA"}) + + +def test_extract_sources_decodes_and_pairs_names(): + body = _map(["src/config.js", "src/app.js"], ["const A = 1;", "const B = 2;"]) + out = scanner.extract_sourcemap_sources(body, "https://x.com/app.js.map") + assert len(out) == 2 + assert out[0][0].endswith("src/config.js") and out[0][1] == "const A = 1;" + assert out[1][0].endswith("src/app.js") + + +def test_malformed_map_returns_empty(): + assert scanner.extract_sourcemap_sources("{not json", "u") == [] + assert scanner.extract_sourcemap_sources("[]", "u") == [] # not a dict + assert scanner.extract_sourcemap_sources('{"version":3}', "u") == [] # no sourcesContent + + +def test_null_and_nonstring_entries_are_skipped(): + body = _map(["a.js", "b.js", "c.js"], ["const A=1;", None, 123]) + out = scanner.extract_sourcemap_sources(body, "m") + assert len(out) == 1 and out[0][1] == "const A=1;" + + +def test_looks_like_sourcemap_detection(): + assert scanner.looks_like_sourcemap("https://x.com/app.js.map", "") + assert scanner.looks_like_sourcemap("https://x.com/app.js.map?v=2", "") + assert scanner.looks_like_sourcemap("https://x.com/bundle", '{"version":3,"mappings":"A","sources":[]}') + assert scanner.looks_like_sourcemap("u", '{"sourcesContent":["x"]}') + assert not scanner.looks_like_sourcemap("https://x.com/app.js", "console.log(1)") + + +def test_secret_in_original_source_is_found_and_attributed(): + """A secret living in the map's original source is decoded, scanned, and + reported against a precise per-file virtual source URL (R5's core value).""" + key = _awskey("r5") + body = _map(["src/secrets.js"], [f'export const AWS_KEY = "{key}";']) + for vsrc_url, content in scanner.extract_sourcemap_sources(body, "https://x.com/app.js.map"): + findings = scanner.extract_secrets("s", "https://x.com", vsrc_url, content) + aws = [f for f in findings if f.secret_type == "AWS Access Key"] + assert aws, "AWS key in sourcesContent was not detected" + assert aws[0].raw_match == key + assert "src/secrets.js" in aws[0].source_url # precise attribution + + +def test_respects_max_sources_cap(monkeypatch): + monkeypatch.setattr(scanner, "MAX_SOURCEMAP_SOURCES", 5) + body = _map([f"s{i}.js" for i in range(50)], [f"const X{i}=1;" for i in range(50)]) + out = scanner.extract_sourcemap_sources(body, "m") + assert len(out) == 5 diff --git a/backend/tests/test_verifier.py b/backend/tests/test_verifier.py index 2e08c17..c13409c 100644 --- a/backend/tests/test_verifier.py +++ b/backend/tests/test_verifier.py @@ -91,3 +91,83 @@ def test_every_verifier_maps_to_a_real_pattern(): names = {p.name for p in scanner.SECRET_PATTERNS} for secret_type in verifier.VERIFIERS: assert secret_type in names, f"{secret_type} not in SECRET_PATTERNS" + + +@pytest.mark.asyncio +async def test_detailed_github_captures_identity_and_scopes(): + class _R: + status_code = 200 + headers = {"x-oauth-scopes": "repo, read:org"} + def json(self): return {"login": "acme-bot"} + class _C: + async def get(self, *a, **k): return _R() + async def post(self, *a, **k): return _R() + res = await verifier.verify_finding_detailed("GitHub Personal Access Token", "ghp_x", _C()) + assert res.status == "verified" + assert "acme-bot" in res.detail and "repo" in res.detail + + +@pytest.mark.asyncio +async def test_detailed_backward_compatible_string_api(): + # verify_finding() still returns a bare status string. + s = await verifier.verify_finding("GitHub Personal Access Token", "ghp_x", _MockClient(_Resp(200))) + assert s == "verified" + + +@pytest.mark.asyncio +async def test_detailed_no_detail_when_body_empty(): + # 200 but no login/scopes (mock without headers) → verified, empty detail, no crash. + res = await verifier.verify_finding_detailed("GitHub Personal Access Token", "ghp_x", _MockClient(_Resp(200))) + assert res.status == "verified" and res.detail == "" + + +@pytest.mark.asyncio +async def test_detailed_fails_closed(): + class _Boom: + async def get(self, *a, **k): raise RuntimeError("down") + async def post(self, *a, **k): raise RuntimeError("down") + res = await verifier.verify_finding_detailed("OpenAI API Key", "sk-x", _Boom()) + assert res.status == "unverified" and res.detail == "" + + +@pytest.mark.asyncio +async def test_r6_verifiers_registered(): + for t in ("Cloudflare API Token", "DigitalOcean PAT", "Datadog API Key", + "Notion Integration Token", "Linear API Key", + "Figma Personal Access Token", "Postman API Key", "Doppler Token"): + assert verifier.is_supported(t), t + + +@pytest.mark.asyncio +async def test_cloudflare_active_vs_disabled(): + active = await verifier.verify_finding_detailed( + "Cloudflare API Token", "cf", _MockClient(_Resp(200, {"success": True, "result": {"status": "active"}}))) + disabled = await verifier.verify_finding_detailed( + "Cloudflare API Token", "cf", _MockClient(_Resp(200, {"success": True, "result": {"status": "disabled"}}))) + assert active.status == "verified" + assert disabled.status == "unverified" + + +@pytest.mark.asyncio +async def test_datadog_valid_field(): + ok = await verifier.verify_finding("Datadog API Key", "dd", _MockClient(_Resp(200, {"valid": True}))) + no = await verifier.verify_finding("Datadog API Key", "dd", _MockClient(_Resp(200, {"valid": False}))) + assert ok == "verified" and no == "unverified" + + +@pytest.mark.asyncio +async def test_linear_identity_extracted(): + res = await verifier.verify_finding_detailed( + "Linear API Key", "lin", + _MockClient(_Resp(200, {"data": {"viewer": {"name": "Ada", "email": "ada@x.com"}}}))) + assert res.status == "verified" and "ada@x.com" in res.detail + + +@pytest.mark.asyncio +async def test_figma_and_digitalocean_identity(): + fig = await verifier.verify_finding_detailed( + "Figma Personal Access Token", "fig", _MockClient(_Resp(200, {"handle": "ada"}))) + do = await verifier.verify_finding_detailed( + "DigitalOcean PAT", "do", _MockClient(_Resp(200, {"account": {"email": "ops@acme.io"}}))) + assert fig.status == "verified" and "ada" in fig.detail + assert do.status == "verified" and "ops@acme.io" in do.detail diff --git a/backend/verifier.py b/backend/verifier.py index e448004..f0a0590 100644 --- a/backend/verifier.py +++ b/backend/verifier.py @@ -1,16 +1,24 @@ """ -SecretNode — verifier.py (v2.3.0) +SecretNode — verifier.py (v2.6.0) Optional, OFF-BY-DEFAULT live credential verification — the "is this secret actually active?" step that modern scanners (e.g. TruffleHog's --only-verified) use to eliminate false-positive fatigue. Instead of only asking "does this look -like a secret?", we can ask "does this secret still work?" +like a secret?", we can ask "does this secret still work?" — and, when it does, +"WHO does it belong to and what can it reach?" + +v2.6.0 (R1 — verification enrichment): a successful check now also captures a +short, non-sensitive **identity/scope** detail (which account, which scopes, +which workspace) so a client report states the concrete blast radius of a live +key — e.g. "GitHub account @acme-bot · scopes: repo, read:org" — instead of a +bare "verified". This is the impact signal Cindrasec reports are built to sell. Each verifier makes exactly ONE read-only identity/"whoami" call to the credential's own provider (never to the scan target) and reports whether the credential is currently active. Verifiers: • are strictly read-only (no writes, no destructive calls), • never reveal or transmit the secret anywhere except to its own issuer, + • never store or return the secret itself — only a derived identity label, • fail closed (any error / timeout → "unverified", never a crash). This capability is disabled unless the operator sets VERIFY_SECRETS=true (or @@ -28,6 +36,7 @@ from __future__ import annotations import logging +from dataclasses import dataclass from typing import Any, Awaitable, Callable logger = logging.getLogger("secretnode.verifier") @@ -36,88 +45,259 @@ # A "client" here is any object exposing async .get()/.post() like httpx.AsyncClient, # which lets these functions be unit-tested with a lightweight mock. -Verifier = Callable[[str, Any], Awaitable[bool]] +# A verifier returns (active, detail): whether the credential is live, and a short +# non-sensitive identity/scope label ("" when nothing safe could be extracted). +Verifier = Callable[[str, Any], Awaitable["tuple[bool, str]"]] + + +@dataclass +class VerifyResult: + """Result of a live verification attempt. + + status — one of verified / unverified / unsupported + detail — short, non-sensitive identity/scope label for a VERIFIED credential + (empty otherwise); never contains the secret value itself. + """ + status: str + detail: str = "" + + +# ── helpers ────────────────────────────────────────────────────────────────── + +def _safe_json(r: Any) -> dict: + try: + data = r.json() + return data if isinstance(data, dict) else {} + except Exception: + return {} -async def _github(token: str, client: Any) -> bool: +def _safe_headers(r: Any) -> dict: + try: + h = getattr(r, "headers", {}) or {} + # httpx.Headers is mapping-like; dict() normalises for .get() + return {str(k).lower(): v for k, v in dict(h).items()} + except Exception: + return {} + + +def _detail(*parts: str) -> str: + """Join non-empty identity fragments into one short label.""" + return " · ".join(p for p in parts if p) + + +# ── verifiers (one read-only call each) ────────────────────────────────────── + +async def _github(token: str, client: Any) -> tuple[bool, str]: r = await client.get( "https://api.github.com/user", headers={"Authorization": f"Bearer {token}", "Accept": "application/vnd.github+json"}, timeout=VERIFY_TIMEOUT, ) - return r.status_code == 200 + if r.status_code != 200: + return False, "" + login = _safe_json(r).get("login", "") + scopes = _safe_headers(r).get("x-oauth-scopes", "") + return True, _detail(f"GitHub @{login}" if login else "", + f"scopes: {scopes}" if scopes else "") -async def _gitlab(token: str, client: Any) -> bool: +async def _gitlab(token: str, client: Any) -> tuple[bool, str]: r = await client.get( "https://gitlab.com/api/v4/user", headers={"PRIVATE-TOKEN": token}, timeout=VERIFY_TIMEOUT, ) - return r.status_code == 200 + if r.status_code != 200: + return False, "" + j = _safe_json(r) + user = j.get("username", "") + return True, _detail(f"GitLab @{user}" if user else "", + "admin" if j.get("is_admin") else "") -async def _stripe(token: str, client: Any) -> bool: +async def _stripe(token: str, client: Any) -> tuple[bool, str]: r = await client.get("https://api.stripe.com/v1/account", auth=(token, ""), timeout=VERIFY_TIMEOUT) - return r.status_code == 200 + if r.status_code != 200: + return False, "" + j = _safe_json(r) + acct = j.get("id", "") + mode = "LIVE mode" if token.startswith("sk_live") else "" + return True, _detail(f"Stripe {acct}" if acct else "", mode, + "charges enabled" if j.get("charges_enabled") else "") -async def _sendgrid(token: str, client: Any) -> bool: +async def _sendgrid(token: str, client: Any) -> tuple[bool, str]: r = await client.get( "https://api.sendgrid.com/v3/scopes", headers={"Authorization": f"Bearer {token}"}, timeout=VERIFY_TIMEOUT, ) - return r.status_code == 200 + if r.status_code != 200: + return False, "" + scopes = _safe_json(r).get("scopes", []) + can_send = "mail.send" in scopes if isinstance(scopes, list) else False + return True, _detail("can send mail" if can_send else "", + f"{len(scopes)} scopes" if isinstance(scopes, list) and scopes else "") -async def _openai(token: str, client: Any) -> bool: +async def _openai(token: str, client: Any) -> tuple[bool, str]: r = await client.get( "https://api.openai.com/v1/models", headers={"Authorization": f"Bearer {token}"}, timeout=VERIFY_TIMEOUT, ) - return r.status_code == 200 + if r.status_code != 200: + return False, "" + org = _safe_headers(r).get("openai-organization", "") + return True, _detail(f"org: {org}" if org and org != "user" else "") -async def _slack(token: str, client: Any) -> bool: +async def _slack(token: str, client: Any) -> tuple[bool, str]: r = await client.post( "https://slack.com/api/auth.test", headers={"Authorization": f"Bearer {token}"}, timeout=VERIFY_TIMEOUT, ) if r.status_code != 200: - return False - try: - return bool(r.json().get("ok")) - except Exception: - return False + return False, "" + j = _safe_json(r) + if not j.get("ok"): + return False, "" + team, user = j.get("team", ""), j.get("user", "") + return True, _detail(f"Slack {team}" if team else "", + f"as {user}" if user else "") -async def _npm(token: str, client: Any) -> bool: +async def _npm(token: str, client: Any) -> tuple[bool, str]: r = await client.get( "https://registry.npmjs.org/-/whoami", headers={"Authorization": f"Bearer {token}"}, timeout=VERIFY_TIMEOUT, ) - return r.status_code == 200 + if r.status_code != 200: + return False, "" + user = _safe_json(r).get("username", "") + return True, _detail(f"npm @{user}" if user else "") -async def _mailgun(token: str, client: Any) -> bool: +async def _mailgun(token: str, client: Any) -> tuple[bool, str]: r = await client.get( "https://api.mailgun.net/v3/domains", auth=("api", token), timeout=VERIFY_TIMEOUT ) - return r.status_code == 200 + if r.status_code != 200: + return False, "" + n = _safe_json(r).get("total_count") + return True, _detail(f"{n} domain(s)" if isinstance(n, int) else "") -async def _telegram(token: str, client: Any) -> bool: +async def _telegram(token: str, client: Any) -> tuple[bool, str]: r = await client.get(f"https://api.telegram.org/bot{token}/getMe", timeout=VERIFY_TIMEOUT) if r.status_code != 200: - return False - try: - return bool(r.json().get("ok")) - except Exception: - return False + return False, "" + j = _safe_json(r) + if not j.get("ok"): + return False, "" + bot = (j.get("result") or {}).get("username", "") if isinstance(j.get("result"), dict) else "" + return True, _detail(f"Telegram bot @{bot}" if bot else "") + + +async def _cloudflare(token: str, client: Any) -> tuple[bool, str]: + r = await client.get( + "https://api.cloudflare.com/client/v4/user/tokens/verify", + headers={"Authorization": f"Bearer {token}"}, timeout=VERIFY_TIMEOUT, + ) + if r.status_code != 200: + return False, "" + j = _safe_json(r) + res = j.get("result") if isinstance(j.get("result"), dict) else {} + active = bool(j.get("success")) and res.get("status", "active") == "active" + return (True, "") if active else (False, "") + + +async def _digitalocean(token: str, client: Any) -> tuple[bool, str]: + r = await client.get( + "https://api.digitalocean.com/v2/account", + headers={"Authorization": f"Bearer {token}"}, timeout=VERIFY_TIMEOUT, + ) + if r.status_code != 200: + return False, "" + acct = _safe_json(r).get("account") + email = acct.get("email", "") if isinstance(acct, dict) else "" + return True, _detail(f"DigitalOcean {email}" if email else "") + + +async def _datadog(token: str, client: Any) -> tuple[bool, str]: + r = await client.get( + "https://api.datadoghq.com/api/v1/validate", + headers={"DD-API-KEY": token}, timeout=VERIFY_TIMEOUT, + ) + if r.status_code != 200: + return False, "" + return (True, "") if _safe_json(r).get("valid") else (False, "") + + +async def _notion(token: str, client: Any) -> tuple[bool, str]: + r = await client.get( + "https://api.notion.com/v1/users/me", + headers={"Authorization": f"Bearer {token}", "Notion-Version": "2022-06-28"}, + timeout=VERIFY_TIMEOUT, + ) + if r.status_code != 200: + return False, "" + name = _safe_json(r).get("name", "") + return True, _detail(f"Notion {name}" if name else "") + + +async def _linear(token: str, client: Any) -> tuple[bool, str]: + r = await client.post( + "https://api.linear.app/graphql", + headers={"Authorization": token, "Content-Type": "application/json"}, + json={"query": "{ viewer { name email } }"}, timeout=VERIFY_TIMEOUT, + ) + if r.status_code != 200: + return False, "" + viewer = ((_safe_json(r).get("data") or {}).get("viewer") or {}) + if not isinstance(viewer, dict) or not viewer: + return False, "" + who = viewer.get("email") or viewer.get("name") or "" + return True, _detail(f"Linear {who}" if who else "") + + +async def _figma(token: str, client: Any) -> tuple[bool, str]: + r = await client.get( + "https://api.figma.com/v1/me", + headers={"X-Figma-Token": token}, timeout=VERIFY_TIMEOUT, + ) + if r.status_code != 200: + return False, "" + j = _safe_json(r) + who = j.get("handle") or j.get("email") or "" + return True, _detail(f"Figma {who}" if who else "") + + +async def _postman(token: str, client: Any) -> tuple[bool, str]: + r = await client.get( + "https://api.getpostman.com/me", + headers={"X-Api-Key": token}, timeout=VERIFY_TIMEOUT, + ) + if r.status_code != 200: + return False, "" + user = _safe_json(r).get("user") + who = user.get("username", "") if isinstance(user, dict) else "" + return True, _detail(f"Postman {who}" if who else "") + + +async def _doppler(token: str, client: Any) -> tuple[bool, str]: + r = await client.get( + "https://api.doppler.com/v3/me", + headers={"Authorization": f"Bearer {token}"}, timeout=VERIFY_TIMEOUT, + ) + if r.status_code != 200: + return False, "" + wp = _safe_json(r).get("workplace") + who = wp.get("name", "") if isinstance(wp, dict) else "" + return True, _detail(f"Doppler {who}" if who else "") # secret_type (from scanner.SECRET_PATTERNS) -> verifier @@ -136,6 +316,15 @@ async def _telegram(token: str, client: Any) -> bool: "npm Access Token": _npm, "Mailgun API Key": _mailgun, "Telegram Bot Token": _telegram, + # R6 — additional read-only whoami/validate verifiers for existing detectors. + "Cloudflare API Token": _cloudflare, + "DigitalOcean PAT": _digitalocean, + "Datadog API Key": _datadog, + "Notion Integration Token": _notion, + "Linear API Key": _linear, + "Figma Personal Access Token": _figma, + "Postman API Key": _postman, + "Doppler Token": _doppler, } @@ -143,14 +332,21 @@ def is_supported(secret_type: str) -> bool: return secret_type in VERIFIERS -async def verify_finding(secret_type: str, raw_value: str, client: Any) -> str: - """Return one of: 'verified', 'unverified', 'unsupported'. Never raises.""" +async def verify_finding_detailed(secret_type: str, raw_value: str, client: Any) -> VerifyResult: + """Verify a credential and, if active, capture a short identity/scope label. + Returns a VerifyResult. Never raises (fails closed to 'unverified').""" fn = VERIFIERS.get(secret_type) if fn is None: - return "unsupported" + return VerifyResult("unsupported", "") try: - active = await fn(raw_value, client) - return "verified" if active else "unverified" + active, detail = await fn(raw_value, client) + return VerifyResult("verified", detail) if active else VerifyResult("unverified", "") except Exception as exc: # noqa: BLE001 — fail closed, never crash a scan logger.warning("Verification error for %s: %s", secret_type, exc) - return "unverified" + return VerifyResult("unverified", "") + + +async def verify_finding(secret_type: str, raw_value: str, client: Any) -> str: + """Backward-compatible status-only API. Return one of: + 'verified', 'unverified', 'unsupported'. Never raises.""" + return (await verify_finding_detailed(secret_type, raw_value, client)).status diff --git a/docs/TECHNICAL-AUDIT-AND-ROADMAP.md b/docs/TECHNICAL-AUDIT-AND-ROADMAP.md new file mode 100644 index 0000000..aa06285 --- /dev/null +++ b/docs/TECHNICAL-AUDIT-AND-ROADMAP.md @@ -0,0 +1,117 @@ +# SecretNode — Technical Audit & Enhancement Roadmap +*Prepared 18 Jul 2026 · baseline: v2.5.4 · grounded in 2026 secret-scanning SOTA (TruffleHog, Gitleaks, GitHub Secret Scanning).* + +## Executive summary +SecretNode is **already an industrial-grade, well-architected tool** — not a rescue case. v2.5.4 ships +a layered detection pipeline, 147 passing tests, CI, Docker, SARIF/HTML/CSV/JSON, a CLI, a GitHub +Action, live verification, and AI validation. The right engineering move on a mature codebase is +**measured, tested capability expansion — not a blind rewrite.** This document is honest about what's +strong, what's genuinely missing versus the state of the art, and the sequence to close the gaps. + +**Its real niche (and why it fits Cindrasec):** TruffleHog and Gitleaks scan **git history / repos**. +SecretNode scans the **live, deployed public attack surface** — client-side JS, source maps, pages — +with **AI validation + impact-first reporting**. That's a differentiated slice and exactly on-brand +for Cindrasec ("attack surface", "sell impact, not noise"). Don't try to out-TruffleHog TruffleHog at +repo scanning; win the *web-surface* niche. + +## Current strengths (real design, credited) +- **Verification-first** — 14 live, read-only, fail-closed verifiers ("is this key still active?"). This + is the TruffleHog-grade differentiator over regex-only tools like Gitleaks. +- **AI contextual validation** (Gemini structured output) with a **public-by-design downgrade** + (Firebase web key, `pk_live`, Sentry DSN → INFO). Strong, and unusual in OSS. +- **Impact-aware severity** — leads with blast radius, not pattern shape. On-brand. +- **Layered FP/FN control** — placeholder/example allowlist, Shannon-entropy gate, base64-decode pass, + fingerprint de-dup. +- **Operational maturity** — scan diffing (NEW/RECURRING), FP suppression, needs-review findings are + never silently dropped, SSRF guard, same-scope restriction, auth, redaction-before-dispatch. + +## Honest gaps vs. 2026 SOTA (grounded in the code) +1. **Verification depth.** Verifiers return only `verified/unverified`. TruffleHog surfaces the + *identity + scopes* a live key maps to (which account, what permissions). That detail is the + strongest possible "impact" statement — and Cindrasec sells impact. +2. **No FP/FN measurement harness.** There's good FP *handling* but no labeled corpus + precision/recall + report, so changes aren't measured. Industrial tools track precision/recall on a benchmark. +3. **Regex robustness.** No ReDoS/catastrophic-backtracking audit or regex timeout; a hostile minified + bundle could stall a detector. No composite/proximity rules (a Gitleaks 2026 feature) for generic + high-FP patterns. +4. **Surface coverage.** Mines JS + source maps well, but modern leaks also hide in inline JSON + (`__NEXT_DATA__`, `window.__INITIAL_STATE__`), HTML comments, source-map `sourcesContent`, wasm + strings, and common exposed paths (`.env`, `.git/config`, `config.js`, backups). All authorized-only. +5. **Detector breadth.** ~54 patterns vs TruffleHog's 700+. Quality > quantity, but high-impact + providers are missing (Twilio, GCP service-account JSON, Azure AD, Cloudflare, Shopify, Supabase + `service_role`, Vercel, Notion). Each new detector should ship *with* a verifier where safe. +6. **AI dependency.** Validation leans on Gemini; the non-AI path exists (needs_review) but a stronger + offline heuristic tier would make CI-only/air-gapped use first-class. +7. **Performance.** No cross-scan asset caching (ETag / If-Modified-Since); re-scans refetch everything. +8. **ASM breadth.** SecretNode is the *secrets* slice. Cindrasec's brand promises broader ASM + (subdomains, exposed panels, misconfig, dangling DNS). That's a larger, later expansion — keep the + secrets core excellent first (scope discipline). + +## Roadmap — sequenced, each a shippable, tested unit + +### Tier 1 — correctness & brand value (do first) +- **R1 · Verification enrichment.** ✅ **DONE 18 Jul** — a verified credential now yields a short, + non-sensitive identity/scope label (GitHub @acct+scopes, Stripe account+LIVE/charges, Slack + workspace/user, OpenAI org, npm/GitLab/Telegram handle, SendGrid send-scope, Mailgun domain count), + surfaced in HTML/CSV/SARIF as the concrete blast radius. Backward-compatible API; +7 tests. +- **R2 · FP/FN benchmark harness.** ✅ **DONE 18 Jul** — `backend/bench/` labelled corpus (12 synthetic + positives + 15 placeholders/examples/noise), `make bench` reporting **precision/recall/F1**, and a + pytest CI gate (`test_bench.py`) that fails the build on a precision/recall regression. Current: + **precision 1.000 · recall 1.000 · F1 1.000 · 0 false positives.** The harness immediately caught a + malformed test key, confirming the OpenAI detector correctly requires real key structure. +4 tests. +- **R3 · Regex safety.** ✅ **DONE 18 Jul** — a per-pattern match cap (defence-in-depth against + match-flood blobs) plus an automated ReDoS gate: empirical wall-clock fuzz over all 54 detectors × + 17 adversarial 50 KB inputs, a static nested-quantifier guard, and a cap-engagement test. Proves no + catastrophic backtracking and gates future pattern additions. +3 tests. +- **R4 · SARIF full detector catalog.** ✅ **DONE 18 Jul** — the driver now advertises every detector as + a SARIF rule (help text, CWE, severity), even on clean scans. +2 tests. + +### Tier 2 — coverage +- **R5 · Surface expansion.** ✅ **DONE 18 Jul (source-map slice)** — decode a source map's embedded + original source (`sourcesContent`) and scan it as real code, with precise per-file attribution + (`app.js.map → src/config.js`). Replaces raw-`.map` scanning for maps that carry source, which also + removes the high-entropy `mappings` VLQ as a false-positive source. Env-tunable + (`SCAN_SOURCEMAP_CONTENT`, `MAX_SOURCEMAP_SOURCES`); fully defensive; +6 tests. + *Note:* inline JSON (`__NEXT_DATA__`, `__INITIAL_STATE__`) and HTML comments are **already** covered — + the scanner scans the full HTML body wholesale, so they are in scope today. + *Remaining R5 follow-up:* opt-in authorized known-path probe (`.env`, `.git/config`, `config.js`) — + the one genuinely new *network* surface; deferred as higher-risk (extra requests, needs SSRF/scope + review). [MED] +- **R6 · Detector + verifier expansion.** ✅ **DONE 18 Jul (verifier slice)** — added 8 read-only + whoami/validate verifiers for existing detectors that previously returned `unsupported`: Cloudflare, + DigitalOcean, Datadog, Notion, Linear, Figma, Postman, Doppler. Live-verification coverage now spans + **17 providers** (was 9); each extracts R1 identity where available, is read-only, and fails closed. + Chosen deliberately over adding new *detectors* — this widens the verification-first differentiator + with **zero new false-positive risk** (no new regexes). +6 tests. *Follow-up:* new detectors (Twilio + SID-paired, GCP SA JSON, Supabase service_role) when each can ship with a corpus entry + verifier. +- **R7 · Composite/proximity rule engine** for generic high-FP patterns (Gitleaks-style). [MED] + +### Tier 3 — ASM breadth (fulfills the brand fully; larger) +- **R8 · Passive attack-surface map.** ✅ **DONE 18 Jul (security-posture slice)** — `posture.py` + analyses the target root's own response for missing/weak security headers (HSTS, CSP, clickjacking, + X-Content-Type-Options, Referrer/Permissions-Policy), version disclosure, and insecure cookies. Pure + passive analysis (no exploitation, no third-party calls); each issue carries severity/CWE/evidence/ + remediation and renders in a dedicated report section + KPI tile — so even a clean *credential* scan + now returns actionable ASM findings. Env-tunable (`SCAN_HTTP_POSTURE`); +11 tests. This is the first + step from "secret scanner" toward the full attack-surface scanner the brand promises. + *Remaining R8 follow-up (higher-risk / external network):* CT-log subdomain discovery (crt.sh), + DNS resolution + dangling-CNAME takeover checks; posture in CSV/SARIF export. [HIGH] + +### Tier 4 — polish +- **R9 · Executive-summary report page.** ✅ **DONE 18 Jul** — the HTML report now leads with a + verification-evidence callout (each verified-active key + its R1 identity/scope as "confirmed live + access"), a "Verified Active" KPI tile, and an honest measured-precision "Detection quality" + statement (verification-first + CI-gated precision/recall). Turns the engine work into a + client-ready deliverable. +4 tests. +- **R10 · Asset caching** (ETag/If-Modified-Since) + per-provider verify concurrency. [LOW] +- **R11 · Distribution** — PyPI publish, tagged releases, docs. [LOW] + +## Recommended next steps (highest ROI for Cindrasec's stage) +Pre-pilot, the biggest wins are **R1** (impact-rich verification — makes client reports sell), +**R2** (precision/recall harness — credibility + the explicit FP/FN ask), and **R5** (finds more real +leaks). **Chase impact + precision + surface — not a race to 700 detectors.** Keep the secrets core +excellent before broadening to full ASM (R8). + +> Engineering honesty: "enhance every single aspect in one pass" is the wrong move on a working, +> 147-test tool — it trades reliability for the appearance of progress. The right path is incremental, +> measured, tested slices. R4 shipped today; R1 and R2 are the recommended next slices.
    SeverityTypeLocationNote