#!/usr/bin/env python3 """Atomic Registry/Baseline finalizer for an already ACTIVE exact release. This adapter never writes Git refs. It is intended for the AIOS runtime Registry and authoritative runtime-anchor Baseline only. Registry is finalized before Baseline. Baseline apply independently verifies: - exact Authority annotated tag and peeled target - exact ACTIVE production health - exact V2R1C7 Registry record - exact previous Baseline file SHA-256 (CAS) """ from __future__ import annotations import argparse import hashlib import json import os import re import stat import subprocess import sys import tempfile import time import urllib.error import urllib.request from pathlib import Path from typing import Any, Mapping from release_version import validate_release_version OID_RE = re.compile(r"^[0-9a-f]{40}$") SHA_RE = re.compile(r"^[0-9a-f]{64}$") PLAN_RE = re.compile(r"^sha256:[0-9a-f]{64}$") def sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as source: for block in iter(lambda: source.read(1024 * 1024), b""): digest.update(block) return digest.hexdigest() def atomic_json(path: Path, value: Mapping[str, Any], mode: int = 0o640) -> None: path.parent.mkdir(parents=True, exist_ok=True) # Existing runtime-authority files may be root:gitops and read-only to # non-root readers. Preserve their owner/group/mode across os.replace(). if path.exists(): metadata = path.lstat() if not stat.S_ISREG(metadata.st_mode) or path.is_symlink(): raise RuntimeError("ATOMIC_TARGET_NOT_REGULAR") target_mode = stat.S_IMODE(metadata.st_mode) target_uid = metadata.st_uid target_gid = metadata.st_gid else: parent = path.parent.stat() target_mode = mode target_uid = 0 if os.geteuid() == 0 else os.geteuid() target_gid = parent.st_gid fd, name = tempfile.mkstemp(prefix=f".{path.name}.", dir=str(path.parent)) try: with os.fdopen(fd, "w", encoding="utf-8") as output: json.dump(dict(value), output, sort_keys=True, indent=2) output.write("\n") output.flush() os.fsync(output.fileno()) os.chmod(name, target_mode) if os.geteuid() == 0: os.chown(name, target_uid, target_gid) os.replace(name, path) dfd = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY) try: os.fsync(dfd) finally: os.close(dfd) finally: try: os.unlink(name) except FileNotFoundError: pass def load_request() -> dict[str, Any]: value = json.load(sys.stdin) if not isinstance(value, dict) or value.get("schema") != "aios.production-executor-request.v1": raise ValueError("request_schema_invalid") if value.get("repository_identity") != "aios/aios": raise ValueError("repository_identity_invalid") try: validate_release_version(value.get("version", "")) except ValueError as exc: raise ValueError("version_invalid") from exc for name in ("target_oid", "evidence_closure_oid"): if not OID_RE.fullmatch(str(value.get(name, ""))): raise ValueError(f"{name}_invalid") if not SHA_RE.fullmatch(str(value.get("artifact_sha256", ""))): raise ValueError("artifact_sha256_invalid") if not PLAN_RE.fullmatch(str(value.get("approval_plan_id", ""))): raise ValueError("approval_plan_id_invalid") if "tag_oid" in value and not OID_RE.fullmatch(str(value["tag_oid"])): raise ValueError("tag_oid_invalid") return value def exact_response(value: Mapping[str, Any], status: str, **extra: Any) -> dict[str, Any]: return { "status": status, "version": value["version"], "target_oid": value["target_oid"], "evidence_closure_oid": value["evidence_closure_oid"], "artifact_sha256": value["artifact_sha256"], "approval_plan_id": value["approval_plan_id"], **extra, } def git(repository: Path, *args: str, check: bool = True) -> subprocess.CompletedProcess[str]: return subprocess.run( ["git", f"--git-dir={repository}", *args], check=check, capture_output=True, text=True, timeout=30, env={"PATH": os.environ.get("PATH", "")}, ) def verify_local_tag(repository: Path, value: Mapping[str, Any]) -> tuple[str, str]: tag_oid = git(repository, "rev-parse", "--verify", f"refs/tags/{value['version']}").stdout.strip() peeled = git(repository, "rev-parse", "--verify", f"refs/tags/{value['version']}^{{commit}}").stdout.strip() if tag_oid == peeled or peeled != value["target_oid"]: raise RuntimeError("FORMAL_TAG_BINDING_MISMATCH") requested = value.get("tag_oid") if requested and requested != tag_oid: raise RuntimeError("FORMAL_TAG_OBJECT_MISMATCH") return tag_oid, peeled def health(url: str, value: Mapping[str, Any]) -> dict[str, Any]: try: with urllib.request.urlopen(url, timeout=5) as response: body = json.loads(response.read().decode("utf-8")) status = response.status except (urllib.error.URLError, TimeoutError, ValueError) as exc: raise RuntimeError("PRODUCTION_HEALTH_UNAVAILABLE") from exc if not ( status == 200 and body.get("version") == value["version"] and body.get("commit_oid") == value["target_oid"] and body.get("artifact_sha256") == value["artifact_sha256"] and body.get("deployment_state") == "ACTIVE" and body.get("production_deployed") is True ): raise RuntimeError("PRODUCTION_HEALTH_BINDING_MISMATCH") return body def registry_record(value: Mapping[str, Any], tag_oid: str, production_health: Mapping[str, Any]) -> dict[str, Any]: return { "schema": "aios.production-registry.v2", "version": value["version"], "repository": "aios/aios", "target_oid": value["target_oid"], "tag": value["version"], "tag_oid": tag_oid, "tag_object": tag_oid, "peeled_target_oid": value["target_oid"], "evidence_closure_oid": value["evidence_closure_oid"], "artifact_sha256": value["artifact_sha256"], "approval_plan_id": value["approval_plan_id"], "production_service": value["production_service"], "production_port": value["production_port"], "registered": True, "enabled": True, "production_deployed": True, "status": "PRODUCTION_DEPLOYED", "deployment_state": production_health["deployment_state"], "updated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), } def registry_matches(record: Mapping[str, Any], value: Mapping[str, Any], tag_oid: str) -> bool: return all( record.get(key) == expected for key, expected in { "version": value["version"], "target_oid": value["target_oid"], "tag_oid": tag_oid, "evidence_closure_oid": value["evidence_closure_oid"], "artifact_sha256": value["artifact_sha256"], "approval_plan_id": value["approval_plan_id"], "production_service": value["production_service"], "production_port": value["production_port"], "production_deployed": True, }.items() ) and record.get("registered") is True and record.get("enabled") is True def baseline_matches(record: Mapping[str, Any], value: Mapping[str, Any], tag_oid: str) -> bool: current = record.get("current_baseline") or record.get("version") return current == value["version"] and all( record.get(key) == expected for key, expected in { "target_oid": value["target_oid"], "tag_oid": tag_oid, "evidence_closure_oid": value["evidence_closure_oid"], "artifact_sha256": value["artifact_sha256"], "approval_plan_id": value["approval_plan_id"], "production_service": value["production_service"], "production_port": value["production_port"], "status": "PRODUCTION_DEPLOYED", }.items() ) def registry_probe(args: argparse.Namespace, value: Mapping[str, Any]) -> dict[str, Any]: tag_oid, _ = verify_local_tag(args.repository, value) path = args.registry_root / f"{value['version']}.json" if not path.exists(): return exact_response(value, "NOT_APPLIED", tag_oid=tag_oid) record = json.loads(path.read_text(encoding="utf-8")) if registry_matches(record, value, tag_oid): return exact_response( value, "VERIFIED", tag_oid=tag_oid, registry_path=str(path), registry_sha256=sha256_file(path), ) return exact_response(value, "CONFLICT", tag_oid=tag_oid, reason="registry_binding_mismatch") def registry_apply(args: argparse.Namespace, value: Mapping[str, Any]) -> dict[str, Any]: existing = registry_probe(args, value) if existing["status"] in {"VERIFIED", "CONFLICT"}: return existing tag_oid, _ = verify_local_tag(args.repository, value) production_health = health(args.health_url, value) path = args.registry_root / f"{value['version']}.json" atomic_json(path, registry_record(value, tag_oid, production_health)) return registry_probe(args, value) def secure_baseline_authority_file(path: Path) -> None: if not path.is_file() or path.is_symlink(): raise RuntimeError("BASELINE_AUTHORITY_FILE_INVALID") baseline_stat = path.lstat() if baseline_stat.st_uid != 0 or baseline_stat.st_mode & 0o022: raise RuntimeError("BASELINE_AUTHORITY_PERMISSIONS_INVALID") def baseline_probe(args: argparse.Namespace, value: Mapping[str, Any]) -> dict[str, Any]: tag_oid, _ = verify_local_tag(args.repository, value) secure_baseline_authority_file(args.baseline_path) record = json.loads(args.baseline_path.read_text(encoding="utf-8")) if baseline_matches(record, value, tag_oid): return exact_response( value, "VERIFIED", tag_oid=tag_oid, baseline_path=str(args.baseline_path), baseline_sha256=sha256_file(args.baseline_path), ) current = record.get("current_baseline") or record.get("version") current_sha = sha256_file(args.baseline_path) if ( current == args.expected_previous_baseline and current_sha == args.expected_previous_baseline_sha256 ): return exact_response( value, "NOT_APPLIED", tag_oid=tag_oid, previous_baseline=current, previous_baseline_sha256=current_sha, ) return exact_response( value, "CONFLICT", tag_oid=tag_oid, reason="baseline_precondition_or_binding_mismatch", observed_baseline=current, observed_sha256=current_sha, ) def baseline_apply(args: argparse.Namespace, value: Mapping[str, Any]) -> dict[str, Any]: existing = baseline_probe(args, value) if existing["status"] in {"VERIFIED", "CONFLICT"}: return existing tag_oid, _ = verify_local_tag(args.repository, value) production_health = health(args.health_url, value) registry_path = args.registry_root / f"{value['version']}.json" if not registry_path.is_file(): raise RuntimeError("REGISTRY_NOT_FINALIZED") registry = json.loads(registry_path.read_text(encoding="utf-8")) if not registry_matches(registry, value, tag_oid): raise RuntimeError("REGISTRY_BINDING_NOT_VERIFIED") previous = json.loads(args.baseline_path.read_text(encoding="utf-8")) if sha256_file(args.baseline_path) != args.expected_previous_baseline_sha256: raise RuntimeError("BASELINE_CAS_MISMATCH") generation = max( int(previous.get("generation", 0) or 0), int(previous.get("authority_generation", 0) or 0), ) + 1 record = { "schema": "aios.production-baseline.v2", "source_identity": "aios-gitops-runtime-anchor-baseline-authority-v1", "current_baseline": value["version"], "version": value["version"], "previous_baseline": args.expected_previous_baseline, "previous_version": args.expected_previous_baseline, "status": "PRODUCTION_DEPLOYED", "target_oid": value["target_oid"], "peeled_target_oid": value["target_oid"], "tag_oid": tag_oid, "evidence_closure_oid": value["evidence_closure_oid"], "approval_plan_id": value["approval_plan_id"], "artifact_sha256": value["artifact_sha256"], "production_artifact_sha256": value["artifact_sha256"], "production_service": value["production_service"], "production_port": value["production_port"], "production_deployed": True, "deployment_state": production_health["deployment_state"], "registry_digest": sha256_file(registry_path), "registry_source": str(registry_path), "generation": generation, "authority_generation": generation, # The actual previously deployed production baseline is the rollback # target. V2R1C6 has a formal tag but was never production-deployed. "rollback_version": args.expected_previous_baseline, "rollback_tag_oid": previous.get("tag_oid"), "rollback_target_oid": previous.get("target_oid"), "promoted_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "promoted_by": "aios-approved-production-release-executor", } atomic_json(args.baseline_path, record) return baseline_probe(args, value) def main() -> int: parser = argparse.ArgumentParser() parser.add_argument( "--mode", choices=("registry", "baseline"), default=os.environ.get("AIOS_RELEASE_METADATA_MODE"), required=os.environ.get("AIOS_RELEASE_METADATA_MODE") is None, ) parser.add_argument( "--repository", type=Path, default=Path(os.environ["AIOS_GITOPS_AUTHORITATIVE_GIT_DIR"]), ) parser.add_argument( "--registry-root", type=Path, default=Path(os.environ["AIOS_RUNTIME_REGISTRY_ROOT"]), ) parser.add_argument( "--baseline-path", type=Path, default=Path(os.environ["AIOS_RUNTIME_BASELINE_PATH"]), ) parser.add_argument( "--expected-previous-baseline", default=os.environ.get("AIOS_PREVIOUS_BASELINE", "V2R1C5"), ) parser.add_argument( "--expected-previous-baseline-sha256", default=os.environ.get("AIOS_PREVIOUS_BASELINE_SHA256"), required=os.environ.get("AIOS_PREVIOUS_BASELINE_SHA256") is None, ) parser.add_argument( "--health-url", default=os.environ.get("AIOS_RELEASE_HEALTH_URL", "http://127.0.0.1:18927/health"), ) args = parser.parse_args() value = load_request() operation = value.get("operation") if args.mode == "registry": if operation == "probe": result = registry_probe(args, value) elif operation == "apply": result = registry_apply(args, value) else: raise ValueError("unsupported_operation") else: if operation == "probe": result = baseline_probe(args, value) elif operation == "apply": result = baseline_apply(args, value) else: raise ValueError("unsupported_operation") print(json.dumps(result, sort_keys=True)) return 0 if __name__ == "__main__": try: raise SystemExit(main()) except Exception as exc: print( json.dumps( {"status": "FAIL_CLOSED", "error": type(exc).__name__, "reason": str(exc)}, sort_keys=True, ) ) raise SystemExit(2)