115 lines
4.7 KiB
Python
Executable file
115 lines
4.7 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""Prove a candidate SHA may be published. Exit 0 only if nothing objected.
|
|
|
|
Run inside the consuming repository, by CI, before any push to the destination.
|
|
Every failure path exits non-zero; there is no mode in which an error is treated
|
|
as permission.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
sys.path.insert(0, os.path.join(HERE, "..", "lib"))
|
|
|
|
import gates # noqa: E402
|
|
import history # noqa: E402
|
|
import mechanism # noqa: E402
|
|
import policy as policy_mod # noqa: E402
|
|
|
|
|
|
def main(argv=None):
|
|
ap = argparse.ArgumentParser(prog="verify-publication")
|
|
ap.add_argument("--policy-dir", required=True,
|
|
help="checkout of the policy repository (must not be inside --repo-root)")
|
|
ap.add_argument("--repo", required=True, help="internal repository, owner/name")
|
|
ap.add_argument("--sha", required=True, help="candidate commit, full lowercase hex")
|
|
ap.add_argument("--ref", required=True, help="candidate branch short name")
|
|
ap.add_argument("--repo-root", default=".")
|
|
ap.add_argument("--gitea-url", help="internal Gitea base URL; required unless --no-gates")
|
|
ap.add_argument("--token-env", default="GITEA_READ_TOKEN")
|
|
ap.add_argument("--no-gates", action="store_true",
|
|
help="check policy, mechanism and history only; NEVER sufficient to publish")
|
|
args = ap.parse_args(argv)
|
|
|
|
repo_root = os.path.abspath(args.repo_root)
|
|
reasons = []
|
|
|
|
try:
|
|
policy_set = policy_mod.PolicySet.from_directory(args.policy_dir)
|
|
pol = policy_set.authorize(args.repo)
|
|
policy_mod.assert_policy_is_external(pol, repo_root)
|
|
except policy_mod.PolicyError as exc:
|
|
_emit(["POLICY " + r for r in exc.reasons], allowed=False)
|
|
return 1
|
|
|
|
pinned_paths = ()
|
|
manifest_path = os.path.join(repo_root, ".publication", "MANIFEST.json")
|
|
if not os.path.isfile(manifest_path):
|
|
reasons.append("MECHANISM .publication/MANIFEST.json is absent")
|
|
else:
|
|
try:
|
|
manifest = mechanism.read_manifest(manifest_path)
|
|
pinned_paths = tuple(manifest["files"]) + (".publication/MANIFEST.json",)
|
|
reasons.extend("MECHANISM " + r for r in mechanism.verify(
|
|
repo_root, manifest, pol.mechanism_manifest_sha256, pol.mechanism_version))
|
|
except (json.JSONDecodeError, KeyError) as exc:
|
|
reasons.append(f"MECHANISM MANIFEST.json is unusable ({exc})")
|
|
|
|
cleanup = []
|
|
try:
|
|
reasons.extend("HISTORY " + r for r in history.verify(
|
|
args.ref, repo_root, pol, pinned_paths=pinned_paths, warnings=cleanup))
|
|
except history.GitError as exc:
|
|
reasons.append(f"HISTORY {exc}")
|
|
|
|
unproven = []
|
|
if args.no_gates:
|
|
reasons.append("GATES --no-gates was passed; gate proof was not attempted "
|
|
"and this run must not authorize a push")
|
|
else:
|
|
if not args.gitea_url:
|
|
reasons.append("GATES --gitea-url is required to prove gates")
|
|
else:
|
|
import giteaapi # noqa: E402
|
|
try:
|
|
client = giteaapi.GiteaClient(args.gitea_url, token_env=args.token_env)
|
|
except giteaapi.ApiError as exc:
|
|
client = None
|
|
reasons.append(f"GATES {exc}")
|
|
if client is not None:
|
|
request = gates.PublicationRequest(
|
|
internal_repository=args.repo,
|
|
destination_host=pol.destination_host,
|
|
destination_repository=pol.destination_repository,
|
|
sha=args.sha, ref=args.ref)
|
|
report = gates.evaluate(pol, request, client)
|
|
reasons.extend("GATES " + d for d in report.denials)
|
|
unproven = report.unproven
|
|
|
|
_emit(reasons, allowed=not reasons, unproven=unproven, policy=pol, cleanup=cleanup)
|
|
return 1 if reasons else 0
|
|
|
|
|
|
def _emit(reasons, allowed, unproven=(), policy=None, cleanup=()):
|
|
if policy is not None:
|
|
print(f"[publication] {policy.internal_repository} -> "
|
|
f"{policy.destination_host}/{policy.destination_repository} "
|
|
f"({policy.projection}, publisher {policy.publisher_identity})")
|
|
print(f"[publication] verdict: {'ALLOW' if allowed else 'DENY'}")
|
|
for r in reasons:
|
|
print(f"[publication] DENY {r}")
|
|
for u in unproven:
|
|
print(f"[publication] unproven: {u}")
|
|
if cleanup:
|
|
print(f"[publication] {len(cleanup)} cleanup finding(s) -- recorded, not blocking:")
|
|
for c in cleanup[:40]:
|
|
print(f"[publication] {c}")
|
|
if len(cleanup) > 40:
|
|
print(f"[publication] ... and {len(cleanup) - 40} more")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|