aios/active-runtime/app/release_metadata_adapter.py

#!/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

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")
    if not re.fullmatch(r"V\d+R\d+C\d+(?:-[A-Za-z0-9_.-]+)?", str(value.get("version", ""))):
        raise ValueError("version_invalid")
    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)