Compare commits
2 Commits
d28da26000
...
ea02ade0cb
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
ea02ade0cb | ||
|
|
497222094f |
@@ -61,13 +61,35 @@ def test_allow_is_by_hash_only():
|
||||
F = sg.cg.Finding
|
||||
fake = F("tests/t.py", 3, "telegram bot token", f"telegram bot token #{ss.h8(TG)}")
|
||||
real = F("tests/t.py", 9, "telegram bot token", "telegram bot token #deadbeef")
|
||||
kept = sg._drop_allowed("windy-chat", [fake, real], {"windy-chat": {ss.h8(TG)}})
|
||||
assert kept == [real]
|
||||
assert sg._drop_allowed("windy-mail", [fake], {"windy-chat": {ss.h8(TG)}}) == [fake]
|
||||
allow = {"windy-chat": {"hashes": {ss.h8(TG)}, "paths": []}}
|
||||
assert sg._drop_allowed("windy-chat", [fake, real], allow) == [real]
|
||||
assert sg._drop_allowed("windy-mail", [fake], allow) == [fake]
|
||||
|
||||
|
||||
def test_private_key_blocks_are_allowed_by_path_never_by_hash():
|
||||
F = sg.cg.Finding
|
||||
hdr = "private key block #" + ss.h8("-----BEGIN PRIVATE KEY-----")
|
||||
test_key = F("tests/keys/test.pem", 1, "private key block", hdr)
|
||||
prod_key = F("deploy/prod.pem", 1, "private key block", hdr)
|
||||
allow = {"r": {"hashes": {hdr.rsplit("#", 1)[1]}, "paths": [("tests/keys/*", {"private key block"})]}}
|
||||
assert sg._drop_allowed("r", [test_key, prod_key], allow) == [prod_key]
|
||||
|
||||
|
||||
def test_path_allow_cannot_cover_real_token_kinds(tmp_path):
|
||||
bad = tmp_path / "a.yml"
|
||||
bad.write_text("allow:\n - repo: r\n paths: [tests/*]\n kinds: [telegram bot token]\n reason: no\n")
|
||||
with pytest.raises(ValueError):
|
||||
sg.load_allow(bad)
|
||||
|
||||
|
||||
def test_shipped_allow_file_never_excuses_the_real_leaked_tokens():
|
||||
a = sg.load_allow()
|
||||
every = set().union(*(v["hashes"] for v in a.values())) if a else set()
|
||||
assert not {"1354fc9b", "d49dc2ba"} & every # real (now revoked) credentials: remove, never allow
|
||||
|
||||
|
||||
def test_allow_file_loads_and_needs_reasons(tmp_path):
|
||||
assert sg.load_allow() == {} or isinstance(sg.load_allow(), dict)
|
||||
assert isinstance(sg.load_allow(), dict)
|
||||
bad = tmp_path / "a.yml"
|
||||
bad.write_text("allow:\n - repo: r\n hashes: [abcd1234]\n")
|
||||
with pytest.raises(ValueError):
|
||||
@@ -82,3 +104,16 @@ def test_block_and_warn(monkeypatch):
|
||||
monkeypatch.setattr(sg, "MODE", "warn")
|
||||
state, desc, _ = sg.status_for([f], True)
|
||||
assert state == "success" and desc.startswith("⚠ WARN (not blocking): 1 secret-shaped string in tree")
|
||||
|
||||
|
||||
def test_public_scan_excuses_by_hash_and_by_path():
|
||||
spec = importlib.util.spec_from_file_location("public_secret_scan", ROOT / "scripts" / "public_secret_scan.py")
|
||||
ps = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(ps)
|
||||
allow = {"windy-agent": {"hashes": {"aaaa1111"}, "paths": [("tests/keys/*", {"private key block"})]}}
|
||||
assert ps.excused("windy-agent", "openai key", "aaaa1111", ["tests/x.py"], allow)
|
||||
assert not ps.excused("windy-agent", "telegram bot token", "1354fc9b", ["tests/test_log_redaction.py"], allow)
|
||||
assert ps.excused("windy-agent", "private key block", "ffff0000", ["tests/keys/a.pem"], allow)
|
||||
# a PEM header anywhere outside the allowed paths still counts
|
||||
assert not ps.excused("windy-agent", "private key block", "ffff0000", ["tests/keys/a.pem", "deploy/k.pem"], allow)
|
||||
assert not ps.excused("other", "openai key", "aaaa1111", ["x"], allow)
|
||||
|
||||
@@ -1,5 +1,50 @@
|
||||
# Secret guard allow-list: KNOWN FAKE values that look like secrets (test
|
||||
# fixtures, docs). Allowed BY HASH (sha256[:8] of the value), never by path, so a
|
||||
# real secret in the same file still flags. Every entry MUST say why.
|
||||
# Owner: Windy Git lane (13); changes go through the orchestrator.
|
||||
allow: []
|
||||
# fixtures, docs). Allowed BY HASH (sha256[:8] of the value), so a real secret
|
||||
# in the same file still flags. Private-key blocks (the match is only the BEGIN
|
||||
# line, same hash everywhere) are allowed by PATH + kind instead.
|
||||
# Every entry MUST say why. Owner: Windy Git lane (13); changes via the orchestrator.
|
||||
# Triage 09-24 (values never printed): each hash checked against every version of
|
||||
# the lockbox; fakes judged by impossible length for the kind (real Anthropic keys
|
||||
# ~108 chars, OpenAI 51 or 160+), fake-words, or identity with upstream public
|
||||
# fixtures. NOT allowed, remove instead: 1354fc9b (old @Windy_0_bot token) and
|
||||
# d49dc2ba (old Anthropic key), both real and revoked, in public windy-agent.
|
||||
allow:
|
||||
- repo: windy-code
|
||||
hashes: [ac9265e5, 46eb1235]
|
||||
reason: "Upstream microsoft/vscode terminalEnvironment.test.ts fixtures (identical hash upstream; public)."
|
||||
- repo: windy-code
|
||||
paths: ["build/azure-pipelines/common/publish.ts"]
|
||||
kinds: [private key block]
|
||||
reason: "Upstream VS Code build script (PEM header string in code, not a key)."
|
||||
- repo: windy-agent
|
||||
hashes: [a9235a6d, dd6a2baa, 02c362d8, a6f6ff79, f7503b21, d0c94833, 4ab092e3, e3aa1eb8, 833382ee]
|
||||
reason: "Redaction/sanitizer test fixtures; lengths impossible for real Anthropic/OpenAI keys; never in the lockbox."
|
||||
- repo: windy-agent
|
||||
paths: ["tests/test_agent_keys.py"]
|
||||
kinds: [private key block]
|
||||
reason: "Test-generated key material for agent-key tests."
|
||||
- repo: windy-mind
|
||||
hashes: [f8a630b2]
|
||||
reason: "Provider test fixture (27 chars; a real Anthropic key is ~108)."
|
||||
- repo: windy-pro
|
||||
hashes: [756de8d8, 7828319d, 1a5d44a2]
|
||||
reason: ".env.production.example placeholder + crash-summary test fixtures (AWS doc EXAMPLE key shape, short fake Slack token)."
|
||||
- repo: windy-pro
|
||||
paths: ["account-server/docs/oauth-providers.md"]
|
||||
kinds: [private key block]
|
||||
reason: "Docs show the PEM header format; no key material."
|
||||
- repo: windytalk
|
||||
hashes: [baf8656a]
|
||||
reason: "Diagnostics redaction test fixture (fake-word in value)."
|
||||
- repo: eternitas
|
||||
paths: ["tests/golden_vectors/**", "tests/test_soul_vault_key_separation.py"]
|
||||
kinds: [private key block]
|
||||
reason: "Test vectors and throwaway keys for signature/vault tests."
|
||||
- repo: windy-drops
|
||||
paths: ["tools/conformance/test-keys/*"]
|
||||
kinds: [private key block]
|
||||
reason: "Conformance-suite test keys (named test-private.pem)."
|
||||
- repo: windy-git
|
||||
paths: ["api/tests/test_secret_guard.py"]
|
||||
kinds: [private key block]
|
||||
reason: "The guard's own test uses a PEM header string as a sample."
|
||||
|
||||
118
scripts/public_secret_scan.py
Normal file
118
scripts/public_secret_scan.py
Normal file
@@ -0,0 +1,118 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Weekly: every PUBLIC repo in Grant's GitHub accounts, full history (every object,
|
||||
reachable or not, incl. PR refs), for secret-shaped strings. Leak hunt 09-24: a
|
||||
real bot token sat in a public test fixture for five months.
|
||||
|
||||
Output is hash + location only, never a value (house rule 10). Known fakes are
|
||||
excused by ci/secret-guard-allow.yml (same file as secret-guard; the repo NAME is
|
||||
matched across accounts). Runs on Veron as user1-gpu (gh is logged in there):
|
||||
|
||||
python3 scripts/public_secret_scan.py [--out FILE] [--owners a,b,...]
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import fnmatch
|
||||
import json
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||||
import secret_shapes as ss # noqa: E402
|
||||
|
||||
OWNERS = ["sneakyfree", "VERONTECH", "Windstorm-Institute", "Windstorm-Labs", "Public-Streamer"]
|
||||
WORK = Path.home() / "leakscan" / "public"
|
||||
MAX_BLOB = 20_000_000
|
||||
|
||||
|
||||
def _run(*a: str) -> subprocess.CompletedProcess:
|
||||
return subprocess.run(a, capture_output=True)
|
||||
|
||||
|
||||
def blob_hits(bare: Path) -> dict[str, list[tuple[str, str]]]:
|
||||
"""{blob: [(kind, hash8)]} over every blob object in the repo."""
|
||||
chk = _run("git", "-C", str(bare), "cat-file", "--batch-all-objects",
|
||||
"--batch-check=%(objectname) %(objecttype) %(objectsize)")
|
||||
blobs = [p[0] for p in (ln.split() for ln in chk.stdout.decode().splitlines())
|
||||
if len(p) == 3 and p[1] == "blob" and int(p[2]) < MAX_BLOB]
|
||||
if not blobs:
|
||||
return {}
|
||||
import threading
|
||||
p = subprocess.Popen(["git", "-C", str(bare), "cat-file", "--batch"], stdin=subprocess.PIPE, stdout=subprocess.PIPE)
|
||||
|
||||
def feed() -> None: # separate thread: writing everything first deadlocks (both pipes fill)
|
||||
p.stdin.write(("\n".join(blobs) + "\n").encode())
|
||||
p.stdin.close()
|
||||
threading.Thread(target=feed, daemon=True).start()
|
||||
out: dict[str, list[tuple[str, str]]] = {}
|
||||
for _ in blobs:
|
||||
hdr = p.stdout.readline().split()
|
||||
data = p.stdout.read(int(hdr[2]))
|
||||
p.stdout.read(1)
|
||||
found = ss.find(data.decode("utf-8", "ignore"))
|
||||
if found:
|
||||
out[hdr[0].decode()] = found
|
||||
p.wait()
|
||||
return out
|
||||
|
||||
|
||||
def locate(bare: Path, blob: str) -> dict:
|
||||
lg = _run("git", "-C", str(bare), "log", "--all", "--format=@@%H %cI", "--name-only",
|
||||
f"--find-object={blob}").stdout.decode()
|
||||
commits, paths = [], set()
|
||||
for line in lg.splitlines():
|
||||
if line.startswith("@@"):
|
||||
commits.append(line[2:].split())
|
||||
elif line.strip():
|
||||
paths.add(line.strip())
|
||||
head = _run("git", "-C", str(bare), "ls-tree", "-r", "HEAD").stdout.decode()
|
||||
first = commits[-1] if commits else ["unreachable", "-"]
|
||||
return {"paths": sorted(paths), "first_commit": first[0][:10], "first_date": first[1],
|
||||
"in_head": blob in head}
|
||||
|
||||
|
||||
def excused(repo: str, kind: str, h: str, paths: list[str], allow: dict) -> bool:
|
||||
a = allow.get(repo) or {"hashes": set(), "paths": []}
|
||||
if kind in {"private key block"}:
|
||||
return bool(paths) and all(any(kind in k and fnmatch.fnmatch(p, g) for g, k in a["paths"]) for p in paths)
|
||||
return h in a["hashes"]
|
||||
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("--out", default=str(WORK / "latest.json"))
|
||||
ap.add_argument("--owners", default=",".join(OWNERS))
|
||||
args = ap.parse_args()
|
||||
import secret_guard as sg
|
||||
allow = sg.load_allow()
|
||||
WORK.mkdir(parents=True, exist_ok=True)
|
||||
rows, scanned, errors = [], [], []
|
||||
for owner in args.owners.split(","):
|
||||
lst = _run("gh", "repo", "list", owner, "--limit", "1000", "--visibility", "public", "--json", "name")
|
||||
for r in json.loads(lst.stdout or b"[]"):
|
||||
name = r["name"]
|
||||
bare = WORK / owner / f"{name}.git"
|
||||
if bare.exists():
|
||||
c = _run("git", "-C", str(bare), "remote", "update", "--prune")
|
||||
else:
|
||||
bare.parent.mkdir(parents=True, exist_ok=True)
|
||||
c = _run("gh", "repo", "clone", f"{owner}/{name}", str(bare), "--", "--mirror", "-q")
|
||||
if c.returncode:
|
||||
errors.append(f"{owner}/{name}")
|
||||
continue
|
||||
scanned.append(f"{owner}/{name}")
|
||||
for blob, found in blob_hits(bare).items():
|
||||
loc = locate(bare, blob)
|
||||
for kind, h in sorted(set(found)):
|
||||
rows.append({"repo": f"{owner}/{name}", "kind": kind, "hash8": h, **loc,
|
||||
"excused": excused(name, kind, h, loc["paths"], allow)})
|
||||
Path(args.out).write_text(json.dumps({"scanned": scanned, "errors": errors, "rows": rows}, indent=1))
|
||||
new = [r for r in rows if not r["excused"]]
|
||||
print(f"scanned {len(scanned)} public repos, {len(errors)} errors, {len(rows)} secret-shaped, {len(new)} NOT excused")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -11,6 +11,7 @@ BY HASH in ci/secret-guard-allow.yml (repo + hashes + reason).
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import fnmatch
|
||||
import hashlib
|
||||
import os
|
||||
import re
|
||||
@@ -33,14 +34,24 @@ def path_ok(path: str) -> bool:
|
||||
return not NEVER.search(path)
|
||||
|
||||
|
||||
def load_allow(path: Path = ALLOW_FILE) -> dict[str, set[str]]:
|
||||
"""{repo: {hash8, ...}}; every entry needs a reason."""
|
||||
# A private-key match is only its BEGIN line, so its hash is the same everywhere:
|
||||
# those are allowed by PATH (entries with `paths` + `kinds`), everything else by HASH.
|
||||
PATH_ONLY_KINDS = {"private key block"}
|
||||
|
||||
|
||||
def load_allow(path: Path = ALLOW_FILE) -> dict[str, dict]:
|
||||
"""{repo: {"hashes": {hash8}, "paths": [(glob, {kind})]}}; every entry needs a reason."""
|
||||
data = yaml.safe_load(path.read_text()) if path.exists() else {}
|
||||
out: dict[str, set[str]] = {}
|
||||
out: dict[str, dict] = {}
|
||||
for e in (data or {}).get("allow") or []:
|
||||
if not (e.get("repo") and e.get("hashes") and str(e.get("reason", "")).strip()):
|
||||
raise ValueError(f"allow entry needs repo, hashes and a reason: {e}")
|
||||
out.setdefault(e["repo"], set()).update(str(h) for h in e["hashes"])
|
||||
if not (e.get("repo") and (e.get("hashes") or (e.get("paths") and e.get("kinds")))
|
||||
and str(e.get("reason", "")).strip()):
|
||||
raise ValueError(f"allow entry needs repo, hashes (or paths + kinds) and a reason: {e}")
|
||||
if e.get("paths") and not set(e["kinds"]) <= PATH_ONLY_KINDS:
|
||||
raise ValueError(f"path allows are only for {sorted(PATH_ONLY_KINDS)}: {e}")
|
||||
r = out.setdefault(e["repo"], {"hashes": set(), "paths": []})
|
||||
r["hashes"].update(str(h) for h in e.get("hashes") or [])
|
||||
r["paths"] += [(g, set(e["kinds"])) for g in e.get("paths") or []]
|
||||
return out
|
||||
|
||||
|
||||
@@ -48,9 +59,14 @@ def scan_line(path: str, text: str) -> list[tuple[str, str]]:
|
||||
return [(kind, f"{kind} #{h}") for kind, h in ss.find(text)]
|
||||
|
||||
|
||||
def _drop_allowed(repo: str, findings, allow: dict[str, set[str]]):
|
||||
ok = allow.get(repo, set())
|
||||
return [f for f in findings if f.match.rsplit("#", 1)[-1] not in ok]
|
||||
def _drop_allowed(repo: str, findings, allow: dict[str, dict]):
|
||||
a = allow.get(repo) or {"hashes": set(), "paths": []}
|
||||
|
||||
def ok(f) -> bool:
|
||||
if f.kind in PATH_ONLY_KINDS:
|
||||
return any(f.kind in kinds and fnmatch.fnmatch(f.path, g) for g, kinds in a["paths"])
|
||||
return f.match.rsplit("#", 1)[-1] in a["hashes"]
|
||||
return [f for f in findings if not ok(f)]
|
||||
|
||||
|
||||
def check(repo: str, sha: str, default_branch: str, is_default_head: bool):
|
||||
|
||||
Reference in New Issue
Block a user