secret-scan + env-names: shared hash-only tools (Boss 10-01: lanes printed secrets while hunting them)
secret-scan reports file:line/commit + lockbox KEY NAME or shape, never a value or fragment; env-names lists variable names/length/hash only. Same shapes as secret-guard. Seeded-fake tests. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
106
api/tests/test_secret_scan.py
Normal file
106
api/tests/test_secret_scan.py
Normal file
@@ -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
|
||||||
153
scripts/env_names.py
Normal file
153
scripts/env_names.py
Normal file
@@ -0,0 +1,153 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
"""env-names: list environment variable NAMES only (Boss ruling 10-01, house rule 10).
|
||||||
|
|
||||||
|
env-names <docker container | systemd unit | env file> [--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)
|
||||||
14
scripts/install_secret_tools.sh
Executable file
14
scripts/install_secret_tools.sh
Executable file
@@ -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)"
|
||||||
210
scripts/secret_scan.py
Normal file
210
scripts/secret_scan.py
Normal file
@@ -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 <path> [--history] [--repo <git url or path>] [--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"(?<![0-9a-f])[0-9a-f]{32}\.[A-Za-z0-9]{16}(?![A-Za-z0-9])"))]
|
||||||
|
SKIP_DIRS = {".git", "node_modules", "vendor", "third_party", "__pycache__", ".venv"}
|
||||||
|
MAX_BYTES = 5_000_000
|
||||||
|
|
||||||
|
RUN = re.compile(r"[A-Za-z0-9][A-Za-z0-9_\-]{15,199}")
|
||||||
|
UUID = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$")
|
||||||
|
NAME = re.compile(r"[\s>*`|-]*([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:<kind>` and/or `lockbox:<NAMES>`. 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)
|
||||||
Reference in New Issue
Block a user