railiance-infra/scripts/handoff_contract.py

#!/usr/bin/env python3
"""Enforce the remote read-only contract of the S1 handoff playbook."""

from __future__ import annotations

import argparse
import json
import re
from pathlib import Path
from typing import Any

import yaml


REMOTE_MODULES = {
    "ansible.builtin.assert",
    "ansible.builtin.command",
    "ansible.builtin.stat",
}
CONTROLLER_MODULES = {
    "ansible.builtin.copy",
    "ansible.builtin.file",
}
TASK_CONTROL_KEYS = {
    "become",
    "changed_when",
    "delegate_to",
    "failed_when",
    "name",
    "register",
    "tags",
    "when",
}
EXPECTED_GOSS_ARGV = [
    "{{ goss_bin }}",
    "-g",
    "{{ goss_config }}",
    "validate",
    "--format",
    "tap",
]
EXPECTED_GOSS_COMMANDS = {
    "{{ baseline_firewall.verification.command }}",
    "ufw status | grep -Ec '6443/tcp[[:space:]]+ALLOW[[:space:]]+Anywhere' || true",
    "ufw status | grep -E '6443/tcp[[:space:]]+ALLOW' | grep -vc 'Anywhere' || true",
    "ufw status | grep -Ec '8472/udp[[:space:]]+ALLOW[[:space:]]+Anywhere' || true",
    "ufw status | grep -E '8472/udp[[:space:]]+ALLOW' | grep -vc 'Anywhere' || true",
    "ufw status | grep -Ec '6443/tcp[[:space:]]+ALLOW[[:space:]]+{{ src.address }}' || true",
    "grep NOPASSWD /etc/sudoers.d/{{ baseline_user.name }}",
    "grep -r HISTCONTROL /etc/profile.d/",
    "fail2ban-client status {{ jail }}",
    "test -x /usr/local/bin/age",
    "test -x /usr/local/bin/sops",
}
EXPECTED_PROFILE_COMMANDS = {"iptables -S INPUT", "ufw status"}
COMMAND_KEY_RE = re.compile(r'^  "(.+)":$', re.MULTILINE)


class HandoffContractError(ValueError):
    """The handoff playbook could change a managed host or execute arbitrary code."""


def _module(task: dict[str, Any], label: str) -> tuple[str, Any]:
    modules = [key for key in task if key.startswith("ansible.")]
    if len(modules) != 1:
        raise HandoffContractError(f"{label} must contain exactly one fully-qualified module")
    unknown = set(task) - TASK_CONTROL_KEYS - set(modules)
    if unknown:
        raise HandoffContractError(f"{label} has unsupported task keys: {', '.join(sorted(unknown))}")
    return modules[0], task[modules[0]]


def validate_payload(payload: Any) -> None:
    if not isinstance(payload, list) or len(payload) != 1 or not isinstance(payload[0], dict):
        raise HandoffContractError("handoff playbook must contain exactly one play")
    play = payload[0]
    if play.get("gather_facts") is not False:
        raise HandoffContractError("handoff playbook must set gather_facts: false")
    for forbidden in ("force_handlers", "handlers", "post_tasks", "pre_tasks", "roles"):
        if forbidden in play:
            raise HandoffContractError(f"handoff playbook must not declare {forbidden}")
    tasks = play.get("tasks")
    if not isinstance(tasks, list) or not tasks:
        raise HandoffContractError("handoff playbook must contain tasks")

    command_count = 0
    for index, task in enumerate(tasks, start=1):
        label = f"task {index}"
        if not isinstance(task, dict):
            raise HandoffContractError(f"{label} must be an object")
        module, arguments = _module(task, label)
        delegated = task.get("delegate_to") == "localhost"

        if delegated:
            if module not in CONTROLLER_MODULES:
                raise HandoffContractError(f"{label} uses unsupported controller module {module}")
            if task.get("become") is not False:
                raise HandoffContractError(f"{label} controller output must set become: false")
            if not isinstance(arguments, dict):
                raise HandoffContractError(f"{label} module arguments must be an object")
            destination = str(arguments.get("path", arguments.get("dest", "")))
            if "/reports" not in destination:
                raise HandoffContractError(f"{label} may write only beneath the controller reports path")
            if module == "ansible.builtin.file":
                if arguments.get("state") != "directory" or destination != "{{ playbook_dir }}/../../reports":
                    raise HandoffContractError(f"{label} may only ensure the reports directory")
            elif "src" in arguments or not destination.endswith(".tap"):
                raise HandoffContractError(f"{label} may record inline TAP evidence only")
            continue

        if module not in REMOTE_MODULES:
            raise HandoffContractError(f"{label} uses remote module {module}, which is not read-only")
        if module == "ansible.builtin.command":
            command_count += 1
            if not isinstance(arguments, dict) or arguments.get("argv") != EXPECTED_GOSS_ARGV:
                raise HandoffContractError(f"{label} command is not the fixed Goss validation argv")
            if task.get("changed_when") is not False or task.get("failed_when") is not False:
                raise HandoffContractError(
                    f"{label} command must set changed_when: false and failed_when: false"
                )

    if command_count != 1:
        raise HandoffContractError("handoff playbook must execute exactly one fixed Goss command")


def validate_playbook(path: Path) -> None:
    try:
        payload = yaml.safe_load(path.read_text(encoding="utf-8"))
    except (OSError, yaml.YAMLError) as exc:
        raise HandoffContractError(f"cannot read {path}: {exc}") from exc
    validate_payload(payload)


def validate_goss_commands(template_path: Path, baseline_path: Path) -> None:
    try:
        template = template_path.read_text(encoding="utf-8")
        baseline = yaml.safe_load(baseline_path.read_text(encoding="utf-8"))
    except (OSError, yaml.YAMLError) as exc:
        raise HandoffContractError(f"cannot read handoff command inputs: {exc}") from exc
    commands = set(COMMAND_KEY_RE.findall(template))
    if commands != EXPECTED_GOSS_COMMANDS:
        unexpected = sorted(commands - EXPECTED_GOSS_COMMANDS)
        missing = sorted(EXPECTED_GOSS_COMMANDS - commands)
        detail = []
        if unexpected:
            detail.append(f"unexpected: {unexpected}")
        if missing:
            detail.append(f"missing: {missing}")
        raise HandoffContractError("Goss command surface changed; " + "; ".join(detail))
    profiles = baseline.get("profiles", {}) if isinstance(baseline, dict) else {}
    profile_commands = {
        profile.get("firewall", {}).get("verification", {}).get("command")
        for profile in profiles.values()
        if isinstance(profile, dict)
    }
    if profile_commands != EXPECTED_PROFILE_COMMANDS:
        raise HandoffContractError(
            f"baseline profile command surface must be exactly {sorted(EXPECTED_PROFILE_COMMANDS)}"
        )


def validate_repository_surface(root: Path) -> None:
    validate_playbook(root / "ansible" / "playbooks" / "verify.yaml")
    validate_goss_commands(
        root / "goss" / "baseline.yaml.j2",
        root / "spec" / "server-baseline.yaml",
    )


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "playbook",
        nargs="?",
        type=Path,
        default=Path("ansible/playbooks/verify.yaml"),
    )
    args = parser.parse_args()
    playbook = args.playbook.resolve()
    root = playbook.parents[2]
    validate_repository_surface(root)
    print(
        json.dumps(
            {
                "goss_commands": len(EXPECTED_GOSS_COMMANDS) + len(EXPECTED_PROFILE_COMMANDS),
                "ok": True,
                "playbook": str(args.playbook),
                "remote_mutations": 0,
            }
        )
    )
    return 0


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