87 lines
4.1 KiB
Python
87 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Add the generated-commit guard to an existing shared hook directory.
|
|
|
|
Never changes core.hooksPath, other hooks, or a pre-existing different guard.
|
|
The repo argument discovers the already-configured hook path; it is not a new
|
|
fleet routing or authority configuration.
|
|
"""
|
|
import argparse
|
|
import hashlib
|
|
import os
|
|
from pathlib import Path
|
|
import shlex
|
|
import subprocess
|
|
|
|
|
|
# Reviewed composition retains this guard and adds exact-scope secret scanning.
|
|
# CUST-WP-0080 / RMGR-WP-0021; unknown hooks still require owner integration.
|
|
SCANNED_HOOK_SHA256 = "d50cf763dc9814383c6886e30b6650bbbdcb49ddd78889082e3fda3626a2cbfd"
|
|
|
|
ADOPTED_HOOK_SHA256 = "1a8a8b2a54092311a355a3a51cf6a696bd05e146f2f4a25455ae5c30c502929d"
|
|
|
|
CHECKER = Path(__file__).with_name("check_generated_only_commit.py").resolve()
|
|
|
|
|
|
def install(repo, execute=False, enable_generation=False):
|
|
repo = Path(repo).resolve()
|
|
configured = subprocess.check_output(
|
|
["git", "-C", str(repo), "config", "--get", "core.hooksPath"], text=True).strip()
|
|
directory = Path(configured)
|
|
if not directory.is_absolute():
|
|
directory = repo / directory
|
|
if not directory.is_dir() or directory.is_symlink():
|
|
raise ValueError("existing hook directory must be real and present")
|
|
target = directory / "pre-commit"
|
|
content = ("#!/bin/sh\n# Custodian shared generated-commit guard; preserve other hooks.\n"
|
|
"set -eu\nexec python3 " + shlex.quote(str(CHECKER)) + " --staged\n")
|
|
if target.is_symlink():
|
|
raise ValueError("refuse redirected existing pre-commit hook")
|
|
if target.exists():
|
|
if (target.read_text() != content
|
|
and hashlib.sha256(target.read_bytes()).hexdigest() not in {SCANNED_HOOK_SHA256, ADOPTED_HOOK_SHA256}):
|
|
raise ValueError("existing pre-commit hook differs; owner integration required")
|
|
changed = False
|
|
if enable_generation and hashlib.sha256(target.read_bytes()).hexdigest() == SCANNED_HOOK_SHA256:
|
|
marker = "python3 " + shlex.quote(str(CHECKER)) + " --staged\n"
|
|
existing = target.read_text()
|
|
if existing.count(marker) != 1:
|
|
raise ValueError("reviewed hook checker path differs")
|
|
revised = existing.replace(marker, marker + "python3 " + shlex.quote(str(CHECKER.with_name("run_generated_work_records_hook.py"))) + "\n")
|
|
if hashlib.sha256(revised.encode()).hexdigest() != ADOPTED_HOOK_SHA256:
|
|
raise ValueError("adopted composition differs from reviewed hook")
|
|
if execute:
|
|
fd = os.open(target, os.O_WRONLY | os.O_TRUNC | os.O_NOFOLLOW)
|
|
with os.fdopen(fd, "w") as handle:
|
|
handle.write(revised)
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
changed = True
|
|
return {"repo": str(repo), "hook": str(target), "status": "installed", "changed": changed}
|
|
if execute:
|
|
fd = os.open(target, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o755)
|
|
with os.fdopen(fd, "w") as handle:
|
|
handle.write(content)
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
return {"repo": str(repo), "hook": str(target),
|
|
"status": "installed" if execute else "missing", "changed": execute}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--repo", type=Path, required=True)
|
|
parser.add_argument("--install", action="store_true")
|
|
parser.add_argument("--enable-generation", action="store_true",
|
|
help="upgrade only the reviewed scanner composition with opt-in generation")
|
|
args = parser.parse_args()
|
|
import json
|
|
try:
|
|
print(json.dumps(install(args.repo, args.install, args.enable_generation), sort_keys=True))
|
|
except PermissionError:
|
|
parser.exit(1, "Permission denied updating the reviewed hook; rerun through the permitted filesystem path.\n")
|
|
except (ValueError, OSError, subprocess.SubprocessError):
|
|
parser.exit(1, "Shared hook unverified; existing files and routing are not replaced.\n")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|