chore(ci): sync secret gate to runpod/secret_detector@ee7e0f664181

Source: runpod/secret_detector@ee7e0f6

Author: erinavllazagaj-eng

.github/security/ai_security_review.py

@@ -312,13 +312,21 @@ def clip(text: str, keep_around: str = "", limit: int = MAX_LINE_CHARS) -> str:
     return text[:limit] + "…"
 
 
-def build_evidence(repo: str, token: str, findings: list[dict]):
+def build_evidence(repo: str, token: str, findings: list[dict],
+                   mask_also: list[dict] = ()):
     """Turn scanner findings into redacted windows. Returns (evidence, errors).
 
-    Any error fails the gate: an unreadable window is an unjudged detection.
+    `mask_also` holds detections past the triage cap: they get no window, but
+    are blanked wherever they fall inside one. Any error fails the gate: an
+    unreadable window is an unjudged detection.
     """
     evidence, errors = [], []
 
+    masks: dict[tuple[str, str], dict[int, dict]] = {}
+    for f in [*findings, *mask_also]:
+        if f["location_kind"] != "archive":
+            masks.setdefault((f["blob_sha"], f["file"]), {})[f["id"]] = f
+
     # Grouped by (blob, path) so identical files each get their own markers;
     # fetched once per blob.
     by_path: dict[tuple[str, str], list[dict]] = {}
@@ -331,7 +339,7 @@ def build_evidence(repo: str, token: str, findings: list[dict]):
             by_path.setdefault((f["blob_sha"], f["file"]), []).append(f)
 
     blobs: dict[str, tuple[bytes | None, str]] = {}
-    for (blob_sha, _path), group in by_path.items():
+    for (blob_sha, path), group in by_path.items():
         if blob_sha not in blobs:
             blobs[blob_sha] = fetch_blob(repo, blob_sha, token, MAX_BLOB_BYTES)
         raw, reason = blobs[blob_sha]
@@ -343,8 +351,8 @@ def build_evidence(repo: str, token: str, findings: list[dict]):
         # Split on b"\n" only, to match the scanner's line numbers; decode
         # after masking.
         lines = raw.split(b"\n")
-        masked = [m.decode("utf-8", "replace")
-                  for m in mask_detections(lines, group)]
+        in_file = list(masks[(blob_sha, path)].values())
+        masked = [m.decode("utf-8", "replace") for m in mask_detections(lines, in_file)]
 
         for f in group:
             if f["location_kind"] == "path":
@@ -684,7 +692,8 @@ def render_comment(review, scan_conclusion, blocked, note=""):
     findings = review["findings"]
     if not findings:
         # No green tick under a red banner.
-        clean = not (review["injection"] or review["unanswered"] or blocked)
+        clean = not (review["injection"] or review["unanswered"] or blocked
+                     or scan_conclusion == "failure")
         L += ["✅ No secrets in the changed files." if clean
               else "No secrets among the detections that *were* triaged.", ""]
     for f in findings:
@@ -804,7 +813,7 @@ def main() -> int:
         else:
             # Runs even when `blocked`: the status stays red, but reviewers still
             # need verdicts on what was scanned.
-            evidence, errors = build_evidence(repo, gh_token, findings)
+            evidence, errors = build_evidence(repo, gh_token, findings, scan["findings"])
             if errors:
                 blocked = blocked or f"{len(errors)} detection(s) could not be read"
                 for e in errors[:10]:

.github/security/sanitize_findings.py

@@ -10,21 +10,26 @@
 from __future__ import annotations
 
 import argparse
+import hashlib
+import hmac
 import json
 import os
+import secrets
 import sys
 
 from redaction import build_redaction_map, redact, safe_rule_name
 
 
-def fingerprint(secret: str) -> str:
-    """Length plus first/last character, so reviewers can correlate findings.
+# Per-run key, never stored, so a digest cannot be brute-forced the way a bare
+# hash of a short secret can.
+_RUN_KEY = secrets.token_bytes(32)
 
-    Not a hash: a hash of a short secret is brute-forceable.
-    """
+
+def fingerprint(secret: str) -> str:
+    """Keyed digest, so reviewers can correlate findings within one report."""
     if not secret:
         return ""
-    return f"len={len(secret)}:{secret[0]}…{secret[-1]}"
+    return hmac.new(_RUN_KEY, secret.encode(), hashlib.sha256).hexdigest()[:12]
 
 
 def sanitize_gitleaks(findings: list) -> list[dict]:

.github/security/scan_head.py

@@ -135,22 +135,22 @@ def run_betterleaks(work: str, config: str, report: str) -> None:
     `report` must be absolute.
     """
     # Empty per-run dir; the ignore-path flag otherwise defaults to the scan target.
-    noignore = tempfile.mkdtemp(prefix="bl-noignore-")
-    proc = subprocess.run(
-        [
-            "betterleaks", "git", ".",
-            f"--config={config}",
-            f"--gitleaks-ignore-path={noignore}",
-            "--ignore-gitleaks-allow",
-            "--report-format=json",
-            f"--report-path={report}",
-            # Findings exit 0, so any non-zero exit is a scanner error.
-            "--exit-code=0",
-            "--max-target-megabytes=25",
-            "--no-banner",
-        ],
-        cwd=work, capture_output=True, text=True,
-    )
+    with tempfile.TemporaryDirectory(prefix="bl-noignore-") as noignore:
+        proc = subprocess.run(
+            [
+                "betterleaks", "git", ".",
+                f"--config={config}",
+                f"--gitleaks-ignore-path={noignore}",
+                "--ignore-gitleaks-allow",
+                "--report-format=json",
+                f"--report-path={report}",
+                # Findings exit 0, so any non-zero exit is a scanner error.
+                "--exit-code=0",
+                "--max-target-megabytes=25",
+                "--no-banner",
+            ],
+            cwd=work, capture_output=True, text=True,
+        )
     sys.stderr.write(proc.stderr[-4000:])
     if proc.returncode != 0:
         raise RuntimeError(f"betterleaks exited {proc.returncode}")
@@ -299,11 +299,11 @@ def main() -> int:
     if altered:
         log(f"::error title=Files not scanned::{len(altered)} materialised file(s) "
             "did not reach the scan commit byte-identically")
-    report_path = os.path.join(os.path.dirname(out_path) or ".", "betterleaks.raw.json")
-    run_betterleaks(work, config, report_path)
+    raw_report = os.path.join(os.path.dirname(out_path) or ".", "betterleaks.raw.json")
+    run_betterleaks(work, config, raw_report)
 
     blob_by_path = {f["path"]: f["blob_sha"] for f in written}
-    raw = load_report(report_path)
+    raw = load_report(raw_report)
     findings, unmappable = merge_findings(raw, blob_by_path, work)
     skipped.extend(unmappable)
     merged = len(raw) - len(findings) - len(unmappable)
@@ -314,7 +314,7 @@ def main() -> int:
 
     # The raw report holds plaintext secrets.
     try:
-        os.remove(report_path)
+        os.remove(raw_report)
     except FileNotFoundError:
         pass
     except OSError as exc:

.github/security/test_secret_gate.py

@@ -16,6 +16,7 @@
 
 import ai_security_review as air
 import redaction
+import sanitize_findings
 import scan_head
 
 # --------------------------------------------------------------------- helpers
@@ -593,6 +594,7 @@ def fake_run(args, **kw):
     scan_head.run_betterleaks(str(tmp_path), "cfg.toml", str(tmp_path / "r.json"))
     assert seen["dir"] != "/tmp/bl-noignore"
     assert seen["listing"] == []
+    assert not os.path.exists(seen["dir"]), "left the ignore dir behind"
 
 
 @pytest.mark.parametrize("line, masked", [
@@ -623,3 +625,45 @@ def test_every_gate_workflow_is_valid_yaml():
     for f in sorted(workflows.glob("*.yml")):
         doc = yaml.safe_load(f.read_text())
         assert isinstance(doc, dict) and doc.get("jobs"), f.name
+
+
+# ------------------------------------------------- detections past the cap
+
+
+def test_detection_past_the_cap_is_masked_in_a_kept_window(monkeypatch):
+    """`redact` misses this value, so only the span mask keeps it from the model."""
+    kept, dropped = "kkkkQ7zv", "hunter2-but-longer"
+    lines = [f'first = "{kept}"', f'second = "{dropped}"']
+    monkeypatch.setattr(air, "fetch_blob",
+                        lambda *a, **k: ("\n".join(lines).encode(), ""))
+
+    def det(fid, n, value):
+        at = lines[n - 1].index(value)
+        return {**_finding(fid, "generic-api-key", n, at + 1, at + len(value)),
+                "location_kind": "line", "blob_sha": "cafe", "file": "a.py"}
+
+    first, second = det(1, 1, kept), det(2, 2, dropped)
+    assert dropped in air.build_evidence("o/r", "tok", [first])[0][0]["window"]
+
+    evidence, errors = air.build_evidence("o/r", "tok", [first], [first, second])
+
+    assert errors == [] and len(evidence) == 1
+    assert dropped not in evidence[0]["window"]
+    assert "«DETECTION-2:" in evidence[0]["window"]
+
+
+# ------------------------------------------------- comment and fingerprint
+
+
+def test_no_green_tick_when_stage_1_did_not_finish_cleanly():
+    review = {"findings": [], "false_positives": 0, "unanswered": [],
+              "injection": False}
+    assert "✅" in air.render_comment(review, "success", "")
+    assert "✅" not in air.render_comment(review, "failure", "")
+
+
+def test_fingerprint_correlates_without_revealing_the_secret():
+    a = sanitize_findings.fingerprint("s3cr3t-value")
+    assert a == sanitize_findings.fingerprint("s3cr3t-value")
+    assert a != sanitize_findings.fingerprint("s3cr3t-valuf")
+    assert re.fullmatch(r"[0-9a-f]{12}", a), "carries more than a keyed digest"

.github/workflows/pr-security-scan.yml

@@ -190,17 +190,23 @@ jobs:
           fi
 
           if [ -s /tmp/changed.z ]; then
-            # Trailing `--` so a file named "--config=evil.py" is a path, not a flag.
-            xargs -0 -a /tmp/changed.z \
+            # xargs splits a long file list into several semgrep runs, so each run
+            # writes its own report and they are merged below. An empty report
+            # means that run did not complete (e.g. registry unreachable); it marks
+            # the whole scan as not scanned rather than as zero findings. Trailing
+            # `--` so a file named "--config=evil.py" is a path, not a flag.
+            # shellcheck disable=SC2016  # the inner sh expands $out and $@
+            xargs -0 -a /tmp/changed.z sh -c '
+              out=$(mktemp /tmp/semgrep.part.XXXXXX)
               /tmp/sgvenv/bin/semgrep scan \
                 --config=p/secrets --config=p/security-audit --config=p/owasp-top-ten \
-                --json --output=/tmp/semgrep.raw.json \
-                --metrics=off --quiet --timeout=120 -- || true
-            # No report means the scan did not complete (e.g. registry unreachable);
-            # mark it as not scanned rather than as zero findings.
-            [ -f /tmp/semgrep.raw.json ] \
-              || echo '{"results":[],"__not_scanned":"scan did not complete"}' \
-                   > /tmp/semgrep.raw.json
+                --json --output="$out" --metrics=off --quiet --timeout=120 -- "$@" || true
+              [ -s "$out" ] || echo "{\"results\":[],\"__not_scanned\":\"scan did not complete\"}" > "$out"
+            ' sh || true
+            jq -s '{results: (map(.results // []) | add)}
+                   + (map(.__not_scanned // empty) | if length > 0 then {__not_scanned: .[0]} else {} end)' \
+              /tmp/semgrep.part.* > /tmp/semgrep.raw.json \
+              || echo '{"results":[],"__not_scanned":"scan did not complete"}' > /tmp/semgrep.raw.json
           else
             echo '{"results":[]}' > /tmp/semgrep.raw.json
           fi