125 lines
4.4 KiB
Python
Executable File
125 lines
4.4 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Generate one canonical source entry for a governed non-main change.
|
|
|
|
This utility only creates the file in the caller's worktree. It never commits,
|
|
pushes, changes main, rewrites a tag, or updates a production projection. The
|
|
caller must submit the resulting file through the existing governed writer and
|
|
approval flow.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
APP = Path(__file__).resolve().parents[1] / "active-runtime" / "app"
|
|
sys.path.insert(0, str(APP))
|
|
|
|
from version_registry import ( # noqa: E402
|
|
build_source_registry_entry,
|
|
configured_authoritative_git_dir,
|
|
configured_repository_identity,
|
|
source_registry_json,
|
|
validate_source_registry_record,
|
|
verify_approval_authority,
|
|
verify_authoritative_repository,
|
|
verify_source_registry_git_identity,
|
|
)
|
|
|
|
|
|
def parser() -> argparse.ArgumentParser:
|
|
value = argparse.ArgumentParser()
|
|
value.add_argument("--output", required=True, type=Path)
|
|
value.add_argument("--version", required=True)
|
|
value.add_argument("--tag-oid", required=True)
|
|
value.add_argument("--target-oid", required=True)
|
|
value.add_argument("--peeled-target-oid", required=True)
|
|
value.add_argument("--evidence-closure-oid", required=True)
|
|
value.add_argument("--approval-plan-id", required=True)
|
|
value.add_argument(
|
|
"--repository",
|
|
type=Path,
|
|
default=configured_authoritative_git_dir(),
|
|
help="authoritative bare repository; required for release generation",
|
|
)
|
|
value.add_argument(
|
|
"--approval-authority",
|
|
type=Path,
|
|
default=(
|
|
Path(os.environ["AIOS_GITOPS_APPROVAL_AUTHORITY"])
|
|
if os.environ.get("AIOS_GITOPS_APPROVAL_AUTHORITY")
|
|
else None
|
|
),
|
|
help="canonical approval/publication receipt for this release request",
|
|
)
|
|
value.add_argument(
|
|
"--repository-identity",
|
|
default=configured_repository_identity(),
|
|
)
|
|
return value
|
|
|
|
|
|
def main() -> int:
|
|
args = parser().parse_args()
|
|
if args.approval_authority is None:
|
|
raise ValueError("canonical approval authority is required")
|
|
verify_authoritative_repository(
|
|
args.repository,
|
|
repository_identity=args.repository_identity,
|
|
)
|
|
tag_oid = args.tag_oid.lower()
|
|
target_oid = args.target_oid.lower()
|
|
peeled_target_oid = args.peeled_target_oid.lower()
|
|
evidence_closure_oid = args.evidence_closure_oid.lower()
|
|
expected = build_source_registry_entry(
|
|
version=args.version,
|
|
tag_oid=tag_oid,
|
|
target_oid=target_oid,
|
|
peeled_target_oid=peeled_target_oid,
|
|
evidence_closure_oid=evidence_closure_oid,
|
|
approval_plan_id=args.approval_plan_id,
|
|
repository=args.repository_identity,
|
|
)
|
|
verify_source_registry_git_identity(args.repository, expected)
|
|
verify_approval_authority(args.approval_authority, expected)
|
|
destination = args.output.resolve()
|
|
if destination.name != f"{args.version}.json" or destination.parent.name != "versions":
|
|
raise ValueError("output must be versions/<version>.json")
|
|
payload = source_registry_json(expected)
|
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
|
if destination.is_symlink():
|
|
raise ValueError("existing source registry path must not be a symlink")
|
|
if destination.exists() and not destination.is_file():
|
|
raise ValueError("existing source registry path must be a regular file")
|
|
if destination.exists():
|
|
current = json.loads(destination.read_text(encoding="utf-8"))
|
|
validate_source_registry_record(
|
|
current,
|
|
expected_version=args.version,
|
|
expected_tag_oid=tag_oid,
|
|
expected_target_oid=target_oid,
|
|
expected_evidence_closure_oid=evidence_closure_oid,
|
|
expected_approval_plan_id=args.approval_plan_id,
|
|
)
|
|
print(json.dumps({"status": "EXISTING_IDENTITY_VERIFIED", "path": str(destination)}))
|
|
return 0
|
|
descriptor = os.open(
|
|
destination,
|
|
os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0),
|
|
0o640,
|
|
)
|
|
try:
|
|
os.write(descriptor, payload)
|
|
os.fsync(descriptor)
|
|
finally:
|
|
os.close(descriptor)
|
|
print(json.dumps({"status": "GENERATED", "path": str(destination)}))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|