atlas-iac/scripts/ops/hermes_handoff_arming.py

#!/usr/bin/env python3
"""Arming, provenance, and push-target rules for the armed ephemeral mode.

Nothing here touches the network or spawns a process. It is the half of armed
mode that must be true before a single command exists: the confirmation phrase,
the fixed repository and base, the one ephemeral ref grammar, and a local
worktree whose HEAD and ``origin/main`` are byte-exact with the reviewed release.
"""

from __future__ import annotations

import configparser
import re
from dataclasses import dataclass
from pathlib import Path

from hermes_handoff_policy import (
    EXPECTED_BASE,
    EXPECTED_REMOTE,
    EXPECTED_REPO,
    arm_ephemeral_policy,
)

CONFIRMATION = "ARM EPHEMERAL HERMES HANDOFF PUSH"
REF_PREFIX = "ephemeral/hermes-handoff-acceptance"
REF_RE = re.compile(rf"{REF_PREFIX}/[a-z0-9][a-z0-9-]{{7,63}}\Z")
TOKEN_RE = re.compile(r"[a-z0-9][a-z0-9-]{7,63}\Z")
SHA_RE = re.compile(r"[0-9a-f]{40}\Z")
EXPECTED_HEAD_REF = "feature/hermes-full-handoff-acceptance"
EXPECTED_REMOTE_URL = "https://scm.bstein.dev/titan/atlas-iac.git"
PROTECTED_NAMES = frozenset(
    {
        "default",
        "develop",
        "head",
        "main",
        "master",
        "prod",
        "production",
        "release",
        "stable",
        "trunk",
    }
)


class ArmingError(ValueError):
    """Raised when an ephemeral run is not safe to start."""


@dataclass(frozen=True)
class ArmRequest:
    """Every fixed and caller-confirmed input to one armed run."""

    repo: str
    remote: str
    token: str
    confirmation: str
    base: str = EXPECTED_BASE
    expected_head: str = ""
    expected_base_sha: str = ""

    @property
    def ref(self) -> str:
        return f"{REF_PREFIX}/{self.token}"


def normalise_branch(name: str) -> str:
    """Return a branch name stripped of ref decoration, for comparison."""
    return name.strip().removeprefix("refs/heads/").strip("/").lower()


def assert_push_target_allowed(ref: str) -> None:
    """Raise unless ``ref`` is the one ephemeral branch grammar."""
    if normalise_branch(ref) in PROTECTED_NAMES:
        raise ArmingError(f"refusing to push to protected branch {ref!r}")
    if not REF_RE.fullmatch(ref):
        raise ArmingError(f"push target {ref!r} is not an ephemeral acceptance ref")


def bounded_read(path: Path, limit: int = 1024 * 1024) -> str:
    data = path.read_bytes()
    if not data or len(data) > limit:
        raise ArmingError(f"local Git metadata is empty or exceeds {limit} bytes")
    return data.decode("utf-8", "strict")


def _attest_worktree(root: Path, expected_head: str, expected_base_sha: str) -> None:
    """Bind cwd, common Git metadata, branch, remote, and exact local HEAD."""
    resolved = root.resolve(strict=True)
    marker = resolved / ".git"
    if not marker.is_file():
        raise ArmingError(
            "armed mode must run at the root of the existing linked worktree"
        )
    line = bounded_read(marker, 4096).strip()
    if not line.startswith("gitdir: "):
        raise ArmingError("linked-worktree Git metadata is malformed")
    gitdir = Path(line.removeprefix("gitdir: ")).resolve(strict=True)
    common = (gitdir / bounded_read(gitdir / "commondir", 4096).strip()).resolve(
        strict=True
    )
    head = bounded_read(gitdir / "HEAD", 4096).strip()
    expected_ref = f"refs/heads/{EXPECTED_HEAD_REF}"
    if head != f"ref: {expected_ref}":
        raise ArmingError("armed mode must run from the existing PR #19 feature branch")
    ref_path = common / expected_ref
    actual_head = bounded_read(ref_path, 4096).strip() if ref_path.is_file() else ""
    if actual_head != expected_head:
        raise ArmingError(
            "local worktree HEAD does not match the exact reviewed PR head"
        )
    remote_ref = common / "refs" / "remotes" / EXPECTED_REMOTE / EXPECTED_BASE
    remote_main = bounded_read(remote_ref, 4096).strip() if remote_ref.is_file() else ""
    if remote_main != expected_base_sha:
        raise ArmingError("local origin/main does not match the exact release base SHA")
    parser = configparser.RawConfigParser(interpolation=None)
    try:
        parser.read_string(bounded_read(common / "config"))
        remote_url = parser.get(f'remote "{EXPECTED_REMOTE}"', "url")
    except (configparser.Error, KeyError) as exc:
        raise ArmingError("local Git remote metadata is malformed") from exc
    if remote_url != EXPECTED_REMOTE_URL:
        raise ArmingError("origin is not the fixed Titan atlas-iac HTTPS repository")


def preflight(
    request: ArmRequest, expected_repo: str, worktree: Path | None = None
) -> None:
    """Validate arming and local provenance before any subprocess/network call."""
    if request.confirmation != CONFIRMATION:
        raise ArmingError("confirmation phrase does not match")
    if expected_repo != EXPECTED_REPO or request.repo != EXPECTED_REPO:
        raise ArmingError(
            "armed repository is not the fixed Titan atlas-iac repository"
        )
    if (
        request.remote != EXPECTED_REMOTE
        or normalise_branch(request.base) != EXPECTED_BASE
    ):
        raise ArmingError("armed remote and pull-request base are fixed to origin/main")
    if not TOKEN_RE.fullmatch(request.token):
        raise ArmingError(
            "ephemeral token must be 8-64 lowercase alphanumeric or dash characters"
        )
    if not SHA_RE.fullmatch(request.expected_head) or not SHA_RE.fullmatch(
        request.expected_base_sha
    ):
        raise ArmingError("exact reviewed head and release base SHAs are required")
    assert_push_target_allowed(request.ref)
    _attest_worktree(
        worktree or Path.cwd(), request.expected_head, request.expected_base_sha
    )
    arm_ephemeral_policy(request.ref)