aios/active-runtime/app/approved_release_executor.py

#!/usr/bin/env python3
"""One-shot governed executor for an already approved exact release binding.

This module deliberately does not build release artifacts. The build/staging
evidence is completed before Owner production approval. Production execution
consumes that immutable binding, revalidates it, and performs narrow
side-effects through adapters in a durable fail-closed order.

Adapter protocol:
  stdin: JSON request with operation = "probe" | "apply" | "rollback"
  stdout: one JSON object on the final non-empty line
  shell: never used

The executor never rewrites tags, never writes refs/heads/main, and records
Baseline last.
"""
from __future__ import annotations

import argparse
import dataclasses
import fcntl
import hashlib
import json
import os
import re
import stat
import subprocess
import sys
import tempfile
import time
from pathlib import Path
from typing import Any, Mapping, Sequence

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}$")

REQUIRED_V2R1C7_MEMBERS = {
    "payload/VERSION",
    "payload/active-runtime/app/deployment_state.py",
    "payload/active-runtime/app/v2r1c6_privacy_tools.py",
    "payload/active-runtime/app/v2r1c7_health_service.py",
    "payload/deploy/aios-v2/systemd/aios-v2r1c7-production.service",
    "payload/deploy/aios-v2/v2r1c7-runtime-config.json",
    "payload/deploy/aios-v2/v2r1c7-rollback.json",
    "payload/scripts/probe_v2r1c7_health.py",
    "payload/scripts/rollback_v2r1c7_release.sh",
    "payload/BUILD-METADATA.json",
    "payload/ARTIFACT-MANIFEST.json",
    "payload/ROLLBACK-METADATA.json",
}

class ReleaseBlocked(RuntimeError):
    pass

class ManualOwnerReviewRequired(ReleaseBlocked):
    pass

class ReconciliationRequired(ReleaseBlocked):
    pass


@dataclasses.dataclass(frozen=True)
class Binding:
    version: str
    target_oid: str
    evidence_closure_oid: str
    artifact_sha256: str
    approval_plan_id: str
    artifact_path: str
    artifact_size: int
    production_service: str
    production_port: int
    repository_identity: str = "aios/aios"

    def __post_init__(self) -> None:
        if not self.version or not OID_RE.fullmatch(self.target_oid):
            raise ValueError("invalid release target binding")
        if not OID_RE.fullmatch(self.evidence_closure_oid):
            raise ValueError("invalid evidence closure OID")
        if not SHA_RE.fullmatch(self.artifact_sha256):
            raise ValueError("invalid artifact SHA-256")
        if not PLAN_RE.fullmatch(self.approval_plan_id):
            raise ValueError("invalid approval plan id")
        if self.artifact_size <= 0:
            raise ValueError("invalid artifact size")
        if not self.production_service:
            raise ValueError("production service is required")
        if not 1024 <= self.production_port <= 65535:
            raise ValueError("invalid production port")
        if self.repository_identity != "aios/aios":
            raise ValueError("unexpected repository identity")

    @property
    def tag(self) -> str:
        return self.version

    @property
    def fingerprint(self) -> str:
        payload = {
            "version": self.version,
            "target_oid": self.target_oid,
            "evidence_closure_oid": self.evidence_closure_oid,
            "artifact_sha256": self.artifact_sha256,
            "approval_plan_id": self.approval_plan_id,
            "artifact_size": self.artifact_size,
            "production_service": self.production_service,
            "production_port": self.production_port,
            "repository_identity": self.repository_identity,
        }
        return hashlib.sha256(canonical_json(payload)).hexdigest()

    def request(self, operation: str) -> dict[str, Any]:
        return {
            "schema": "aios.production-executor-request.v1",
            "operation": operation,
            "binding_fingerprint": self.fingerprint,
            "version": self.version,
            "tag": self.tag,
            "target_oid": self.target_oid,
            "evidence_closure_oid": self.evidence_closure_oid,
            "artifact_sha256": self.artifact_sha256,
            "artifact_path": self.artifact_path,
            "artifact_size": self.artifact_size,
            "approval_plan_id": self.approval_plan_id,
            "production_service": self.production_service,
            "production_port": self.production_port,
            "repository_identity": self.repository_identity,
        }


def canonical_json(value: object) -> bytes:
    return (json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n").encode("utf-8")


def secure_json(path: Path) -> dict[str, Any]:
    st = path.lstat()
    if not stat.S_ISREG(st.st_mode) or path.is_symlink():
        raise ReleaseBlocked(f"non-regular authorization file: {path}")
    if st.st_uid != 0:
        raise ReleaseBlocked(f"authorization file must be root-owned: {path}")
    if st.st_mode & 0o022:
        raise ReleaseBlocked(f"authorization file is group/world writable: {path}")
    data = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(data, dict):
        raise ReleaseBlocked("authorization receipt must be a JSON object")
    return data


def sha256_file(path: Path) -> str:
    if path.is_symlink() or not path.is_file():
        raise ReleaseBlocked("artifact must be a regular non-symlink file")
    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 run_git(git_dir: Path, *args: str, check: bool = True) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        ["git", f"--git-dir={git_dir}", *args],
        check=check,
        capture_output=True,
        text=True,
        timeout=30,
        env={"PATH": os.environ.get("PATH", "")},
    )


def verify_git_lineage(git_dir: Path, binding: Binding) -> dict[str, Any]:
    authority_main = run_git(git_dir, "rev-parse", "refs/heads/main").stdout.strip()
    if not OID_RE.fullmatch(authority_main):
        raise ReleaseBlocked("Authority main OID is invalid")

    target_type = run_git(git_dir, "cat-file", "-t", binding.target_oid).stdout.strip()
    if target_type != "commit":
        raise ReleaseBlocked("target_oid is not a commit")

    evidence_type = run_git(
        git_dir, "cat-file", "-t", binding.evidence_closure_oid
    ).stdout.strip()
    if evidence_type != "commit":
        raise ReleaseBlocked("evidence_closure_oid is not a commit")

    # A control-plane-only patch may advance main after release approval.
    # The approved release target remains valid only while it is still on main.
    target_on_main = run_git(
        git_dir,
        "merge-base",
        "--is-ancestor",
        binding.target_oid,
        authority_main,
        check=False,
    ).returncode == 0
    if not target_on_main:
        raise ManualOwnerReviewRequired(
            "approved target is no longer an ancestor of Authority main"
        )

    target_in_evidence = run_git(
        git_dir,
        "merge-base",
        "--is-ancestor",
        binding.target_oid,
        binding.evidence_closure_oid,
        check=False,
    ).returncode == 0
    if not target_in_evidence:
        raise ReleaseBlocked("approved target is not an ancestor of evidence closure")

    return {
        "authority_main_oid": authority_main,
        "target_oid": binding.target_oid,
        "target_is_ancestor_of_main": True,
        "evidence_closure_oid": binding.evidence_closure_oid,
        "evidence_object_type": evidence_type,
        "target_is_ancestor_of_evidence": True,
    }


def verify_artifact(binding: Binding) -> dict[str, Any]:
    path = Path(binding.artifact_path)
    if path.stat().st_size != binding.artifact_size:
        raise ReleaseBlocked("artifact size mismatch")
    actual = sha256_file(path)
    if actual != binding.artifact_sha256:
        raise ReleaseBlocked("artifact SHA-256 mismatch")

    if binding.version == "V2R1C7":
        # Python tarfile may not have zstd support on the target host, so
        # decompress into a private temporary tar and inspect it there.
        with tempfile.TemporaryDirectory(prefix="aios-artifact-verify-") as temp_dir:
            tar_path = Path(temp_dir) / "release.tar"
            with tar_path.open("wb") as output:
                completed = subprocess.run(
                    ["zstd", "-dc", str(path)],
                    check=False,
                    stdout=output,
                    stderr=subprocess.PIPE,
                    timeout=60,
                    env={"PATH": os.environ.get("PATH", "")},
                )
            if completed.returncode != 0:
                raise ReleaseBlocked("artifact zstd decompression failed")
            import tarfile
            with tarfile.open(tar_path, "r:") as archive:
                names: set[str] = set()
                for member in archive.getmembers():
                    candidate = Path(member.name)
                    if (
                        candidate.is_absolute()
                        or ".." in candidate.parts
                        or member.issym()
                        or member.islnk()
                        or not (member.isfile() or member.isdir())
                    ):
                        raise ReleaseBlocked(f"unsafe artifact member: {member.name}")
                    names.add(member.name)
            missing = sorted(REQUIRED_V2R1C7_MEMBERS - names)
            if missing:
                raise ReleaseBlocked(
                    "artifact missing required members: " + ",".join(missing)
                )
    return {
        "artifact_path": str(path),
        "artifact_sha256": actual,
        "artifact_size": binding.artifact_size,
        "verified": True,
    }


def verify_authorization(receipt: Mapping[str, Any], binding: Binding) -> dict[str, Any]:
    if receipt.get("accepted") is not True:
        raise ReleaseBlocked("Owner production authorization is not accepted")
    if receipt.get("binding_replaced") not in (None, False):
        raise ManualOwnerReviewRequired("Owner binding was replaced")
    expected = {
        "version": binding.version,
        "target_oid": binding.target_oid,
        "evidence_closure_oid": binding.evidence_closure_oid,
        "artifact_sha256": binding.artifact_sha256,
        "approval_plan_id": binding.approval_plan_id,
    }
    for key, value in expected.items():
        if receipt.get(key) != value:
            raise ManualOwnerReviewRequired(f"Owner authorization mismatch: {key}")
    return {"accepted": True, **expected}


class DurableState:
    def __init__(self, root: Path, binding: Binding):
        self.root = root
        self.root.mkdir(parents=True, exist_ok=True)
        os.chmod(self.root, 0o700)
        self.path = self.root / "execution.json"
        self.binding = binding
        self.value: dict[str, Any]
        if self.path.exists():
            existing = json.loads(self.path.read_text(encoding="utf-8"))
            if existing.get("binding_fingerprint") != binding.fingerprint:
                raise ManualOwnerReviewRequired(
                    "state root contains a different production binding"
                )
            self.value = existing
        else:
            self.value = {
                "schema": "aios.production-execution-state.v1",
                "binding_fingerprint": binding.fingerprint,
                "version": binding.version,
                "target_oid": binding.target_oid,
                "state": "AUTHORIZED",
                "completed": {},
                "history": [],
            }
            self.save()

    def save(self) -> None:
        temporary = self.path.with_suffix(".tmp")
        temporary.write_bytes(canonical_json(self.value))
        os.chmod(temporary, 0o600)
        os.replace(temporary, self.path)

    def record(self, state: str, step: str, result: Mapping[str, Any]) -> None:
        now = time.time()
        sanitized = json.loads(json.dumps(result))
        self.value["state"] = state
        self.value["completed"][step] = sanitized
        self.value["history"].append(
            {"at": now, "state": state, "step": step, "result": sanitized}
        )
        self.save()

    def fail(self, state: str, reason: str) -> None:
        self.value["state"] = state
        self.value["history"].append(
            {"at": time.time(), "state": state, "reason": reason}
        )
        self.save()


class FileLock:
    def __init__(self, path: Path):
        self.path = path
        self.handle = None

    def __enter__(self) -> "FileLock":
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self.handle = self.path.open("a+", encoding="utf-8")
        os.chmod(self.path, 0o600)
        try:
            fcntl.flock(
                self.handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB
            )
        except BlockingIOError as exc:
            raise ReleaseBlocked("production executor is already running") from exc
        return self

    def __exit__(self, exc_type, exc, tb) -> None:
        if self.handle is not None:
            fcntl.flock(self.handle.fileno(), fcntl.LOCK_UN)
            self.handle.close()


def parse_command(values: Sequence[str]) -> list[str]:
    command = [str(x) for x in values if str(x)]
    if not command:
        raise ValueError("adapter command is required")
    if not os.path.isabs(command[0]):
        raise ValueError("adapter executable must be an absolute path")
    return command


def run_adapter(command: Sequence[str], request: Mapping[str, Any]) -> dict[str, Any]:
    completed = subprocess.run(
        list(command),
        input=canonical_json(request),
        check=False,
        capture_output=True,
        timeout=180,
        env={"PATH": os.environ.get("PATH", "")},
    )
    if completed.returncode != 0:
        raise ReleaseBlocked(
            f"adapter failed: {Path(command[0]).name} rc={completed.returncode}"
        )
    lines = [line for line in completed.stdout.decode("utf-8").splitlines() if line.strip()]
    if not lines:
        raise ReleaseBlocked("adapter returned no JSON result")
    try:
        value = json.loads(lines[-1])
    except ValueError as exc:
        raise ReleaseBlocked("adapter returned invalid JSON") from exc
    if not isinstance(value, dict):
        raise ReleaseBlocked("adapter result must be an object")
    return value


def require_exact_response(result: Mapping[str, Any], binding: Binding) -> None:
    for key, expected in {
        "version": binding.version,
        "target_oid": binding.target_oid,
        "evidence_closure_oid": binding.evidence_closure_oid,
        "artifact_sha256": binding.artifact_sha256,
        "approval_plan_id": binding.approval_plan_id,
    }.items():
        if key in result and result[key] != expected:
            raise ManualOwnerReviewRequired(f"adapter binding mismatch: {key}")


def tag_step(command: Sequence[str], binding: Binding) -> dict[str, Any]:
    request = binding.request("probe")
    probe = run_adapter(command, request)
    require_exact_response(probe, binding)
    state = probe.get("status")
    if state == "CONFLICT":
        raise ManualOwnerReviewRequired("same-name tag conflicts with approved binding")
    if state == "INDETERMINATE":
        raise ReconciliationRequired("tag publication state is indeterminate")
    if state not in {"APPLIED_VERIFIED", "NOT_APPLIED", "PARTIAL_LOCAL"}:
        raise ReleaseBlocked(f"unsupported tag probe status: {state}")
    if state != "APPLIED_VERIFIED":
        applied = run_adapter(command, binding.request("apply"))
        require_exact_response(applied, binding)
        if applied.get("status") == "CONFLICT":
            raise ManualOwnerReviewRequired("tag conflict during publication")
        probe = run_adapter(command, request)
        require_exact_response(probe, binding)
    if probe.get("status") != "APPLIED_VERIFIED":
        raise ReconciliationRequired("tag was not independently verified")
    if probe.get("peeled_target_oid") != binding.target_oid:
        raise ManualOwnerReviewRequired("published tag peeled target mismatch")
    tag_oid = probe.get("tag_oid")
    if not isinstance(tag_oid, str) or not OID_RE.fullmatch(tag_oid):
        raise ReleaseBlocked("verified annotated tag object OID missing")
    return dict(probe)


def deploy_step(command: Sequence[str], binding: Binding) -> dict[str, Any]:
    probe = run_adapter(command, binding.request("probe"))
    require_exact_response(probe, binding)
    if probe.get("status") == "ACTIVE_VERIFIED":
        return dict(probe)
    if probe.get("status") == "CONFLICT":
        raise ManualOwnerReviewRequired("production runtime has conflicting binding")
    applied = run_adapter(command, binding.request("apply"))
    require_exact_response(applied, binding)
    probe = run_adapter(command, binding.request("probe"))
    require_exact_response(probe, binding)
    if probe.get("status") != "ACTIVE_VERIFIED":
        try:
            run_adapter(command, binding.request("rollback"))
        except Exception:
            pass
        raise ReconciliationRequired("production runtime did not verify after apply")
    if (
        probe.get("deployment_state") != "ACTIVE"
        or probe.get("production_deployed") is not True
    ):
        raise ReconciliationRequired("production health contract is not ACTIVE")
    return dict(probe)


def finalize_step(
    command: Sequence[str],
    binding: Binding,
    *,
    name: str,
    tag_oid: str,
) -> dict[str, Any]:
    request = binding.request("probe")
    request["tag_oid"] = tag_oid
    probe = run_adapter(command, request)
    require_exact_response(probe, binding)
    if probe.get("status") == "CONFLICT":
        raise ManualOwnerReviewRequired(f"{name} has conflicting release binding")
    if probe.get("status") != "VERIFIED":
        apply_request = binding.request("apply")
        apply_request["tag_oid"] = tag_oid
        applied = run_adapter(command, apply_request)
        require_exact_response(applied, binding)
        probe = run_adapter(command, request)
        require_exact_response(probe, binding)
    if probe.get("status") != "VERIFIED":
        raise ReconciliationRequired(f"{name} did not verify after apply")
    if probe.get("tag_oid") not in (None, tag_oid):
        raise ManualOwnerReviewRequired(f"{name} tag object mismatch")
    return dict(probe)


def execute(
    *,
    binding: Binding,
    authorization_path: Path,
    repository: Path,
    state_root: Path,
    tag_command: Sequence[str],
    deploy_command: Sequence[str],
    registry_command: Sequence[str],
    baseline_command: Sequence[str],
) -> dict[str, Any]:
    authorization = verify_authorization(secure_json(authorization_path), binding)
    lineage = verify_git_lineage(repository, binding)
    artifact = verify_artifact(binding)
    state = DurableState(state_root, binding)
    with FileLock(state_root / "executor.lock"):
        state.record(
            "PREFLIGHT_VERIFIED",
            "preflight",
            {"authorization": authorization, "lineage": lineage, "artifact": artifact},
        )
        tag: dict[str, Any] | None = None
        runtime: dict[str, Any] | None = None
        try:
            tag = tag_step(tag_command, binding)
            state.record("TAG_VERIFIED", "tag", tag)

            runtime = deploy_step(deploy_command, binding)
            state.record("RUNTIME_VERIFIED", "runtime", runtime)

            registry = finalize_step(
                registry_command,
                binding,
                name="registry",
                tag_oid=tag["tag_oid"],
            )
            state.record("REGISTRY_VERIFIED", "registry", registry)

            baseline = finalize_step(
                baseline_command,
                binding,
                name="baseline",
                tag_oid=tag["tag_oid"],
            )
            state.record("BASELINE_VERIFIED", "baseline", baseline)

            # Final readback: no mutation, all four planes must still agree.
            final_tag = tag_step(tag_command, binding)
            final_runtime = run_adapter(deploy_command, binding.request("probe"))
            final_registry_request = binding.request("probe")
            final_registry_request["tag_oid"] = tag["tag_oid"]
            final_registry = run_adapter(registry_command, final_registry_request)
            final_baseline_request = binding.request("probe")
            final_baseline_request["tag_oid"] = tag["tag_oid"]
            final_baseline = run_adapter(baseline_command, final_baseline_request)
            for value in (final_runtime, final_registry, final_baseline):
                require_exact_response(value, binding)
            if final_runtime.get("status") != "ACTIVE_VERIFIED":
                raise ReconciliationRequired("final runtime readback failed")
            if final_registry.get("status") != "VERIFIED":
                raise ReconciliationRequired("final Registry readback failed")
            if final_baseline.get("status") != "VERIFIED":
                raise ReconciliationRequired("final Baseline readback failed")
            final = {
                "status": "PRODUCTION_ACCEPTED",
                "version": binding.version,
                "target_oid": binding.target_oid,
                "evidence_closure_oid": binding.evidence_closure_oid,
                "artifact_sha256": binding.artifact_sha256,
                "approval_plan_id": binding.approval_plan_id,
                "tag_oid": tag["tag_oid"],
                "tag": final_tag,
                "runtime": final_runtime,
                "registry": final_registry,
                "baseline": final_baseline,
                "force_push_used": False,
                "tag_rewritten": False,
                "direct_main_push": False,
                "secrets_committed": False,
            }
            state.record("COMPLETED", "final_consistency", final)
            return final
        except ManualOwnerReviewRequired as exc:
            state.fail("MANUAL_OWNER_REVIEW_REQUIRED", str(exc))
            raise
        except ReconciliationRequired as exc:
            state.fail("RECONCILIATION_REQUIRED", str(exc))
            raise
        except Exception as exc:
            # Once a formal tag has been verified, no later error is a clean
            # pre-side-effect failure. Preserve the tag and require idempotent
            # reconciliation rather than pretending the release never started.
            if tag is not None:
                reason = f"{type(exc).__name__}: {exc}"
                state.fail("RECONCILIATION_REQUIRED", reason)
                raise ReconciliationRequired(reason) from exc
            state.fail("FAIL_CLOSED", f"{type(exc).__name__}: {exc}")
            raise


def parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser()
    p.add_argument("--authorization", required=True, type=Path)
    p.add_argument("--repository", required=True, type=Path)
    p.add_argument("--state-root", required=True, type=Path)
    p.add_argument("--artifact", required=True, type=Path)
    p.add_argument("--artifact-size", required=True, type=int)
    p.add_argument("--version", required=True)
    p.add_argument("--target-oid", required=True)
    p.add_argument("--evidence-closure-oid", required=True)
    p.add_argument("--artifact-sha256", required=True)
    p.add_argument("--approval-plan-id", required=True)
    p.add_argument("--production-service", required=True)
    p.add_argument("--production-port", required=True, type=int)
    p.add_argument("--repository-identity", default="aios/aios")
    p.add_argument("--tag-adapter", required=True, nargs="+")
    p.add_argument("--deploy-adapter", required=True, nargs="+")
    p.add_argument("--registry-adapter", required=True, nargs="+")
    p.add_argument("--baseline-adapter", required=True, nargs="+")
    return p


def main() -> int:
    args = parser().parse_args()
    binding = Binding(
        version=args.version,
        target_oid=args.target_oid,
        evidence_closure_oid=args.evidence_closure_oid,
        artifact_sha256=args.artifact_sha256,
        approval_plan_id=args.approval_plan_id,
        artifact_path=str(args.artifact),
        artifact_size=args.artifact_size,
        production_service=args.production_service,
        production_port=args.production_port,
        repository_identity=args.repository_identity,
    )
    try:
        result = execute(
            binding=binding,
            authorization_path=args.authorization,
            repository=args.repository,
            state_root=args.state_root,
            tag_command=parse_command(args.tag_adapter),
            deploy_command=parse_command(args.deploy_adapter),
            registry_command=parse_command(args.registry_adapter),
            baseline_command=parse_command(args.baseline_adapter),
        )
        print(json.dumps(result, sort_keys=True))
        return 0
    except ManualOwnerReviewRequired as exc:
        print(json.dumps({"status": "MANUAL_OWNER_REVIEW_REQUIRED", "reason": str(exc)}, sort_keys=True))
        return 20
    except ReconciliationRequired as exc:
        print(json.dumps({"status": "RECONCILIATION_REQUIRED", "reason": str(exc)}, sort_keys=True))
        return 21
    except Exception as exc:
        print(json.dumps({"status": "FAIL_CLOSED", "error": type(exc).__name__, "reason": str(exc)}, sort_keys=True))
        return 22


if __name__ == "__main__":
    raise SystemExit(main())