334 lines
15 KiB
Python
334 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""security_scan_gate.py — a dependency and secret scanner on every push, WARN-FIRST (R-986).
|
|
|
|
Usage: python3 scripts/security_scan_gate.py <repo-root>
|
|
|
|
Shared by the four felhom repos (it lives HERE, in felhom.eu/scripts/, and is never copied — the
|
|
reuse_refs_check.py rule). Each repo's gate runner calls it with its own root, in the FAST set, so the
|
|
pre-push hook runs it too.
|
|
|
|
WHAT IT RUNS (pinned versions; `documentation/audits/security-review-2026-10-12/SCANNERS.md`):
|
|
|
|
govulncheck v1.8.0 on every Go module of the repo (each directory with a tracked go.mod), source mode.
|
|
Only CALLED vulnerabilities are listed (a trace that reaches a function); an imported-
|
|
or required-only one is counted, not listed — the brief's „called vulnerabilities only".
|
|
gitleaks 8.24.3 on the commits being pushed: `@{upstream}..HEAD` in a clone with an upstream, the newest
|
|
commit when that range is empty, and the pushed commit in CI (a depth-1 clone). Values are
|
|
redacted (`--redact=100`): this gate never prints a secret.
|
|
|
|
WARN-FIRST. A hit is printed and listed; the exit code stays 0. A tool that is missing, a toolchain that is
|
|
absent (CI's runner has python3 and git, no Go), or a network that cannot reach the vulnerability database
|
|
is printed as NOT CHECKED — never a pass in silence, never a refusal. Turning a hit into a refusal is a
|
|
later step with its own row. EXIT: 0 always, except a usage error (2).
|
|
|
|
TOUCHES THE NETWORK, and is still in the fast set — deliberately, and only because it cannot refuse:
|
|
govulncheck reads vuln.go.dev. Offline, it says NOT CHECKED and the push goes on (the fast set's promise is
|
|
„a push never fails because the network blinked", and that promise holds). Each tool has a time limit.
|
|
|
|
ACCEPTED HITS live in `<repo-root>/.security-accepted`, one per line, with a reason:
|
|
|
|
govulncheck GO-2026-1234 hub # why it is accepted
|
|
gitleaks generic-api-key@3f2a9c01d4e5 hub/internal/x_test.go # why
|
|
|
|
A govulncheck line names the OSV id and the module directory („." for the root). A gitleaks line names the
|
|
rule id, `@`, and the first 12 hex of the sha256 of the hit's whole LINE (read from git, never printed), then
|
|
the file — so one accepted test constant does not accept the next secret written into the same file. The
|
|
WARN line prints that `rule@hash` to copy. A line with no `# reason` is itself a WARN. An accepted hit is
|
|
listed as ACCEPTED.
|
|
|
|
CI (`FELHOM_SCAN_FETCH=1`, set in each repo's gates.yml): when gitleaks is not on this machine, the pinned
|
|
release tarball is fetched from GitHub and both the tarball and the binary are checked against the sha256
|
|
values below before it runs; a mismatch is NOT CHECKED, never a run.
|
|
|
|
SEAMS (for the decoy suite only): `FELHOM_SCAN_GOVULNCHECK_JSON=<file>` reads govulncheck's JSON from a file
|
|
instead of running it (each module gets that file); `FELHOM_SCAN_RANGE=<git log range>` pins the commits.
|
|
"""
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tarfile
|
|
import tempfile
|
|
import urllib.request
|
|
|
|
for _stream in (sys.stdout, sys.stderr):
|
|
try:
|
|
_stream.reconfigure(encoding="utf-8", errors="replace")
|
|
except (AttributeError, ValueError, OSError): # pragma: no cover
|
|
pass
|
|
|
|
GOVULNCHECK_VERSION = "v1.8.0"
|
|
GITLEAKS_VERSION = "8.24.3"
|
|
GITLEAKS_URL = ("https://github.com/gitleaks/gitleaks/releases/download/v%s/gitleaks_%s_linux_x64.tar.gz"
|
|
% (GITLEAKS_VERSION, GITLEAKS_VERSION))
|
|
# The release's own checksums file (gitleaks_8.24.3_checksums.txt) and the binary inside, read 2026-10-19;
|
|
# the binary's sha is the one SCANNERS.md recorded for the 2026-10-12 review.
|
|
GITLEAKS_TGZ_SHA256 = "9991e0b2903da4c8f6122b5c3186448b927a5da4deef1fe45271c3793f4ee29c"
|
|
GITLEAKS_BIN_SHA256 = "e18325d568268b6efb6cd0cc721f8bea5667f4cd1c8d03618ab61987f41269ad"
|
|
# FELHOM_SECURITY_TOOLS_DIR moves the default (the decoy suite points it at an empty dir to make a tool „missing").
|
|
TOOLS_DIR = os.environ.get("FELHOM_SECURITY_TOOLS_DIR", "/mnt/5_hdd/felhom.eu/tools/security/bin")
|
|
TIMEOUT = 180
|
|
ACCEPTED = ".security-accepted"
|
|
|
|
|
|
def find_tool(name):
|
|
for d in (os.environ.get("FELHOM_SECURITY_BIN", ""), TOOLS_DIR):
|
|
p = os.path.join(d, name) if d else ""
|
|
if p and os.path.isfile(p) and os.access(p, os.X_OK):
|
|
return p
|
|
return shutil.which(name)
|
|
|
|
|
|
def sha256(path):
|
|
h = hashlib.sha256()
|
|
with open(path, "rb") as f:
|
|
for chunk in iter(lambda: f.read(1 << 20), b""):
|
|
h.update(chunk)
|
|
return h.hexdigest()
|
|
|
|
|
|
def fetch_gitleaks():
|
|
"""CI only: the pinned release, both checksums verified. Returns (path, note)."""
|
|
cache = os.path.join(os.environ.get("XDG_CACHE_HOME") or os.path.join(os.path.expanduser("~"), ".cache"),
|
|
"felhom-scan", GITLEAKS_VERSION)
|
|
binp = os.path.join(cache, "gitleaks")
|
|
if os.path.isfile(binp) and sha256(binp) == GITLEAKS_BIN_SHA256:
|
|
return binp, "cached pinned release"
|
|
try:
|
|
os.makedirs(cache, exist_ok=True)
|
|
req = urllib.request.Request(GITLEAKS_URL, headers={"User-Agent": "felhom-security-scan/1.0"})
|
|
data = urllib.request.urlopen(req, timeout=60).read()
|
|
except Exception as e: # network, DNS, TLS
|
|
return None, "could not fetch the pinned release (%s)" % str(e)[:120]
|
|
if hashlib.sha256(data).hexdigest() != GITLEAKS_TGZ_SHA256:
|
|
return None, "the fetched tarball's sha256 is not the pinned one — not run"
|
|
with tarfile.open(fileobj=io.BytesIO(data), mode="r:gz") as t:
|
|
m = t.getmember("gitleaks")
|
|
with open(binp + ".new", "wb") as out:
|
|
out.write(t.extractfile(m).read())
|
|
if sha256(binp + ".new") != GITLEAKS_BIN_SHA256:
|
|
os.remove(binp + ".new")
|
|
return None, "the binary inside the tarball is not the pinned one — not run"
|
|
os.chmod(binp + ".new", 0o755)
|
|
os.replace(binp + ".new", binp)
|
|
return binp, "fetched pinned release, sha256 checked"
|
|
|
|
|
|
def read_accepted(root, warns):
|
|
out = {"govulncheck": {}, "gitleaks": {}}
|
|
p = os.path.join(root, ACCEPTED)
|
|
if not os.path.isfile(p):
|
|
return out
|
|
for n, line in enumerate(io.open(p, encoding="utf-8"), 1):
|
|
body, _, reason = line.partition("#")
|
|
f = body.split()
|
|
if not f:
|
|
continue
|
|
if len(f) != 3 or f[0] not in out:
|
|
warns.append("%s:%d: not „<tool> <id> <where> # reason\" — ignored" % (ACCEPTED, n))
|
|
continue
|
|
if not reason.strip():
|
|
warns.append("%s:%d: %s %s %s has no reason — accepted anyway, but say why" % (ACCEPTED, n, f[0], f[1], f[2]))
|
|
out[f[0]][(f[1], f[2])] = reason.strip() or "(no reason given)"
|
|
return out
|
|
|
|
|
|
def git(root, *args):
|
|
r = subprocess.run(["git", "-C", root] + list(args), capture_output=True, text=True, timeout=60)
|
|
return r.returncode, r.stdout.strip()
|
|
|
|
|
|
def go_modules(root):
|
|
rc, out = git(root, "ls-files", "*go.mod", "go.mod")
|
|
if rc != 0:
|
|
return []
|
|
mods = []
|
|
for f in out.splitlines():
|
|
parts = f.split("/")
|
|
# product modules only: not a test fixture, a vendored copy, or an audit's throwaway program
|
|
if f.endswith("go.mod") and not {"testdata", "vendor", "documentation", "audits"} & set(parts[:-1]):
|
|
mods.append(os.path.dirname(f) or ".")
|
|
return sorted(set(mods))
|
|
|
|
|
|
def parse_govulncheck(text):
|
|
"""govulncheck -format json is a stream of objects. Returns (called {id: (module, summary)}, other_ids, error)."""
|
|
dec = json.JSONDecoder()
|
|
i, osv, called, other = 0, {}, {}, set()
|
|
while i < len(text):
|
|
while i < len(text) and text[i].isspace():
|
|
i += 1
|
|
if i >= len(text):
|
|
break
|
|
try:
|
|
o, i = dec.raw_decode(text, i)
|
|
except ValueError:
|
|
return called, other, text[i:i + 200].strip().splitlines()[0] if text[i:].strip() else "unreadable output"
|
|
if "osv" in o:
|
|
osv[o["osv"].get("id", "?")] = o["osv"].get("summary", "")
|
|
if "finding" in o:
|
|
f = o["finding"]
|
|
tr = f.get("trace") or []
|
|
if tr and tr[0].get("function"):
|
|
called[f.get("osv", "?")] = tr[0].get("module", "")
|
|
else:
|
|
other.add(f.get("osv", "?"))
|
|
return {k: (m, osv.get(k, "")) for k, m in called.items()}, other - set(called), None
|
|
|
|
|
|
def run_govulncheck(root, accepted, lines):
|
|
new = acc = 0
|
|
mods = go_modules(root)
|
|
if not mods:
|
|
lines.append(" govulncheck: no Go module in this repo — nothing to scan")
|
|
return 0, 0
|
|
seam = os.environ.get("FELHOM_SCAN_GOVULNCHECK_JSON")
|
|
tool = None if seam else find_tool("govulncheck")
|
|
if not seam and not tool:
|
|
lines.append(" govulncheck: NOT CHECKED — govulncheck %s is not on this machine (tried $FELHOM_SECURITY_BIN, %s, PATH)"
|
|
% (GOVULNCHECK_VERSION, TOOLS_DIR))
|
|
return 0, 0
|
|
if not seam and not shutil.which("go"):
|
|
lines.append(" govulncheck: NOT CHECKED — no Go toolchain here (CI's runner has python3 and git only); the "
|
|
"pre-push hook runs it")
|
|
return 0, 0
|
|
for m in mods:
|
|
if seam:
|
|
text, note = io.open(seam, encoding="utf-8").read(), "(seam: %s)" % os.path.basename(seam)
|
|
else:
|
|
try:
|
|
r = subprocess.run([tool, "-format", "json", "./..."], cwd=os.path.join(root, m),
|
|
capture_output=True, text=True, timeout=TIMEOUT)
|
|
except subprocess.TimeoutExpired:
|
|
lines.append(" govulncheck %s: NOT CHECKED — over %ds" % (m, TIMEOUT))
|
|
continue
|
|
text, note = r.stdout + r.stderr, ""
|
|
called, other, err = parse_govulncheck(text)
|
|
if err and not called:
|
|
lines.append(" govulncheck %s: NOT CHECKED — %s %s" % (m, err[:160], note))
|
|
continue
|
|
lines.append(" govulncheck %s: %d called vulnerabilit%s (%d imported or required only, not listed) %s"
|
|
% (m, len(called), "y" if len(called) == 1 else "ies", len(other), note))
|
|
for vid in sorted(called):
|
|
mod, summary = called[vid]
|
|
why = accepted.get((vid, m))
|
|
if why:
|
|
acc += 1
|
|
lines.append(" ACCEPTED %s (%s) — %s" % (vid, mod, why))
|
|
else:
|
|
new += 1
|
|
lines.append(" WARN %s called, in %s — %s" % (vid, mod, summary[:100]))
|
|
return new, acc
|
|
|
|
|
|
def scan_range(root):
|
|
r = os.environ.get("FELHOM_SCAN_RANGE")
|
|
if r:
|
|
return r, "pinned by FELHOM_SCAN_RANGE"
|
|
if os.environ.get("GITHUB_ACTIONS") or os.environ.get("GITEA_ACTIONS"):
|
|
return "-n 1 HEAD", "the pushed commit; CI clones at depth 1"
|
|
rc, up = git(root, "rev-parse", "--verify", "-q", "@{upstream}")
|
|
if rc == 0 and up:
|
|
rc2, n = git(root, "rev-list", "--count", "@{upstream}..HEAD")
|
|
if rc2 == 0 and n.isdigit() and int(n) > 0:
|
|
return "@{upstream}..HEAD", "the %s commit(s) not yet on the upstream" % n
|
|
return "-n 1 HEAD", "the newest commit; nothing unpushed"
|
|
|
|
|
|
def line_hash(root, commit, path, lineno):
|
|
"""sha256[:12] of the hit's stripped line, read from the commit (or the working tree) — in memory only."""
|
|
text = None
|
|
if commit:
|
|
r = subprocess.run(["git", "-C", root, "show", "%s:%s" % (commit, path)], capture_output=True, timeout=60)
|
|
if r.returncode == 0:
|
|
text = r.stdout.decode("utf-8", "replace")
|
|
if text is None:
|
|
try:
|
|
text = io.open(os.path.join(root, path), encoding="utf-8", errors="replace").read()
|
|
except OSError:
|
|
return "unread"
|
|
ls = text.splitlines()
|
|
if not 0 < lineno <= len(ls):
|
|
return "unread"
|
|
return hashlib.sha256(ls[lineno - 1].strip().encode("utf-8")).hexdigest()[:12]
|
|
|
|
|
|
def run_gitleaks(root, accepted, lines):
|
|
tool = find_tool("gitleaks")
|
|
note = ""
|
|
if not tool and os.environ.get("FELHOM_SCAN_FETCH") == "1":
|
|
tool, note = fetch_gitleaks()
|
|
if not tool:
|
|
lines.append(" gitleaks: NOT CHECKED — gitleaks %s is not on this machine%s" % (GITLEAKS_VERSION, "; " + note if note else ""))
|
|
return 0, 0
|
|
for f in (".gitleaks.toml", ".gitleaksignore"):
|
|
if os.path.exists(os.path.join(root, f)):
|
|
lines.append(" gitleaks: WARN this repo carries %s — the scan below ignores it (its config and ignore list are "
|
|
"pinned by this gate, so a file in the repo cannot switch rules off)" % f)
|
|
rng, why = scan_range(root)
|
|
tmp = tempfile.mkdtemp(prefix="felhom-scan-")
|
|
try:
|
|
rep = os.path.join(tmp, "report.json")
|
|
empty_ignore = os.path.join(tmp, "ignore")
|
|
os.makedirs(empty_ignore)
|
|
cfg_env = dict(os.environ)
|
|
for k in ("GITLEAKS_CONFIG", "GITLEAKS_CONFIG_TOML"):
|
|
cfg_env.pop(k, None)
|
|
argv = [tool, "git", root, "--log-opts=" + rng, "--redact=100", "--no-banner", "--report-format", "json",
|
|
"--report-path", rep, "--exit-code", "0", "--gitleaks-ignore-path", empty_ignore]
|
|
# The default rules, never a .gitleaks.toml from the scanned tree: hand gitleaks its own default config.
|
|
dflt = os.path.join(tmp, "default.toml")
|
|
io.open(dflt, "w").write('[extend]\nuseDefault = true\n')
|
|
argv += ["--config", dflt]
|
|
try:
|
|
r = subprocess.run(argv, capture_output=True, text=True, timeout=TIMEOUT, env=cfg_env)
|
|
except subprocess.TimeoutExpired:
|
|
lines.append(" gitleaks: NOT CHECKED — over %ds" % TIMEOUT)
|
|
return 0, 0
|
|
if r.returncode != 0 or not os.path.isfile(rep):
|
|
lines.append(" gitleaks: NOT CHECKED — exit %d: %s" % (r.returncode, (r.stderr or r.stdout).strip()[-160:]))
|
|
return 0, 0
|
|
hits = json.load(io.open(rep, encoding="utf-8")) or []
|
|
finally:
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
lines.append(" gitleaks %s on %s (%s)%s: %d hit(s)" % (GITLEAKS_VERSION, rng, why, " [" + note + "]" if note else "", len(hits)))
|
|
new = acc = 0
|
|
for h in hits:
|
|
rule, path = h.get("RuleID", "?"), h.get("File", "?")
|
|
key = "%s@%s" % (rule, line_hash(root, h.get("Commit") or "", path, h.get("StartLine") or 0))
|
|
why = accepted.get((key, path))
|
|
where = "%s:%s, commit %s" % (path, h.get("StartLine", "?"), (h.get("Commit") or "")[:10])
|
|
if why:
|
|
acc += 1
|
|
lines.append(" ACCEPTED %s %s — %s" % (key, where, why))
|
|
else:
|
|
new += 1
|
|
lines.append(" WARN %s %s (value redacted)" % (key, where))
|
|
return new, acc
|
|
|
|
|
|
def main(argv):
|
|
if len(argv) != 2 or not os.path.isdir(argv[1]):
|
|
print(__doc__.split("\n\n")[1])
|
|
return 2
|
|
root = os.path.abspath(argv[1])
|
|
warns, lines = [], []
|
|
accepted = read_accepted(root, warns)
|
|
print("security-scan (R-986, WARN-FIRST — it lists, it never refuses): %s" % root)
|
|
n1, a1 = run_govulncheck(root, accepted["govulncheck"], lines)
|
|
n2, a2 = run_gitleaks(root, accepted["gitleaks"], lines)
|
|
for l in lines:
|
|
print(l)
|
|
for w in warns:
|
|
print(" WARN %s" % w)
|
|
print("security-scan: %d new hit(s), %d accepted — warn-first, the push goes on" % (n1 + n2, a1 + a2))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main(sys.argv))
|