diff --git a/api/tests/test_secret_scan.py b/api/tests/test_secret_scan.py new file mode 100644 index 0000000..bcd2ea5 --- /dev/null +++ b/api/tests/test_secret_scan.py @@ -0,0 +1,106 @@ +"""secret-scan + env-names: findings carry label/location/hash, NEVER a value or fragment.""" + +from __future__ import annotations + +import os +import subprocess +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[2] +SCRIPTS = ROOT / "scripts" +# Synthetic, random-looking, never real. FAKE_LB is in the fake lockbox; TWI matches a shape. +FAKE_LB = "k7Xv2QpLm9RtZw4HnB8dYc3S" +TWI = "9f3a7c1e5b2d48806a1f4e7d2c9b0835" + + +def run(script, *args, env=None): + e = {**os.environ, **(env or {})} + r = subprocess.run([sys.executable, str(SCRIPTS / script), *args], capture_output=True, text=True, env=e) + return r.returncode, r.stdout + r.stderr + + +def no_fragment(out: str, value: str, n: int = 6): + assert value not in out + for i in range(len(value) - n + 1): + assert value[i:i + n] not in out, f"fragment of the value leaked at {i}" + + +def seeded(tmp_path): + lb = tmp_path / "lockbox.md" + lb.write_text(f"FAKE_VENDOR_API_KEY={FAKE_LB}\nNOTE: nothing here\n") + repo = tmp_path / "repo" + repo.mkdir() + g = lambda *a: subprocess.run(["git", "-C", str(repo), *a], check=True, capture_output=True) # noqa: E731 + g("init", "-q", "-b", "main") + g("config", "user.email", "t@t") + g("config", "user.name", "t") + (repo / "app.py").write_text(f'KEY = "{FAKE_LB}"\nTWILIO_AUTH_TOKEN = "{TWI}"\n') + g("add", "-A") + g("commit", "-qm", "add secrets") + (repo / "app.py").write_text("KEY = None\n") # removed from HEAD, still in history + g("commit", "-qam", "remove") + return lb, repo + + +def test_tree_scan_labels_locations_no_value(tmp_path): + lb, repo = seeded(tmp_path) + (repo / "live.py").write_text(f'x = "{FAKE_LB}"\n') + rc, out = run("secret_scan.py", str(repo / "live.py"), env={"SECRET_SCAN_LOCKBOX_PATHS": str(lb)}) + assert rc == 1 + assert "live.py:1" in out and "lockbox:FAKE_VENDOR_API_KEY" in out + no_fragment(out, FAKE_LB) + + +def test_history_finds_removed_secret_with_commit(tmp_path): + lb, repo = seeded(tmp_path) + rc, out = run("secret_scan.py", str(repo), "--history", env={"SECRET_SCAN_LOCKBOX_PATHS": str(lb)}) + assert rc == 1 + assert "app.py:1" in out and "commit=" in out and "lockbox:FAKE_VENDOR_API_KEY" in out + assert "app.py:2" in out and "32-hex secret assignment" in out + no_fragment(out, FAKE_LB) + no_fragment(out, TWI) + + +def test_repo_flag_clones_scans_and_cleans_up(tmp_path): + lb, repo = seeded(tmp_path) + cache = tmp_path / "home" + (cache / ".cache").mkdir(parents=True) + rc, out = run("secret_scan.py", "--repo", str(repo), + env={"SECRET_SCAN_LOCKBOX_PATHS": str(lb), "HOME": str(cache)}) + assert rc == 1 and "app.py:1" in out + no_fragment(out, FAKE_LB) + assert not [p for p in (cache / ".cache").iterdir() if p.name.startswith("secret-scan-")] + + +def test_clean_tree_and_bad_input(tmp_path): + (tmp_path / "ok.txt").write_text("hello world\n") + rc, out = run("secret_scan.py", str(tmp_path / "ok.txt"), "--no-lockbox") + assert rc == 0 and "0 finding(s)" in out + rc, out = run("secret_scan.py", str(tmp_path), "--history", "--no-lockbox") + assert rc == 2 and "needs a git repo" in out + + +def test_env_names_never_prints_values(tmp_path): + a = tmp_path / "a.env" + b = tmp_path / "b.env" + a.write_text(f"# c\nexport DB_PASSWORD={FAKE_LB}\nTOKEN=\"{TWI}\"\nEMPTY=\nONLY_A=1\n") + b.write_text(f"DB_PASSWORD={FAKE_LB}\nTOKEN=different-value-here\nONLY_B=2\n") + rc, out = run("env_names.py", str(a), "--hash") + assert rc == 0 and "DB_PASSWORD" in out and "set" in out and "empty" in out and "len=24" in out + assert "sha256:" in out + no_fragment(out, FAKE_LB) + no_fragment(out, TWI, 7) + rc, out = run("env_names.py", str(a), "--compare", str(b)) + assert "DB_PASSWORD" in out and "SAME" in out and "DIFFERENT" in out + assert "only-in-A" in out and "only-in-B" in out + no_fragment(out, FAKE_LB) + rc, out = run("env_names.py", str(tmp_path / "missing.env")) + assert rc == 2 + + +def test_shapes_are_the_guards_shapes(): + sys.path.insert(0, str(SCRIPTS)) + import secret_scan as sc + import secret_shapes as ss + assert sc.ss is ss # one source of shapes: a new guard shape is automatically a scan shape diff --git a/scripts/env_names.py b/scripts/env_names.py new file mode 100644 index 0000000..f94a7c8 --- /dev/null +++ b/scripts/env_names.py @@ -0,0 +1,153 @@ +#!/usr/bin/env python3 +"""env-names: list environment variable NAMES only (Boss ruling 10-01, house rule 10). + + env-names [--host H] [--sudo] [--hash] + env-names A --compare B [--host H] [--host2 H2] + +Prints NAME, set|empty, and value LENGTH. Never a value or fragment. --hash adds sha256[:8] +(compare two places for equality; a hash of a weak value can be guessed, so use it for +real secrets only). --compare prints SAME / DIFFERENT / only-in-A / only-in-B per name +(equality by full-value hash, nothing else shown). Values live in memory only. +Targets: an existing file (dotenv style) | a docker container name | a systemd unit +(Environment= + EnvironmentFile=; --sudo to read root-only files). --host runs the docker / +systemctl / file read over ssh (alias from ~/.ssh/config). +""" +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import shlex +import subprocess +import sys + +KV = ("=",) + + +def run(cmd: list[str], host: str | None, sudo: bool = False) -> tuple[int, str]: + if sudo: + cmd = ["sudo", "-n", *cmd] + if host: + cmd = ["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=10", host, shlex.join(cmd)] + r = subprocess.run(cmd, capture_output=True, text=True, errors="ignore", timeout=60) + return r.returncode, r.stdout + + +def parse_dotenv(text: str) -> dict[str, str]: + out: dict[str, str] = {} + for raw in text.splitlines(): + line = raw.strip() + if not line or line.startswith("#") or "=" not in line: + continue + if line.startswith("export "): + line = line[7:].lstrip() + k, v = line.split("=", 1) + k = k.strip() + v = v.strip() + if len(v) >= 2 and v[0] == v[-1] and v[0] in "\"'": + v = v[1:-1] + if k.replace("_", "").isalnum() and not k[0].isdigit(): + out[k] = v + return out + + +def from_file(path: str, host: str | None, sudo: bool) -> dict[str, str] | None: + if host or sudo: + rc, out = run(["cat", path], host, sudo) + return parse_dotenv(out) if rc == 0 else None + try: + with open(path, errors="ignore") as fh: + return parse_dotenv(fh.read()) + except OSError: + return None + + +def from_docker(name: str, host: str | None, sudo: bool) -> dict[str, str] | None: + rc, out = run(["docker", "inspect", "-f", "{{json .Config.Env}}", name], host, sudo) + if rc != 0 or not out.strip(): + return None + try: + items = json.loads(out) + except ValueError: + return None + return {k: v for k, _, v in (i.partition("=") for i in (items or []))} + + +def from_systemd(unit: str, host: str | None, sudo: bool) -> dict[str, str] | None: + rc, out = run(["systemctl", "show", unit, "-p", "Environment", "-p", "EnvironmentFiles"], host) + if rc != 0 or "LoadState=not-found" in out: + return None + env: dict[str, str] = {} + files: list[str] = [] + for line in out.splitlines(): + if line.startswith("Environment="): + for tok in shlex.split(line[len("Environment="):]): + k, _, v = tok.partition("=") + env[k] = v + elif line.startswith("EnvironmentFiles="): + f = line[len("EnvironmentFiles="):].split(" (")[0].strip().lstrip("-") + if f: + files.append(f) + for f in files: # later files override earlier, like systemd + d = from_file(f, host, sudo) + if d is None: + print(f"# note: EnvironmentFile {f} unreadable (try --sudo)", file=sys.stderr) + else: + env.update(d) + return env + + +def load(target: str, host: str | None, sudo: bool) -> dict[str, str] | None: + if (not host and os.path.isfile(target)) or target.startswith(("/", "./", "~")): + return from_file(os.path.expanduser(target), host, sudo) + if target.endswith((".service", ".timer", ".socket")): + return from_systemd(target, host, sudo) + return from_docker(target, host, sudo) or from_systemd(target, host, sudo) + + +def sh(v: str) -> str: + return hashlib.sha256(v.encode()).hexdigest() + + +def main(argv=None) -> int: + ap = argparse.ArgumentParser(prog="env-names", description="env var NAMES only") + ap.add_argument("target") + ap.add_argument("--host") + ap.add_argument("--host2", help="ssh host for the --compare target") + ap.add_argument("--sudo", action="store_true") + ap.add_argument("--hash", action="store_true", help="add sha256[:8] per variable") + ap.add_argument("--compare", metavar="TARGET2") + a = ap.parse_args(argv) + env = load(a.target, a.host, a.sudo) + if env is None: + print(f"error: could not read {a.target!r} (file, docker container or systemd unit)") + return 2 + if a.compare: + env2 = load(a.compare, a.host2 or a.host, a.sudo) + if env2 is None: + print(f"error: could not read {a.compare!r}") + return 2 + for k in sorted(set(env) | set(env2)): + if k not in env2: + print(f"{k:<40} only-in-A") + elif k not in env: + print(f"{k:<40} only-in-B") + else: + print(f"{k:<40} {'SAME' if sh(env[k]) == sh(env2[k]) else 'DIFFERENT'}" + f" (len {len(env[k])} vs {len(env2[k])})") + return 0 + for k in sorted(env): + v = env[k] + extra = f" sha256:{sh(v)[:8]}" if a.hash and v else "" + print(f"{k:<40} {'set ' if v else 'empty'} len={len(v)}{extra}") + print(f"# {len(env)} variable(s); values never printed") + return 0 + + +if __name__ == "__main__": + try: + sys.exit(main()) + except Exception as e: # never a traceback + print(f"error: {type(e).__name__}") + sys.exit(2) diff --git a/scripts/install_secret_tools.sh b/scripts/install_secret_tools.sh new file mode 100755 index 0000000..8dc7204 --- /dev/null +++ b/scripts/install_secret_tools.sh @@ -0,0 +1,14 @@ +#!/usr/bin/env bash +# Install the shared hash-only secret tools on THIS machine (Windy 0): secret-scan + env-names. +# Source of truth is this repo (scripts/); re-run after a pull to update. +set -euo pipefail +here=$(cd "$(dirname "$0")" && pwd) +dest="$HOME/.local/share/secret-tools" +mkdir -p "$dest" "$HOME/.local/bin" +cp "$here/secret_shapes.py" "$here/secret_scan.py" "$here/env_names.py" "$dest/" +for pair in "secret-scan:secret_scan.py" "env-names:env_names.py"; do + n=${pair%%:*}; f=${pair##*:} + printf '#!/usr/bin/env bash\nexec python3 "%s/%s" "$@"\n' "$dest" "$f" > "$HOME/.local/bin/$n" + chmod 755 "$HOME/.local/bin/$n" +done +echo "installed secret-scan and env-names (shapes from secret_shapes.py, same as secret-guard)" diff --git a/scripts/secret_scan.py b/scripts/secret_scan.py new file mode 100644 index 0000000..d611d6f --- /dev/null +++ b/scripts/secret_scan.py @@ -0,0 +1,210 @@ +#!/usr/bin/env python3 +"""secret-scan: hash-only secret finder. Boss ruling 10-01 after three lanes printed secrets +into their own transcripts while hunting secrets (house rule 10). + + secret-scan [--history] [--repo ] [--no-lockbox] + +Reports `file:line` (and the commit with --history), WHICH lockbox entry matched (the KEY +NAME only) or which secret SHAPE matched (twilio, zai, aws, ...), plus a sha256[:8] of the +token for allow-listing. It NEVER prints, logs or writes a value or any fragment of one +(no context line, no masking). The lockbox is loaded in memory only. stdout only. +Exit 0 = clean, 1 = findings, 2 = usage/error. +""" +from __future__ import annotations + +import argparse +import hashlib +import os +import re +import shutil +import subprocess +import sys +import tempfile +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +import secret_shapes as ss # noqa: E402 (the SAME shapes as secret-guard) + +HOME = Path.home() +LOCKBOX_PATHS = [p for p in os.environ.get("SECRET_SCAN_LOCKBOX_PATHS", "").split(":") if p] or [ + str(HOME / "kit-army-config" / "secrets"), str(HOME / "kit-army-config" / "ACCESS_LOCKBOX.md")] +# Extra shapes that are scan-only (not in the blocking guard): label, regex. +EXTRA = [("zai key", re.compile(r"(?*`|-]*([A-Za-z][A-Za-z0-9_]{2,60})\s*[=:|]") + + +def h16(t: str) -> str: + return hashlib.sha256(t.encode()).hexdigest()[:16] + + +def cands(line: str): + """Token candidates: runs >=16 chars with a digit and a letter; git shas/uuids excluded.""" + for chunk in re.split(r"[^A-Za-z0-9_\-.]+", line): + parts = [chunk, *chunk.split(".")] if "." in chunk else [chunk] + for p in parts: + for m in RUN.finditer(p): + t = m.group(0) + if not (re.search(r"\d", t) and re.search(r"[A-Za-z]", t)): + continue + if re.fullmatch(r"[0-9a-fA-F]{40}|[0-9a-fA-F]{64}", t) or UUID.match(t.lower()): + continue + yield t + + +def load_lockbox() -> dict[str, set[str]]: + """{hash16: {key-name labels}}; values never leave this dict.""" + out: dict[str, set[str]] = {} + files: list[str] = [] + for p in LOCKBOX_PATHS: + if os.path.isdir(p): + for root, _d, fs in os.walk(p): + files += [os.path.join(root, f) for f in fs] + elif os.path.isfile(p): + files.append(p) + for f in files: + try: + with open(f, errors="ignore") as fh: + for line in fh: + m = NAME.match(line) + label = m.group(1) if m else "?" + for t in cands(line): + out.setdefault(h16(t), set()).add(label) + except OSError: + continue + return out + + +def scan_line(line: str, lockbox: dict[str, set[str]]) -> list[tuple[str, str]]: + """[(label, hash8)]: `shape:` and/or `lockbox:`. No value escapes.""" + res: list[tuple[str, str]] = [] + for kind, h in ss.find(line): + res.append((f"shape:{kind}", h)) + for kind, rx in EXTRA: + for m in rx.finditer(line): + res.append((f"shape:{kind}", ss.h8(m.group(0)))) + for t in cands(line): + k = h16(t) + if k in lockbox: + names = sorted(n for n in lockbox[k]) + res.append(("lockbox:" + ",".join(names)[:70], k[:8])) + return sorted(set(res)) + + +def is_text(path: Path) -> bool: + try: + with open(path, "rb") as fh: + return b"\0" not in fh.read(4096) + except OSError: + return False + + +def scan_tree(root: Path, lockbox): + files = [root] if root.is_file() else [ + Path(dp) / f for dp, dn, fn in os.walk(root) for f in fn + if not set(Path(dp).relative_to(root).parts) & SKIP_DIRS] + for p in sorted(files): + try: + if p.stat().st_size > MAX_BYTES or not is_text(p): + continue + with open(p, errors="ignore") as fh: + for n, line in enumerate(fh, 1): + for label, h in scan_line(line, lockbox): + yield (str(p), n, None, label, h) + except OSError: + continue + + +def git(repo: str, *a: str) -> subprocess.Popen: + return subprocess.Popen(["git", "--git-dir", repo, *a], stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, text=True, errors="ignore") + + +def scan_history(gitdir: str, lockbox): + """Every ADDED line on every ref (incl. PR refs). Oldest commit per (hash, file, line).""" + seen: dict[tuple, str] = {} + p = git(gitdir, "log", "--all", "-p", "-U0", "--no-color", "--format=@@C %h", "-a") + commit = path = None + ln = 0 + for row in p.stdout: # type: ignore[union-attr] + if row.startswith("@@C "): + commit = row[4:].strip() + elif row.startswith("+++ "): + path = row[6:].strip() if row.startswith("+++ b/") else None + elif row.startswith("@@ "): + m = re.search(r"\+(\d+)", row) + ln = int(m.group(1)) - 1 if m else 0 + elif row.startswith("+") and path: + ln += 1 + for label, h in scan_line(row[1:], lockbox): + seen[(label, h, path, ln)] = commit or "?" + p.wait() + for (label, h, path, ln), c in sorted(seen.items(), key=lambda x: (x[0][2], x[0][3])): + yield (path, ln, c, label, h) + + +def resolve_gitdir(p: Path) -> str | None: + for cand in (p / ".git", p): + if (cand / "HEAD").exists() and ((cand / "objects").exists()): + return str(cand) + return None + + +def main(argv=None) -> int: + ap = argparse.ArgumentParser(prog="secret-scan", description=__doc__.split("\n\n")[1] if __doc__ else "") + ap.add_argument("path", nargs="?", help="file, directory, or git repo (with --history)") + ap.add_argument("--history", action="store_true", help="scan every added line in all git history") + ap.add_argument("--repo", help="git URL or path to scan (mirror-cloned to a temp dir, then deleted)") + ap.add_argument("--no-lockbox", action="store_true", help="shapes only") + a = ap.parse_args(argv) + if not (a.path or a.repo): + ap.print_usage() + return 2 + lockbox = {} if a.no_lockbox else load_lockbox() + print(f"# secret-scan: {len(lockbox)} lockbox tokens in memory, values never printed", flush=True) + tmp = None + findings = 0 + try: + if a.repo: + tmp = tempfile.mkdtemp(prefix="secret-scan-", dir=str(HOME / ".cache") if (HOME / ".cache").is_dir() else None) + os.chmod(tmp, 0o700) + target = os.path.join(tmp, "r.git") + rc = subprocess.run(["git", "clone", "-q", "--mirror", a.repo, target], + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL).returncode + if rc != 0: + print("error: clone failed (details withheld: URLs can carry tokens)") + return 2 + a.history = True + gitdir = target + elif a.history: + gitdir = resolve_gitdir(Path(a.path)) + if not gitdir: + print("error: --history needs a git repo path") + return 2 + if a.history: + for path, ln, c, label, h in scan_history(gitdir, lockbox): + findings += 1 + print(f"{path}:{ln} commit={c} {label} #{h}") + else: + for path, ln, _c, label, h in scan_tree(Path(a.path), lockbox): + findings += 1 + print(f"{path}:{ln} {label} #{h}") + finally: + if tmp: + shutil.rmtree(tmp, ignore_errors=True) + print(f"# {findings} finding(s)") + return 1 if findings else 0 + + +if __name__ == "__main__": + try: + sys.exit(main()) + except KeyboardInterrupt: + sys.exit(130) + except Exception as e: # never a traceback: it could carry data + print(f"error: {type(e).__name__}") + sys.exit(2)