RedFlag/.publication/surface_gate.py
Fimeg 5a7b122d4e projection: admit retired-path classification
The screenshot names remain bounded in reachable history and absent from the current public tree.

Source-Sha: c357423ec432febcfce49adb42b863ee1a309b99
Policy-Sha: c357423ec432febcfce49adb42b863ee1a309b99
Tree-Digest: 2c39e139c90830e419de6510b326b6de9be288d4f5121c94cfb921734f815d55
2026-09-10 09:21:32 -04:00

553 lines
23 KiB
Python

#!/usr/bin/env python3
"""Editorial authority over a public tree.
The existing `.publication` mechanism answers a transport question: was this
SHA produced by the right gates, on the right repository, under a policy that
authorizes projection to this destination. It is a good answer to that
question and this file does not replace it.
This file answers the question nobody was asking:
Does this tree belong outside at all?
Six gates, in the order a reviewer would actually apply them:
1. path-authority every recursive path is named by the exact manifest
2. new-surface what the public diff P1 -> P2 newly admits
3. file-class forbidden classes anywhere in the tree
4. content the whole candidate tree, not the changed lines
5. history subjects, bodies, and author identities
6. presentation what a stranger sees in the root listing
Gate 1 is the admission authority. Gate 4 supplies the disclosure inspection
that an incremental scanner structurally cannot provide. A patch scanner sees
a file the day it lands. It never sees it again. A tree that was clean when
every one of its commits was scanned can still be a tree that should not be
public, because publication is a property of the tree, not of the diffs that
built it.
The severity model is deliberately harsher than the incremental scanner's.
There, a home path or a LAN address is a WARN: one line in one patch, and a
human is reading the patch anyway. Here the same finding is a DENY, because
nobody reads a whole tree, and because the finding means the path is standing
in the public product right now, not that it passed through once.
Absence of known-secret content is not authorization to publish.
Exceptions are narrow and auditable: path, rule, reason, review date. A rule
turned off globally is not an exception, it is a retreat, so this file has no
syntax for one.
"""
import argparse
import fnmatch
import json
import os
import re
import subprocess
import sys
from datetime import date
SCHEMA_VERSION = 2
DENY = "deny"
REVIEW = "review"
NOTE = "note"
SEVERITY_ORDER = {DENY: 0, REVIEW: 1, NOTE: 2}
# Anything a forge will not render as source. Reviewed by eye, not by rule.
BINARY_HINT = re.compile(
r"\.(png|jpe?g|gif|webp|ico|pdf|zip|gz|xz|zst|bz2|tar|so|a|o|dll|dylib"
r"|exe|bin|wav|mp3|mp4|ogg|woff2?|ttf|otf|jar|whl|deb|rpm|img|iso)$",
re.IGNORECASE,
)
# Host suffixes a stranger can actually resolve. A submodule pointed anywhere
# else is both a leak and a repository that does not clone.
PUBLIC_FORGE_HOSTS = (
"forge.caseytunturi.com",
"codeberg.org",
"github.com",
"gitlab.com",
"git.sr.ht",
)
class Rule:
__slots__ = ("rule_id", "description", "pattern", "severity")
def __init__(self, rule_id, description, pattern, severity):
self.rule_id = rule_id
self.description = description
self.pattern = re.compile(pattern, re.IGNORECASE)
self.severity = severity
# Capabilities deny because publishing one hands control to whoever reads it
# and no later cleanup takes it back. Topology denies here — see the module
# docstring — because in a tree it is a standing disclosure, not a transit.
CONTENT_RULES = (
Rule("private-key", "private key material",
r"-----BEGIN (?:RSA|DSA|EC|OPENSSH|PGP) PRIVATE KEY", DENY),
# The value must be quoted. An unquoted run of letters after `authorization:`
# is a type name in every language that has types, and matching it made the
# gate cry wolf over `authorization: MutationAuthorization` on first run.
Rule("bearer-token", "embedded token or bearer credential",
r"(?:ghp|gho|ghs|ghu|github_pat)_[A-Za-z0-9_]{20,}"
r"|(?:authorization|private[-_]?token|api[-_]?key|client[-_]?secret)"
r"\s*[:=]\s*['\"][A-Za-z0-9._~+/=-]{16,}['\"]", DENY),
Rule("rfc1918", "private LAN address",
r"\b(?:10\.\d{1,3}\.\d{1,3}\.\d{1,3}"
r"|192\.168\.\d{1,3}\.\d{1,3}"
r"|172\.(?:1[6-9]|2\d|3[01])\.\d{1,3}\.\d{1,3})\b", DENY),
# Case-sensitive on purpose. Every rule here is otherwise IGNORECASE, and
# a case-blind `/Users/` matched the phrase "sessions/users/seats" in a
# comment about loginctl. A macOS home is capitalised; a POSIX path segment
# spelled `users` is not a home directory.
Rule("home-path", "developer home directory",
r"/home/[a-z][a-z0-9_-]*|(?-i:/Users/[A-Za-z])|(?-i:C:\\\\Users\\\\)", DENY),
Rule("internal-host", "internal hostname or forge",
r"\bwiuf|\barchdev\b|\bwiufph\b", DENY),
# The internal domain only. An author's own public contact address is not a
# leak — it is on the security page on purpose, and matching it made the
# gate refuse commits for being signed by the person who wrote them.
Rule("internal-email", "internal service identity or internal domain",
r"@wiuf\.net", DENY),
Rule("ai-attribution", "model attribution",
r"co-authored-by:\s*(?:claude|gpt|copilot|codex)"
r"|generated with \[?claude", REVIEW),
)
MESSAGE_RULES = CONTENT_RULES
def run(repo, args, ok=(0,)):
p = subprocess.run(["git", "-C", repo] + args,
capture_output=True, text=True, errors="replace")
if p.returncode not in ok:
raise RuntimeError(f"git {' '.join(args)}: {p.stderr.strip()}")
return p.stdout
class Finding:
__slots__ = ("gate", "severity", "rule", "path", "detail", "excerpt")
def __init__(self, gate, severity, rule, path, detail, excerpt=""):
self.gate = gate
self.severity = severity
self.rule = rule
self.path = path
self.detail = detail
self.excerpt = excerpt
def redact(text, match):
"""Keep the shape, drop the value. A report is itself a publishable object."""
s, e = match.span()
line = text[max(0, s - 40):e + 40].replace("\n", " ").strip()
hit = match.group(0)
keep = 2 if len(hit) > 6 else 1
masked = hit[:keep] + "*" * max(1, len(hit) - keep * 2) + (hit[-keep:] if len(hit) > 6 else "")
return line.replace(hit, masked)[:150]
class Manifest:
"""The exact authority for what may cross."""
def __init__(self, raw, path, repo, sha):
self.path = path
self.schema_version = raw.get("schema_version")
if self.schema_version != SCHEMA_VERSION:
raise ValueError(f"{path}: schema_version must be {SCHEMA_VERSION}")
if "public_roots" in raw or "public_root_files" in raw:
raise ValueError(
f"{path}: broad root authority is not valid in schema {SCHEMA_VERSION}")
self.repository = raw.get("repository", "?")
self.path_manifest = raw.get("path_manifest", "")
validate_policy_path(self.path_manifest, "path_manifest")
manifest_text = run(repo, ["show", f"{sha}:{self.path_manifest}"])
paths = manifest_text.splitlines()
if not paths:
raise ValueError(f"{self.path_manifest}: exact path manifest is empty")
for candidate in paths:
validate_policy_path(candidate, "admitted path")
if paths != sorted(paths):
raise ValueError(f"{self.path_manifest}: paths must be bytewise sorted")
if len(paths) != len(set(paths)):
raise ValueError(f"{self.path_manifest}: duplicate path")
self.allowed_paths = frozenset(paths)
history_only = list(raw.get("history_only", []))
for candidate in history_only:
validate_policy_path(candidate, "history_only path")
if history_only != sorted(history_only):
raise ValueError(f"{path}: history_only paths must be bytewise sorted")
if len(history_only) != len(set(history_only)):
raise ValueError(f"{path}: duplicate history_only path")
self.history_only_paths = frozenset(history_only)
overlap = self.allowed_paths & self.history_only_paths
if overlap:
raise ValueError(f"{path}: current and history_only paths overlap: {sorted(overlap)[0]}")
self.exclude = list(raw.get("exclude", []))
self.review_required = list(raw.get("review_required", []))
self.forbidden = list(raw.get("forbidden", []))
self.forbidden_classes = list(raw.get("forbidden_classes", []))
self.max_blob_bytes = int(raw.get("max_blob_bytes", 2 * 1024 * 1024))
self.exceptions = list(raw.get("exceptions", []))
def excluded(self, path):
return next((pat for pat in self.exclude if fnmatch.fnmatch(path, pat)), None)
def needs_review(self, path):
return next((pat for pat in self.review_required if fnmatch.fnmatch(path, pat)), None)
def excepted(self, path, rule_id):
"""An exception names one path and one rule, and says why, and when."""
for exc in self.exceptions:
if exc.get("rule") != rule_id:
continue
if not fnmatch.fnmatch(path, exc.get("path", "")):
continue
if not exc.get("reason") or not exc.get("reviewed"):
continue
return exc
return None
def validate_policy_path(path, label):
if not isinstance(path, str) or not path:
raise ValueError(f"{label}: non-empty string required")
if (path.startswith("/") or "\\" in path or "\0" in path or "\n" in path
or "\r" in path or "\t" in path):
raise ValueError(f"{label}: unsafe path {path!r}")
if path != os.path.normpath(path) or path.startswith("../") or path == "..":
raise ValueError(f"{label}: path must be normalized and repository-relative: {path!r}")
def tree_entries(repo, sha):
"""(mode, type, oid, path) for every entry, recursively."""
out = run(repo, ["ls-tree", "-r", "-l", "-z", sha])
entries = []
for record in out.split("\0"):
if not record:
continue
meta, path = record.split("\t", 1)
parts = meta.split()
mode, otype, oid = parts[0], parts[1], parts[2]
size = parts[3] if len(parts) > 3 else "-"
entries.append((mode, otype, oid, size, path))
return entries
def top_level(repo, sha):
return [l for l in run(repo, ["ls-tree", "--name-only", sha]).splitlines() if l.strip()]
# ---------------------------------------------------------------- gate 1
def gate_path_authority(repo, sha, manifest, findings):
if run(repo, ["rev-parse", "--is-shallow-repository"]).strip() != "false":
findings.append(Finding(
"path-authority", DENY, "shallow-history", "-",
"complete history is required to establish path authority"))
return
entries = tree_entries(repo, sha)
present = {entry[4] for entry in entries}
origins = {path: sha for path in present}
commits = run(repo, ["rev-list", sha]).splitlines()
for commit in commits[1:]:
for _mode, _otype, _oid, _size, path in tree_entries(repo, commit):
origins.setdefault(path, commit)
for path, origin in sorted(origins.items()):
location = "candidate" if origin == sha else f"reachable commit {origin[:12]}"
root = path.split("/", 1)[0]
if root in manifest.forbidden:
findings.append(Finding(
"path-authority", DENY, "forbidden-root", path,
"top-level path is forbidden by the public-surface manifest"))
excluded_by = manifest.excluded(path)
if excluded_by:
findings.append(Finding(
"path-authority", DENY, "excluded-path", path,
f"{location} contains a path excluded by {excluded_by!r}"))
admitted = manifest.allowed_paths if origin == sha else (
manifest.allowed_paths | manifest.history_only_paths)
if path not in admitted:
findings.append(Finding(
"path-authority", DENY, "unlisted-path", path,
f"no exact manifest entry admits this path in {location}"))
review_pattern = manifest.needs_review(path)
if review_pattern and path in present:
findings.append(Finding(
"path-authority", REVIEW, "review-path", path,
f"manifest preserves human review under {review_pattern!r}"))
for path in sorted(manifest.allowed_paths - present):
findings.append(Finding(
"path-authority", DENY, "missing-path", path,
"exact manifest entry is absent from the candidate tree"))
for path in sorted(manifest.history_only_paths & present):
findings.append(Finding(
"path-authority", DENY, "history-only-path", path,
"path is classified for reachable history but reappears in the candidate tree"))
findings.append(Finding(
"path-authority", NOTE, "exact-census", "-",
f"{len(present)} candidate paths checked against "
f"{len(manifest.allowed_paths)} exact manifest entries; "
f"{len(origins)} distinct paths checked across {len(commits)} reachable commits"))
# ---------------------------------------------------------------- gate 2
def gate_new_surface(repo, sha, previous, manifest, findings):
if not previous:
findings.append(Finding(
"new-surface", NOTE, "no-baseline", "-",
"no previous public SHA given; the diff gate did not run"))
return
out = run(repo, ["diff", "--name-status", "--diff-filter=ACR", previous, sha])
added = []
for line in out.splitlines():
parts = line.split("\t")
if len(parts) >= 2:
added.append(parts[-1])
for path in added:
severity = NOTE if path in manifest.allowed_paths else DENY
findings.append(Finding(
"new-surface", severity, "new-path", path,
"new candidate path has exact manifest authority" if severity == NOTE
else "new candidate path has no exact manifest authority"))
if os.path.basename(path).startswith("."):
findings.append(Finding(
"new-surface", REVIEW, "new-dotfile", path,
"new dotfile or dotdirectory entering the public tree"))
# Symlinks and submodules are surface changes disguised as files.
for mode, otype, _oid, _size, path in tree_entries(repo, sha):
if mode == "120000":
target = run(repo, ["show", f"{sha}:{path}"]).strip()
escapes = target.startswith("/") or ".." in target.split("/")
findings.append(Finding(
"new-surface", DENY if escapes else NOTE, "symlink", path,
"symlink target leaves the public tree" if escapes
else "symlink stays inside the public tree"))
if otype == "commit":
findings.append(Finding(
"new-surface", REVIEW, "submodule", path,
"submodule gitlink; its URL must resolve publicly"))
def gate_submodule_urls(repo, sha, findings):
try:
text = run(repo, ["show", f"{sha}:.gitmodules"])
except RuntimeError:
return
name = None
for line in text.splitlines():
line = line.strip()
if line.startswith("[submodule"):
name = line
if line.startswith("url"):
url = line.split("=", 1)[1].strip()
host = re.sub(r"^[a-z+]+://", "", url).split("/")[0].split("@")[-1].split(":")[0]
if not any(host == h or host.endswith("." + h) for h in PUBLIC_FORGE_HOSTS):
findings.append(Finding(
"new-surface", DENY, "submodule-private-url", ".gitmodules",
f"submodule url host is not publicly resolvable ({host}); "
"the public repository cannot clone and the host leaks",
excerpt=name or ""))
# ---------------------------------------------------------------- gate 3
def gate_file_class(repo, sha, manifest, findings):
for _mode, _otype, _oid, size, path in tree_entries(repo, sha):
for pat in manifest.forbidden_classes:
if fnmatch.fnmatch(path, pat) or fnmatch.fnmatch(os.path.basename(path), pat):
findings.append(Finding(
"file-class", DENY, "forbidden-class", path,
f"matches forbidden class {pat!r}"))
break
if size not in ("-", None) and size.isdigit() and int(size) > manifest.max_blob_bytes:
findings.append(Finding(
"file-class", REVIEW, "oversized-blob", path,
f"{int(size):,} bytes exceeds the reviewed maximum "
f"({manifest.max_blob_bytes:,})"))
# ---------------------------------------------------------------- gate 4
def gate_content(repo, sha, manifest, findings, limit_per_rule=40):
counts = {}
for _mode, otype, _oid, _size, path in tree_entries(repo, sha):
if otype != "blob" or BINARY_HINT.search(path):
continue
try:
text = run(repo, ["show", f"{sha}:{path}"])
except RuntimeError:
continue
if "\0" in text[:8000]:
continue
for rule in CONTENT_RULES:
m = rule.pattern.search(text)
if not m:
continue
exc = manifest.excepted(path, rule.rule_id)
if exc:
findings.append(Finding(
"content", NOTE, rule.rule_id, path,
f"excepted: {exc['reason']} (reviewed {exc['reviewed']})"))
continue
counts[rule.rule_id] = counts.get(rule.rule_id, 0) + 1
if counts[rule.rule_id] > limit_per_rule:
continue
findings.append(Finding(
"content", rule.severity, rule.rule_id, path,
rule.description, excerpt=redact(text, m)))
# ---------------------------------------------------------------- gate 5
def gate_history(repo, sha, manifest, findings, limit_per_rule=25):
sep = "\x1e"
out = run(repo, ["log", f"--format=%H{sep}%an <%ae>{sep}%s{sep}%b\x1d", sha])
counts = {}
total = 0
for record in out.split("\x1d"):
record = record.strip("\n")
if not record.strip():
continue
parts = record.split(sep)
if len(parts) < 4:
continue
h, ident, subject, body = parts[0], parts[1], parts[2], parts[3]
total += 1
blob = f"{ident}\n{subject}\n{body}"
for rule in MESSAGE_RULES:
m = rule.pattern.search(blob)
if not m:
continue
counts[rule.rule_id] = counts.get(rule.rule_id, 0) + 1
if counts[rule.rule_id] > limit_per_rule:
continue
findings.append(Finding(
"history", rule.severity, rule.rule_id, h[:12],
f"{rule.description} in commit metadata: {subject[:60]}",
excerpt=redact(blob, m)))
findings.append(Finding("history", NOTE, "reachable", "-",
f"{total} commits reachable from {sha[:12]}"))
# ---------------------------------------------------------------- gate 6
def gate_presentation(repo, sha, manifest, findings):
roots = top_level(repo, sha)
findings.append(Finding(
"presentation", NOTE, "root-listing", "-",
f"{len(roots)} top-level entries; every recursive path is exact-manifest checked"))
# ---------------------------------------------------------------- report
GATES = ("path-authority", "new-surface", "file-class", "content", "history", "presentation")
def inventory(repo, sha):
"""Deterministic object inventory for one immutable candidate."""
return "".join(
f"{mode} {oid} {size} {path}\n"
for mode, _otype, oid, size, path in tree_entries(repo, sha)
)
def report(findings, repo, sha, previous, manifest):
lines = []
w = lines.append
w(f"# Public-surface audit — {manifest.repository}")
w("")
w(f"- candidate SHA: `{sha[:12]}`")
w(f"- previous public SHA: `{previous[:12] if previous else '(none given)'}`")
w(f"- manifest: `{os.path.basename(manifest.path)}`")
w(f"- generated: {date.today().isoformat()}")
w("")
deny = [f for f in findings if f.severity == DENY]
review = [f for f in findings if f.severity == REVIEW]
verdict = "DENY" if deny else ("REVIEW" if review else "PASS")
w(f"**Verdict: {verdict}** — {len(deny)} deny, {len(review)} review, "
f"{len([f for f in findings if f.severity == NOTE])} note.")
w("")
for gate in GATES:
rows = [f for f in findings if f.gate == gate]
if not rows:
continue
w(f"## {gate}")
w("")
rows.sort(key=lambda f: (SEVERITY_ORDER[f.severity], f.rule, f.path))
w("| sev | rule | path | detail |")
w("| --- | --- | --- | --- |")
for f in rows:
detail = f.detail
if f.excerpt:
detail += f" — `{f.excerpt}`"
detail = detail.replace("|", "\\|")
w(f"| {f.severity} | {f.rule} | `{f.path}` | {detail} |")
w("")
return "\n".join(lines)
def main():
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
ap.add_argument("--repo", required=True)
ap.add_argument("--sha", required=True)
ap.add_argument("--manifest", required=True)
ap.add_argument("--previous", default="")
ap.add_argument("--out", default="")
ap.add_argument("--inventory-out", default="")
ap.add_argument("--skip-history", action="store_true")
args = ap.parse_args()
sha = run(args.repo, ["rev-parse", args.sha]).strip()
previous = run(args.repo, ["rev-parse", args.previous]).strip() if args.previous else ""
repo_root = run(args.repo, ["rev-parse", "--show-toplevel"]).strip()
manifest_path = os.path.realpath(args.manifest)
manifest_rel = os.path.relpath(manifest_path, repo_root)
validate_policy_path(manifest_rel, "manifest")
with open(manifest_path) as fh:
manifest_text = fh.read()
committed_manifest = run(args.repo, ["show", f"{sha}:{manifest_rel}"])
if manifest_text != committed_manifest:
raise ValueError(
f"{manifest_rel}: working copy does not match candidate {sha[:12]}")
manifest = Manifest(json.loads(manifest_text), manifest_rel, args.repo, sha)
findings = []
gate_path_authority(args.repo, sha, manifest, findings)
gate_new_surface(args.repo, sha, previous, manifest, findings)
gate_submodule_urls(args.repo, sha, findings)
gate_file_class(args.repo, sha, manifest, findings)
gate_content(args.repo, sha, manifest, findings)
if not args.skip_history:
gate_history(args.repo, sha, manifest, findings)
gate_presentation(args.repo, sha, manifest, findings)
text = report(findings, args.repo, sha, previous, manifest)
if args.out:
with open(args.out, "w") as fh:
fh.write(text + "\n")
print(f"wrote {args.out}")
else:
print(text)
if args.inventory_out:
with open(args.inventory_out, "w") as fh:
fh.write(inventory(args.repo, sha))
print(f"wrote {args.inventory_out}")
return 1 if any(f.severity == DENY for f in findings) else 0
if __name__ == "__main__":
sys.exit(main())