Make S1 handoff read-only by default

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a02994-7685-7940-bf34-3555b8256018

scripts/handoff_contract.py

@@ -0,0 +1,196 @@
+#!/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())

scripts/s1_handoff.py

@@ -15,6 +15,7 @@ from pathlib import Path
 from typing import Any
 
 from baseline_contract import load_spec, profile_hostvars, validate_repo_consumers
+from handoff_contract import validate_repository_surface
 from inventory_contract import load_inventory
 from s1_receipt import validate_receipt
 
@@ -110,6 +111,7 @@ def main() -> int:
     inventory = load_inventory(inventory_path)
     baseline = load_spec(spec_path)
     validate_repo_consumers(ROOT)
+    validate_repository_surface(ROOT)
     hosts = _selected_hosts(inventory, args.host)
     for host in hosts:
         profile_hostvars(baseline, host["baseline_profile"])