#!/usr/bin/env bash

# `boot --server` is deliberately public and must be safe before it reaches the
# cloned repository. Exercise its root branch through a PTY, against only a
# temporary filesystem and PATH adapters.

set -uo pipefail

repo_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
boot="$repo_dir/boot"

[[ -x "$boot" ]] || {
    printf 'root server bootstrap: %s is not executable\n' "$boot" >&2
    exit 1
}

python3 - "$boot" <<'PY'
import atexit
import errno
import fcntl
import os
import pty
import re
import select
import signal
import shutil
import stat as stat_module
import subprocess
import sys
import tempfile
import termios
import time
from pathlib import Path

boot = sys.argv[1]
work = Path(tempfile.mkdtemp())
atexit.register(shutil.rmtree, work, ignore_errors=True)
findings: list[str] = []
BOOT_REVISION = "0123456789abcdef0123456789abcdef01234567"
BOOT_SHA256 = subprocess.run(
    ["sha256sum", boot], check=True, capture_output=True, text=True
).stdout.split()[0]


def note(message: str) -> None:
    findings.append(message)


def write_executable(path: Path, contents: str) -> None:
    path.write_text(contents)
    path.chmod(0o755)


def generate_public_key(label: str) -> str:
    key_path = work / label
    subprocess.run(
        ["ssh-keygen", "-q", "-t", "ed25519", "-N", "", "-C", label, "-f", key_path],
        check=True,
    )
    public_key = key_path.with_suffix(".pub").read_text()
    key_path.unlink()
    key_path.with_suffix(".pub").unlink()
    return public_key


TARGET_PUBLIC_KEY = generate_public_key("panama-target-fixture")
ROOT_PUBLIC_KEY = generate_public_key("panama-root-fixture")


def make_stubs(stub_dir: Path, fixture_root: Path, calls: Path) -> None:
    common = f'''#!/usr/bin/env bash
set -u
calls={str(calls)!r}
log() {{
  local argument
  {{ for argument in "$@"; do printf '%q ' "$argument"; done; printf '\\n'; }} >>"$calls"
}}
consume_result() {{
  local name="$1" results result
  results="$PANAMA_BOOT_FIXTURE_ROOT/state/$name"
  if ! IFS= read -r result <"$results"; then
    return 0
  fi
  /usr/bin/tail -n +2 "$results" >"$results.next"
  /usr/bin/mv -f -- "$results.next" "$results"
  [[ "$result" =~ ^[0-9]+$ ]] || exit 97
  return "$result"
}}
'''

    write_executable(stub_dir / "id", common + r'''
log id "$@"
case "${1:-}" in
  -u)
    case "${2:-}" in
      '') printf '0\n' ;;
      root) printf '0\n' ;;
      gib) cat "$PANAMA_BOOT_FIXTURE_ROOT/state/target-uid" ;;
      *) exit 97 ;;
    esac
    ;;
  -nG) [[ "${2:-}" == gib ]] || exit 97; printf 'operators wheel\n' ;;
  *) exit 97 ;;
esac
''')
    write_executable(stub_dir / "passwd", common + r'''
log passwd "$@"
[[ "${1:-}" == -S && "${2:-}" == gib ]] || exit 97
printf 'gib PS\n'
''')
    write_executable(stub_dir / "getent", common + r'''
log getent "$@"
[[ "${1:-}" == passwd && "${2:-}" == gib ]] || exit 97
home="$(<"$PANAMA_BOOT_FIXTURE_ROOT/state/home")"
printf 'gib:x:1000:2000::%s:/bin/bash\n' "$home"
''')
    write_executable(stub_dir / "stat", common + r'''
log stat "$@"
[[ "${1:-}" == -Lc && "${2:-}" == '%u:%a' ]] || exit 97
case "${3:-}" in
  "$PANAMA_BOOT_FIXTURE_ROOT/home/gib/.ssh") cat "$PANAMA_BOOT_FIXTURE_ROOT/state/target-dir-meta" ;;
  "$PANAMA_BOOT_FIXTURE_ROOT/home/gib/.ssh/authorized_keys") cat "$PANAMA_BOOT_FIXTURE_ROOT/state/target-key-meta" ;;
  "$PANAMA_BOOT_FIXTURE_ROOT/root/.ssh/authorized_keys") cat "$PANAMA_BOOT_FIXTURE_ROOT/state/root-key-meta" ;;
  *) exit 97 ;;
esac
''')
    write_executable(stub_dir / "runuser", common + r'''
log runuser "$@"
[[ "${1:-}" == -u && "${2:-}" == gib && "${3:-}" == -- ]] || exit 97
target_user="$2"
shift 3
if [[ "${1:-}" == install && "${2:-}" == -d && "${3:-}" == -m && "${4:-}" == 0700 && "${5:-}" == -- ]]; then
  /usr/bin/install "${@:2}"
  printf '%s:700\n' "$(<"$PANAMA_BOOT_FIXTURE_ROOT/state/target-uid")" \
    >"$PANAMA_BOOT_FIXTURE_ROOT/state/target-dir-meta"
  exit 0
fi
if [[ "${1:-}" == install && "${2:-}" == -m && "${3:-}" == 0600 && "${4:-}" == -- ]]; then
  /usr/bin/install "${@:2}"
  printf '%s:600\n' "$(<"$PANAMA_BOOT_FIXTURE_ROOT/state/target-uid")" \
    >"$PANAMA_BOOT_FIXTURE_ROOT/state/target-key-meta"
  exit 0
fi
PANAMA_BOOT_TARGET_USER="$target_user" "$@"
''')
    write_executable(stub_dir / "ssh-keygen", common + r'''
log ssh-keygen "$@"
exec /usr/bin/ssh-keygen "$@"
''')
    write_executable(stub_dir / "git", common + r'''
if [[ "${PANAMA_BOOT_TARGET_USER:-}" != gib ]]; then
  log git-rejected-direct "$@"
  exit 96
fi
log git "$@"
case "${1:-}" in
  init)
    [[ "$#" -eq 2 ]] || exit 97
    mkdir -p "$2/.git"
    ;;
  -C)
    case "${3:-}" in
      remote)
        [[ "$#" -eq 6 && "$4" == add && "$5" == origin \
          && "$6" == https://git.gbrown.org/gib/Panama.git ]] || exit 97
        ;;
      fetch)
        [[ "$#" -eq 6 && "$4" == --depth=1 && "$5" == origin \
          && "$6" == "$PANAMA_BOOT_REVISION" ]] || exit 97
        ;;
      checkout)
        if [[ "${4:-}" == --detach ]]; then
          [[ "$#" -eq 5 && "$5" == "$PANAMA_BOOT_REVISION" ]] || exit 97
          cp "$PANAMA_BOOT_FIXTURE_ROOT/stub-install" "$2/install"
          chmod +x "$2/install"
        else
          [[ "$#" -eq 5 && "$4" == -b && "$5" == main ]] || exit 97
        fi
        ;;
      rev-parse)
        [[ "$#" -eq 4 && "$4" == 'HEAD^{commit}' ]] || exit 97
        printf '%s\n' "$PANAMA_BOOT_REVISION"
        ;;
      config)
        case "${4:-}:${5:-}:${6:-}" in
          branch.main.remote:origin:|branch.main.merge:refs/heads/main:) ;;
          *) exit 97 ;;
        esac
        ;;
      *) exit 97 ;;
    esac
    ;;
  *) exit 97 ;;
esac
''')
    write_executable(stub_dir / "dnf", common + r'''
log dnf "$@"
[[ "${1:-}" == install && "${2:-}" == -y && "${3:-}" == git ]] || exit 97
''')
    write_executable(stub_dir / "sshd", common + r'''
log sshd "$@"
case "${1:-}" in
  -t)
    [[ "$#" -eq 1 ]] || exit 97
    for artifact in "$PANAMA_BOOT_FIXTURE_ROOT/etc/ssh/sshd_config.d"/.00-panama.*; do
      [[ -e "$artifact" ]] || continue
      log ssh-artifact "$artifact" "$(/usr/bin/stat -c %a -- "$artifact")"
    done
    consume_result SSHD_RESULTS
    ;;
  -T)
    [[ "$#" -eq 3 && "$2" == -C ]] || exit 97
    case "$3" in
      user=root,host=localhost,addr=127.0.0.1)
        cat "$PANAMA_BOOT_FIXTURE_ROOT/state/ROOT_POLICY"
        ;;
      user=gib,host=localhost,addr=127.0.0.1)
        cat "$PANAMA_BOOT_FIXTURE_ROOT/state/TARGET_POLICY"
        ;;
      *) exit 97 ;;
    esac
    ;;
  *) exit 97 ;;
esac
''')
    write_executable(stub_dir / "systemctl", common + r'''
log systemctl "$@"
case "${1:-}:${2:-}" in
  cat:sshd.service) [[ "$(<"$PANAMA_BOOT_FIXTURE_ROOT/state/SSHD_UNIT")" == present ]] ;;
  cat:ssh.service) [[ "$(<"$PANAMA_BOOT_FIXTURE_ROOT/state/SSH_UNIT")" == present ]] ;;
  reload:sshd.service|reload:ssh.service) consume_result RELOAD_RESULTS ;;
  *) exit 97 ;;
esac
''')
    write_executable(stub_dir / "mktemp", common + r'''
log mktemp "$@"
artifact="$(/usr/bin/mktemp "$@")" || exit
case "$artifact" in
  *.tmp)
    if [[ -e "$PANAMA_BOOT_FIXTURE_ROOT/state/HOLD_CANDIDATE_PREPARATION" ]]; then
      : >"$PANAMA_BOOT_FIXTURE_ROOT/state/CANDIDATE_PREPARING"
      while [[ ! -e "$PANAMA_BOOT_FIXTURE_ROOT/state/RELEASE_CANDIDATE_PREPARATION" ]]; do
        /usr/bin/sleep 0.01
      done
    fi
    ;;
  *.backup)
    if [[ -e "$PANAMA_BOOT_FIXTURE_ROOT/state/HOLD_BACKUP_MKTEMP" ]]; then
      : >"$PANAMA_BOOT_FIXTURE_ROOT/state/BACKUP_MKTEMP_RUNNING"
      while [[ ! -e "$PANAMA_BOOT_FIXTURE_ROOT/state/RELEASE_BACKUP_MKTEMP" ]]; do
        /usr/bin/sleep 0.01
      done
    fi
    ;;
esac
printf '%s\n' "$artifact"
''')
    write_executable(stub_dir / "cp", common + r'''
log cp "$@"
destination="${@: -1}"
if [[ "$destination" == *.backup && -e "$PANAMA_BOOT_FIXTURE_ROOT/state/HOLD_BACKUP_PREPARATION" ]]; then
  : >"$PANAMA_BOOT_FIXTURE_ROOT/state/BACKUP_PREPARING"
  while [[ ! -e "$PANAMA_BOOT_FIXTURE_ROOT/state/RELEASE_BACKUP_PREPARATION" ]]; do
    /usr/bin/sleep 0.01
  done
fi
exec /usr/bin/cp "$@"
''')
    write_executable(stub_dir / "mv", common + r'''
log mv "$@" "source-mode=$(/usr/bin/stat -c %a -- "${3:-}")"
if [[ "${3:-}" == *.tmp && -e "$PANAMA_BOOT_FIXTURE_ROOT/state/HOLD_BEFORE_ACTIVATION" ]]; then
  : >"$PANAMA_BOOT_FIXTURE_ROOT/state/TRANSACTION_ARMED"
  while [[ ! -e "$PANAMA_BOOT_FIXTURE_ROOT/state/RELEASE_BEFORE_ACTIVATION" ]]; do
    /usr/bin/sleep 0.01
  done
  exit 98
fi
if [[ "${3:-}" == *.tmp ]]; then
  consume_result MV_ACTIVATION_RESULTS
  result=$?
  (( result == 0 )) || exit "$result"
fi
/usr/bin/mv "$@"
if [[ "${3:-}" == *.tmp && -e "$PANAMA_BOOT_FIXTURE_ROOT/state/HOLD_ACTIVATION" ]]; then
  : >"$PANAMA_BOOT_FIXTURE_ROOT/state/ACTIVATED"
  while [[ ! -e "$PANAMA_BOOT_FIXTURE_ROOT/state/RELEASE_ACTIVATION" ]]; do
    /usr/bin/sleep 0.01
  done
fi
''')
    write_executable(stub_dir / "rm", common + r'''
log rm "$@"
if [[ "${1:-}" == -f && "${2:-}" == -- && "${3:-}" == *.backup ]]; then
  consume_result RM_BACKUP_RESULTS
  result=$?
  (( result == 0 )) || exit "$result"
fi
if [[ "${1:-}" == -f && "${2:-}" == -- && "${3:-}" == *.tmp ]]; then
  consume_result RM_CANDIDATE_RESULTS
  result=$?
  (( result == 0 )) || exit "$result"
fi
if [[ "${1:-}" == -f && "${2:-}" == -- && "${3:-}" == */00-panama.conf ]]; then
  consume_result RM_DROPIN_RESULTS
  result=$?
  (( result == 0 )) || exit "$result"
fi
exec /usr/bin/rm "$@"
''')
    for command in ("chown", "useradd", "usermod"):
        write_executable(stub_dir / command, common + f'''\nlog {command} "$@"\nexit 97\n''')


def configure_case(
    name: str,
    *,
    sshd_results: tuple[int, ...] = (),
    reload_results: tuple[int, ...] = (),
    mv_activation_results: tuple[int, ...] = (),
    rm_backup_results: tuple[int, ...] = (),
    rm_candidate_results: tuple[int, ...] = (),
    rm_dropin_results: tuple[int, ...] = (),
    prior_dropin: bytes | None = None,
    prior_dropin_kind: str = "regular",
    root_policy: str = (
        "permitrootlogin no\n"
        "passwordauthentication no\n"
        "kbdinteractiveauthentication no\n"
    ),
    target_policy: str = (
        "passwordauthentication no\n"
        "kbdinteractiveauthentication no\n"
    ),
    sshd_unit: bool = True,
    ssh_unit: bool = True,
    hold_activation: bool = False,
    hold_before_activation: bool = False,
    hold_candidate_preparation: bool = False,
    hold_backup_preparation: bool = False,
) -> tuple[Path, Path]:
    fixture_root = work / name / "root"
    stub_dir = work / name / "bin"
    calls = fixture_root / "calls"
    state = fixture_root / "state"
    ssh_dir = fixture_root / "home/gib/.ssh"
    root_ssh_dir = fixture_root / "root/.ssh"
    (fixture_root / "etc/ssh/sshd_config.d").mkdir(parents=True)
    ssh_dir.mkdir(parents=True)
    root_ssh_dir.mkdir(parents=True)
    stub_dir.mkdir(parents=True)
    state.mkdir()
    calls.touch()
    (state / "target-uid").write_text("1000\n")
    (state / "home").write_text("/home/gib\n")
    (state / "target-dir-meta").write_text("1000:700\n")
    (state / "target-key-meta").write_text("1000:600\n")
    (state / "root-key-meta").write_text("0:600\n")
    (state / "SSHD_RESULTS").write_text("".join(f"{result}\n" for result in sshd_results))
    (state / "RELOAD_RESULTS").write_text("".join(f"{result}\n" for result in reload_results))
    (state / "MV_ACTIVATION_RESULTS").write_text(
        "".join(f"{result}\n" for result in mv_activation_results)
    )
    (state / "RM_BACKUP_RESULTS").write_text(
        "".join(f"{result}\n" for result in rm_backup_results)
    )
    (state / "RM_CANDIDATE_RESULTS").write_text(
        "".join(f"{result}\n" for result in rm_candidate_results)
    )
    (state / "RM_DROPIN_RESULTS").write_text(
        "".join(f"{result}\n" for result in rm_dropin_results)
    )
    (state / "ROOT_POLICY").write_text(root_policy)
    (state / "TARGET_POLICY").write_text(target_policy)
    (state / "SSHD_UNIT").write_text("present\n" if sshd_unit else "absent\n")
    (state / "SSH_UNIT").write_text("present\n" if ssh_unit else "absent\n")
    if hold_activation:
        (state / "HOLD_ACTIVATION").touch()
    if hold_before_activation:
        (state / "HOLD_BEFORE_ACTIVATION").touch()
    if hold_candidate_preparation:
        (state / "HOLD_CANDIDATE_PREPARATION").touch()
    if hold_backup_preparation:
        (state / "HOLD_BACKUP_PREPARATION").touch()
    (fixture_root / "stub-install").write_text(
        "#!/usr/bin/env bash\nprintf 'install-handoff %s\\n' \"${PANAMA_PATH:-unset}\" >> \"$PANAMA_BOOT_FIXTURE_ROOT/calls\"\n"
    )
    (fixture_root / "stub-install").chmod(0o755)
    make_stubs(stub_dir, fixture_root, calls)

    dropin = fixture_root / "etc/ssh/sshd_config.d/00-panama.conf"
    if prior_dropin_kind != "regular":
        if prior_dropin_kind == "symlink":
            symlink_target = fixture_root / "unsupported-panama-dropin"
            symlink_target.write_bytes(prior_dropin or b"unsupported symlink target\n")
            dropin.symlink_to(symlink_target)
        elif prior_dropin_kind == "directory":
            dropin.mkdir()
        elif prior_dropin_kind == "fifo":
            os.mkfifo(dropin)
        else:
            raise ValueError(prior_dropin_kind)
    elif prior_dropin is not None:
        dropin.write_bytes(prior_dropin)
        dropin.chmod(0o640)
        os.utime(dropin, ns=(1_700_000_000_123_456_789, 1_700_000_000_123_456_789))
        os.setxattr(dropin, b"user.panama-contract", b"preserve-me")
        subprocess.run(["setfacl", "-m", "u:65534:r--", dropin], check=True)

    target_keys = ssh_dir / "authorized_keys"
    root_keys = root_ssh_dir / "authorized_keys"
    if name == "missing":
        pass
    elif name == "empty":
        target_keys.touch()
    elif name == "comment-only":
        target_keys.write_text("# no usable key\n\n")
    elif name == "ssh-directory-symlink":
        shutil.rmtree(ssh_dir)
        alternate = fixture_root / "unsafe-ssh"
        alternate.mkdir()
        (fixture_root / "home/gib/.ssh").symlink_to(alternate)
    elif name == "authorized-keys-symlink":
        alternate = fixture_root / "unsafe-authorized-keys"
        alternate.write_text(TARGET_PUBLIC_KEY)
        target_keys.symlink_to(alternate)
    elif name == "malformed-key":
        target_keys.write_text("this is not OpenSSH key material\n")
    elif name == "mixed-valid-and-malformed-key":
        target_keys.write_text(TARGET_PUBLIC_KEY + "this is not OpenSSH key material\n")
    elif name == "malformed-root-key":
        root_keys.write_text("this is not OpenSSH key material\n")
    elif name == "directory-wrong-mode":
        target_keys.write_text(TARGET_PUBLIC_KEY)
        (state / "target-dir-meta").write_text("1000:755\n")
    elif name == "root-copy-directory-wrong-mode":
        root_keys.write_text(ROOT_PUBLIC_KEY)
        (state / "target-dir-meta").write_text("1000:755\n")
    elif name == "file-wrong-mode":
        target_keys.write_text(TARGET_PUBLIC_KEY)
        (state / "target-key-meta").write_text("1000:644\n")
    elif name == "directory-wrong-owner":
        target_keys.write_text(TARGET_PUBLIC_KEY)
        (state / "target-dir-meta").write_text("0:700\n")
    elif name == "file-wrong-owner":
        target_keys.write_text(TARGET_PUBLIC_KEY)
        (state / "target-key-meta").write_text("0:600\n")
    elif name == "root-target-account":
        target_keys.write_text(TARGET_PUBLIC_KEY)
        (state / "target-uid").write_text("0\n")
    elif name == "relative-home":
        target_keys.write_text(TARGET_PUBLIC_KEY)
        (state / "home").write_text("home/gib\n")
    elif name in (
        "safe-existing-key",
        "declines-hardening",
        "missing-ssh-unit",
        "preexisting-dropin-symlink",
        "preexisting-dropin-directory",
        "preexisting-dropin-fifo",
        "success-without-prior-dropin",
        "success-replaces-prior-dropin",
        "candidate-invalid",
        "candidate-invalid-without-prior",
        "candidate-reload-fails",
        "candidate-reload-fails-without-prior",
        "rollback-validation-fails",
        "rollback-reload-fails",
        "rollback-removal-fails-without-prior",
        "effective-root-policy-conflict",
        "effective-target-policy-conflict",
        "success-backup-cleanup-fails",
        "rollback-backup-cleanup-fails",
        "signal-int-restores-prior",
        "signal-term-removes-new-dropin",
        "candidate-cleanup-fails",
        "signal-int-before-activation-prior",
        "signal-term-before-activation-no-prior",
        "signal-int-before-activation-cleanup-fails",
        "signal-int-during-candidate-preparation",
        "signal-term-during-backup-preparation",
        "direct-root-git-probe",
    ):
        target_keys.write_text(TARGET_PUBLIC_KEY)
    elif name == "safe-root-key-copy":
        shutil.rmtree(ssh_dir)
        (state / "target-dir-meta").write_text("missing\n")
        (state / "target-key-meta").write_text("missing\n")
        root_keys.write_text(ROOT_PUBLIC_KEY)
    else:
        raise ValueError(name)
    return fixture_root, stub_dir


def assert_checkout_runs_as_target(name: str, calls: str, fixture_root: Path) -> None:
    call_lines = calls.splitlines()
    git_indices = [
        index for index, line in enumerate(call_lines) if line.startswith("git ")
    ]
    if not git_indices:
        return

    checkout = fixture_root / "home/gib/.local/share/Panama"
    expected_mkdir = f"runuser -u gib -- mkdir -p {checkout.parent} "
    if expected_mkdir not in call_lines:
        note(f"{name}: checkout parent directory was not created as the target user")

    for index in git_indices:
        expected_runuser = f"runuser -u gib -- {call_lines[index]}"
        if index == 0 or call_lines[index - 1] != expected_runuser:
            note(f"{name}: checkout Git operation bypassed the selected target user")
            break

    if any(line.startswith("git-rejected-direct ") for line in call_lines):
        note(f"{name}: checkout attempted a direct root Git operation")


def run_case(
    name: str,
    *,
    signal_after_activation: int | None = None,
    signal_before_activation: int | None = None,
    signal_during_candidate_preparation: int | None = None,
    signal_during_backup_preparation: int | None = None,
    harden_answer: str = "Y",
    prior_traps: bool = False,
    **configuration: object,
) -> tuple[int, str, str, Path, int]:
    fixture_root, stub_dir = configure_case(
        name,
        hold_activation=signal_after_activation is not None,
        hold_before_activation=signal_before_activation is not None,
        hold_candidate_preparation=signal_during_candidate_preparation is not None,
        hold_backup_preparation=signal_during_backup_preparation is not None,
        **configuration,
    )
    master, slave = pty.openpty()

    def attach_terminal() -> None:
        # The test runner launches contracts as background jobs, which inherit
        # SIGINT ignored. A real interactive bootstrap starts with SIGINT at
        # its default disposition, so restore that state before exec.
        signal.signal(signal.SIGINT, signal.SIG_DFL)
        fcntl.ioctl(0, termios.TIOCSCTTY, 0)

    env = {
        **os.environ,
        "PATH": f"{stub_dir}:/usr/bin:/bin",
        "PANAMA_BOOT_FIXTURE_ROOT": str(fixture_root),
        "PANAMA_BOOT_REVISION": BOOT_REVISION,
        "PANAMA_BOOT_SHA256": BOOT_SHA256,
        "PANAMA_PATH": f"{fixture_root}/home/gib/.local/share/Panama",
        "HOME": f"{fixture_root}/root",
    }
    if prior_traps:
        bash_env = fixture_root / "prior-traps"
        bash_env.write_text(
            '''if [[ "$0" == "$PANAMA_BOOT_SCRIPT" ]]; then
trap 'printf "prior-exit %s\\n" "$BASHPID" >>"$PANAMA_BOOT_FIXTURE_ROOT/calls"' EXIT
trap 'printf "prior-int %s\\n" "$BASHPID" >>"$PANAMA_BOOT_FIXTURE_ROOT/calls"' INT
trap 'printf "prior-term %s\\n" "$BASHPID" >>"$PANAMA_BOOT_FIXTURE_ROOT/calls"' TERM
fi
'''
        )
        env["BASH_ENV"] = str(bash_env)
        env["PANAMA_BOOT_SCRIPT"] = boot
    process = subprocess.Popen(
        ["bash", boot, "--server"],
        stdin=slave,
        stdout=slave,
        stderr=slave,
        env=env,
        start_new_session=True,
        preexec_fn=attach_terminal,
    )
    os.close(slave)
    os.write(master, f"gib\n{harden_answer}\n".encode())
    if signal_during_candidate_preparation is not None:
        marker = fixture_root / "state/CANDIDATE_PREPARING"
        deadline = time.monotonic() + 5
        while not marker.exists() and process.poll() is None and time.monotonic() < deadline:
            time.sleep(0.01)
        if not marker.exists():
            note(f"{name}: fixture did not observe candidate preparation before signaling")
        else:
            os.kill(process.pid, signal_during_candidate_preparation)
        (fixture_root / "state/RELEASE_CANDIDATE_PREPARATION").touch()
    elif signal_during_backup_preparation is not None:
        marker = fixture_root / "state/BACKUP_PREPARING"
        deadline = time.monotonic() + 5
        while not marker.exists() and process.poll() is None and time.monotonic() < deadline:
            time.sleep(0.01)
        if not marker.exists():
            note(f"{name}: fixture did not observe backup preparation before signaling")
        else:
            os.kill(process.pid, signal_during_backup_preparation)
        (fixture_root / "state/RELEASE_BACKUP_PREPARATION").touch()
    elif signal_before_activation is not None:
        armed = fixture_root / "state/TRANSACTION_ARMED"
        deadline = time.monotonic() + 5
        while not armed.exists() and process.poll() is None and time.monotonic() < deadline:
            time.sleep(0.01)
        if not armed.exists():
            note(f"{name}: fixture did not observe transaction arming before signaling")
        else:
            os.kill(process.pid, signal_before_activation)
        (fixture_root / "state/RELEASE_BEFORE_ACTIVATION").touch()
    elif signal_after_activation is not None:
        activation = fixture_root / "state/ACTIVATED"
        deadline = time.monotonic() + 5
        while not activation.exists() and process.poll() is None and time.monotonic() < deadline:
            time.sleep(0.01)
        if not activation.exists():
            note(f"{name}: fixture did not observe atomic activation before signaling")
        else:
            os.kill(process.pid, signal_after_activation)
        (fixture_root / "state/RELEASE_ACTIVATION").touch()
    chunks: list[bytes] = []
    deadline = time.monotonic() + 8
    timed_out = False
    while True:
        readable, _, _ = select.select([master], [], [], 0.1)
        if not readable:
            if process.poll() is not None:
                break
            if time.monotonic() >= deadline:
                timed_out = True
                os.killpg(process.pid, signal.SIGKILL)
                process.wait()
                continue
            continue
        try:
            chunk = os.read(master, 4096)
        except OSError as error:
            if error.errno == errno.EIO:
                break
            raise
        if not chunk:
            break
        chunks.append(chunk)
    os.close(master)
    status = process.wait()
    if timed_out:
        note(f"{name}: bootstrap timed out, likely while reading an unsupported object")
    calls = (fixture_root / "calls").read_text()
    output = b"".join(chunks).decode(errors="replace")
    assert_checkout_runs_as_target(name, calls, fixture_root)
    return status, output, calls, fixture_root, process.pid


guard_root = work / "actual-root-fixture-guard"
guard_root.mkdir()
guard_env = {
    **os.environ,
    "PANAMA_BOOT_FIXTURE_ROOT": str(guard_root),
    "PANAMA_BOOT_REVISION": BOOT_REVISION,
    "PANAMA_BOOT_SHA256": BOOT_SHA256,
    "HOME": str(guard_root),
}
if os.geteuid() == 0:
    guard_command = ["bash", boot, "--server"]
else:
    guard_command = ["unshare", "--user", "--map-root-user", "--", "bash", boot, "--server"]
try:
    guard_result = subprocess.run(
        guard_command,
        env=guard_env,
        capture_output=True,
        text=True,
        timeout=5,
    )
except (FileNotFoundError, subprocess.TimeoutExpired) as error:
    note(f"actual-root-fixture-guard: could not create a hermetic root process: {error}")
else:
    if guard_result.returncode != 1:
        note(
            "actual-root-fixture-guard: actual root did not reject "
            f"PANAMA_BOOT_FIXTURE_ROOT with status 1: {guard_result.returncode}"
        )
    if "PANAMA_BOOT_FIXTURE_ROOT is test-only" not in guard_result.stderr:
        note("actual-root-fixture-guard: rejection diagnostic was missing")
    if any(guard_root.iterdir()):
        note("actual-root-fixture-guard: boot mutated its rejected fixture root")


probe_root, probe_stub_dir = configure_case("direct-root-git-probe")
probe_checkout = probe_root / "home/gib/.local/share/Panama"
probe_env = {
    **os.environ,
    "PATH": f"{probe_stub_dir}:/usr/bin:/bin",
    "PANAMA_BOOT_FIXTURE_ROOT": str(probe_root),
    "PANAMA_BOOT_REVISION": BOOT_REVISION,
}
direct_git = subprocess.run(
    [str(probe_stub_dir / "git"), "init", str(probe_checkout)],
    env=probe_env,
    capture_output=True,
    text=True,
)
probe_calls = (probe_root / "calls").read_text()
if direct_git.returncode != 96:
    note("direct-root-git-probe: Git adapter accepted a root-owned checkout call")
if (probe_checkout / ".git").exists():
    note("direct-root-git-probe: rejected root-owned Git call mutated the checkout")
if "git-rejected-direct init " not in probe_calls:
    note("direct-root-git-probe: fixture did not exercise the direct Git rejection")


unsafe_cases = (
    "missing",
    "empty",
    "comment-only",
    "malformed-key",
    "mixed-valid-and-malformed-key",
    "malformed-root-key",
    "ssh-directory-symlink",
    "authorized-keys-symlink",
    "directory-wrong-mode",
    "root-copy-directory-wrong-mode",
    "file-wrong-mode",
    "directory-wrong-owner",
    "file-wrong-owner",
    "root-target-account",
    "relative-home",
)
for case in unsafe_cases:
    status, output, calls, fixture_root, _ = run_case(case)
    if status != 0:
        note(f"{case}: bootstrap stopped with status {status}: {output.strip()}")
    if "SSH hardening unavailable" not in output:
        note(f"{case}: unsafe login path did not explain why hardening was unavailable")
    if "sshd -t" in calls:
        note(f"{case}: unsafe login path validated sshd")
    if "systemctl reload" in calls:
        note(f"{case}: unsafe login path reloaded SSH")
    if (fixture_root / "etc/ssh/sshd_config.d/00-panama.conf").exists():
        note(f"{case}: unsafe login path changed the SSH drop-in")
    if case == "root-copy-directory-wrong-mode" and (
        fixture_root / "home/gib/.ssh/authorized_keys"
    ).exists():
        note("root-copy-directory-wrong-mode: copied a root key into an unsafe SSH directory")
    if "install-handoff " not in calls:
        note(f"{case}: unsafe login path did not hand off to install")

desired_dropin = (
    b"PermitRootLogin no\n"
    b"PasswordAuthentication no\n"
    b"KbdInteractiveAuthentication no\n"
)
prior_dropin = b"# prior Panama settings\nPasswordAuthentication yes\n"

for case in ("safe-existing-key", "safe-root-key-copy"):
    status, output, calls, fixture_root, _ = run_case(case)
    if status != 0:
        note(f"{case}: safe login path stopped with status {status}: {output.strip()}")
    if "SSH hardening unavailable" in output:
        note(f"{case}: safe login path was rejected: {output.strip()} | {calls.strip()}")
    if "systemctl reload" not in calls:
        note(f"{case}: safe login path did not reach SSH hardening")
    if "install-handoff " not in calls:
        note(f"{case}: safe login path did not hand off to install")
    dropin = fixture_root / "etc/ssh/sshd_config.d/00-panama.conf"
    if (dropin.read_bytes() if dropin.exists() else None) != desired_dropin:
        note(f"{case}: safe login path did not write the expected SSH drop-in")
    if case == "safe-root-key-copy":
        keys = fixture_root / "home/gib/.ssh/authorized_keys"
        ssh_dir = keys.parent
        if not keys.exists() or keys.read_text() != ROOT_PUBLIC_KEY:
            note("safe-root-key-copy: root key was not copied to the target account")
        if keys.exists() and (
            ssh_dir.stat().st_mode & 0o777 != 0o700
            or keys.stat().st_mode & 0o777 != 0o600
        ):
            note("safe-root-key-copy: destination modes were not normalized to 0700/0600")
        call_lines = calls.splitlines()
        install_dir = (
            f"runuser -u gib -- install -d -m 0700 -- {ssh_dir} "
        )
        install_key = (
            "runuser -u gib -- install -m 0600 -- "
            f"/dev/stdin {keys} "
        )
        if install_dir not in call_lines or install_key not in call_lines:
            note("safe-root-key-copy: destination creation and writing did not run as the target user")
        else:
            validation_indices = [
                index
                for index, line in enumerate(call_lines)
                if line.startswith("ssh-keygen -l -f ")
            ]
            if not validation_indices or max(validation_indices) < call_lines.index(install_key):
                note("safe-root-key-copy: copied key validity was not rechecked after installation")
        if any(line.startswith("chown ") for line in call_lines):
            note("safe-root-key-copy: bootstrap still assumes the primary group matches the username")

status, output, calls, fixture_root, _ = run_case(
    "declines-hardening",
    harden_answer="n",
)
declined_dropin = fixture_root / "etc/ssh/sshd_config.d/00-panama.conf"
if status != 0 or "install-handoff " not in calls:
    note("declines-hardening: declining did not continue to install")
if declined_dropin.exists() or "sshd " in calls or "systemctl reload " in calls:
    note("declines-hardening: declining changed or validated SSH state")

status, output, calls, fixture_root, _ = run_case(
    "missing-ssh-unit",
    sshd_unit=False,
    ssh_unit=False,
)
missing_unit_dropin = fixture_root / "etc/ssh/sshd_config.d/00-panama.conf"
if status != 0 or "install-handoff " not in calls:
    note("missing-ssh-unit: unavailable hardening did not continue to install")
if "SSH hardening unavailable" not in output:
    note("missing-ssh-unit: missing units did not explain that hardening was unavailable")
if missing_unit_dropin.exists() or "sshd " in calls or "systemctl reload " in calls:
    note("missing-ssh-unit: unavailable hardening changed or validated SSH state")

unsupported_dropins = {
    "preexisting-dropin-symlink": "symlink",
    "preexisting-dropin-directory": "directory",
    "preexisting-dropin-fifo": "fifo",
}
for case, kind in unsupported_dropins.items():
    status, output, calls, fixture_root, _ = run_case(
        case,
        prior_dropin=prior_dropin,
        prior_dropin_kind=kind,
    )
    dropin = fixture_root / "etc/ssh/sshd_config.d/00-panama.conf"
    if status != 0 or "install-handoff " not in calls:
        note(f"{case}: unsupported object did not continue to install")
    if "SSH hardening unavailable" not in output:
        note(f"{case}: unsupported object did not explain that hardening was unavailable")
    if "sshd " in calls or "systemctl reload " in calls:
        note(f"{case}: unsupported object reached SSH validation or reload")
    if kind == "symlink" and not dropin.is_symlink():
        note(f"{case}: pre-existing symlink was changed")
    if kind == "directory" and not dropin.is_dir():
        note(f"{case}: pre-existing directory was changed")
    if kind == "fifo" and not stat_module.S_ISFIFO(dropin.lstat().st_mode):
        note(f"{case}: pre-existing FIFO was changed")


def regular_metadata(path: Path) -> tuple[object, ...]:
    metadata = path.stat()
    xattrs = tuple((name, os.getxattr(path, name)) for name in sorted(os.listxattr(path)))
    acl = subprocess.check_output(["getfacl", "-cp", path])
    return (
        stat_module.S_IMODE(metadata.st_mode),
        metadata.st_uid,
        metadata.st_gid,
        metadata.st_mtime_ns,
        xattrs,
        acl,
    )


metadata_reference = work / "prior-dropin-metadata-reference"
metadata_reference.write_bytes(prior_dropin)
metadata_reference.chmod(0o640)
os.utime(
    metadata_reference,
    ns=(1_700_000_000_123_456_789, 1_700_000_000_123_456_789),
)
os.setxattr(metadata_reference, b"user.panama-contract", b"preserve-me")
subprocess.run(["setfacl", "-m", "u:65534:r--", metadata_reference], check=True)
expected_prior_metadata = regular_metadata(metadata_reference)


transaction_cases = {
    "success-without-prior-dropin": {
        "sshd_results": (0,),
        "reload_results": (0,),
        "prior_dropin": None,
        "sshd_unit": True,
        "ssh_unit": True,
        "succeeds": True,
    },
    "success-replaces-prior-dropin": {
        "sshd_results": (0,),
        "reload_results": (0,),
        "prior_dropin": prior_dropin,
        "sshd_unit": False,
        "ssh_unit": True,
        "succeeds": True,
    },
    "candidate-invalid": {
        "sshd_results": (1, 0),
        "reload_results": (),
        "prior_dropin": prior_dropin,
        "sshd_unit": True,
        "ssh_unit": True,
        "succeeds": False,
    },
    "candidate-invalid-without-prior": {
        "sshd_results": (1, 0),
        "reload_results": (),
        "prior_dropin": None,
        "sshd_unit": True,
        "ssh_unit": True,
        "succeeds": False,
    },
    "candidate-reload-fails": {
        "sshd_results": (0, 0),
        "reload_results": (1, 0),
        "prior_dropin": prior_dropin,
        "sshd_unit": True,
        "ssh_unit": True,
        "succeeds": False,
    },
    "candidate-reload-fails-without-prior": {
        "sshd_results": (0, 0),
        "reload_results": (1, 0),
        "prior_dropin": None,
        "sshd_unit": True,
        "ssh_unit": True,
        "succeeds": False,
    },
    "rollback-validation-fails": {
        "sshd_results": (0, 1),
        "reload_results": (1,),
        "prior_dropin": prior_dropin,
        "sshd_unit": True,
        "ssh_unit": True,
        "succeeds": False,
        "rollback_fails": True,
    },
    "rollback-reload-fails": {
        "sshd_results": (0, 0),
        "reload_results": (1, 1),
        "prior_dropin": prior_dropin,
        "sshd_unit": True,
        "ssh_unit": True,
        "succeeds": False,
        "rollback_fails": True,
    },
    "rollback-removal-fails-without-prior": {
        "sshd_results": (1, 1),
        "reload_results": (),
        "rm_dropin_results": (1,),
        "prior_dropin": None,
        "sshd_unit": True,
        "ssh_unit": True,
        "succeeds": False,
        "rollback_fails": True,
        "settled_dropin": desired_dropin,
    },
    "effective-root-policy-conflict": {
        "sshd_results": (0, 0),
        "reload_results": (),
        "prior_dropin": prior_dropin,
        "root_policy": (
            "permitrootlogin yes\n"
            "passwordauthentication no\n"
            "kbdinteractiveauthentication no\n"
        ),
        "sshd_unit": True,
        "ssh_unit": True,
        "succeeds": False,
    },
    "effective-target-policy-conflict": {
        "sshd_results": (0, 0),
        "reload_results": (),
        "prior_dropin": prior_dropin,
        "target_policy": (
            "passwordauthentication no\n"
            "kbdinteractiveauthentication yes\n"
        ),
        "sshd_unit": True,
        "ssh_unit": True,
        "succeeds": False,
    },
}

for case, expected in transaction_cases.items():
    configuration = {
        key: value
        for key, value in expected.items()
        if key not in {"succeeds", "rollback_fails", "settled_dropin"}
    }
    status, output, calls, fixture_root, _ = run_case(case, **configuration)
    call_lines = calls.splitlines()
    dropin = fixture_root / "etc/ssh/sshd_config.d/00-panama.conf"
    sshd_dir = dropin.parent
    validations = [index for index, line in enumerate(call_lines) if line.startswith("sshd -t ")]
    reloads = [
        index
        for index, line in enumerate(call_lines)
        if line.startswith("systemctl reload ")
    ]
    activation_lines = [
        (index, line)
        for index, line in enumerate(call_lines)
        if re.fullmatch(
            rf"mv -f -- {re.escape(str(sshd_dir))}/\.00-panama\.[A-Za-z0-9]+\.tmp "
            rf"{re.escape(str(dropin))} source-mode=600 ",
            line,
        )
    ]
    restore_lines = [
        index
        for index, line in enumerate(call_lines)
        if re.fullmatch(
            rf"mv -f -- {re.escape(str(sshd_dir))}/\.00-panama\.[A-Za-z0-9]+\.restore "
            rf"{re.escape(str(dropin))} source-mode=640 ",
            line,
        )
    ]
    removal_lines = [
        index
        for index, line in enumerate(call_lines)
        if line == f"rm -f -- {dropin} "
    ]
    rollback_lines = restore_lines if expected["prior_dropin"] is not None else removal_lines

    succeeds = bool(expected["succeeds"])
    if succeeds and status != 0:
        note(f"{case}: transaction stopped with status {status}: {output.strip()}")
    if not succeeds and status == 0:
        note(f"{case}: failed transaction returned success")
    if succeeds and "install-handoff " not in calls:
        note(f"{case}: successful transaction did not hand off to install")
    if not succeeds and "install-handoff " in calls:
        note(f"{case}: failed transaction handed off to install")

    wanted_contents = (
        desired_dropin
        if succeeds
        else expected.get("settled_dropin", expected["prior_dropin"])
    )
    actual_contents = dropin.read_bytes() if dropin.exists() else None
    if actual_contents != wanted_contents:
        note(f"{case}: SSH drop-in contents were not {'activated' if succeeds else 'restored'}")

    if len(activation_lines) != 1:
        note(f"{case}: candidate was not activated once through a restrictive same-directory rename")
    root_policy_lines = [
        index
        for index, line in enumerate(call_lines)
        if line == r"sshd -T -C user=root\,host=localhost\,addr=127.0.0.1 "
    ]
    target_policy_lines = [
        index
        for index, line in enumerate(call_lines)
        if line == r"sshd -T -C user=gib\,host=localhost\,addr=127.0.0.1 "
    ]
    if case in {
        "candidate-invalid",
        "candidate-invalid-without-prior",
        "rollback-removal-fails-without-prior",
    }:
        expected_policy_users: tuple[str, ...] = ()
    elif case == "effective-root-policy-conflict":
        expected_policy_users = ("root",)
    else:
        expected_policy_users = ("root", "gib")
    if len(root_policy_lines) != (1 if "root" in expected_policy_users else 0):
        note(f"{case}: effective root policy validation count was wrong")
    if len(target_policy_lines) != (1 if "gib" in expected_policy_users else 0):
        note(f"{case}: effective target policy validation count was wrong")

    if succeeds:
        if len(validations) != 1 or len(reloads) != 1:
            note(f"{case}: success did not validate once and reload once")
        elif root_policy_lines and target_policy_lines and activation_lines and not (
            activation_lines[0][0]
            < validations[0]
            < root_policy_lines[0]
            < target_policy_lines[0]
            < reloads[0]
        ):
            note(f"{case}: success did not activate, validate syntax and effective policy, then reload")
    elif case in {
        "candidate-invalid",
        "candidate-invalid-without-prior",
        "rollback-removal-fails-without-prior",
        "effective-root-policy-conflict",
        "effective-target-policy-conflict",
    }:
        if len(validations) != 2 or reloads:
            note(f"{case}: rejected candidate did not validate candidate and restoration without reload")
        elif activation_lines and rollback_lines:
            policy_order = [
                *root_policy_lines,
                *target_policy_lines,
            ]
            if not (
                activation_lines[0][0]
                < validations[0]
                < (policy_order[0] if policy_order else rollback_lines[0])
                and all(
                    left < right
                    for left, right in zip(policy_order, [*policy_order[1:], rollback_lines[0]])
                )
                and rollback_lines[0] < validations[1]
            ):
                note(f"{case}: rollback command order was wrong")
    else:
        if len(validations) != 2 or len(reloads) != 2:
            note(f"{case}: reload failure did not validate and reload the restored configuration")
        elif activation_lines and rollback_lines and root_policy_lines and target_policy_lines and not (
            activation_lines[0][0]
            < validations[0]
            < root_policy_lines[0]
            < target_policy_lines[0]
            < reloads[0]
            < rollback_lines[0]
            < validations[1]
            < reloads[1]
        ):
            note(f"{case}: rollback command order was wrong")

    if succeeds and restore_lines:
        note(f"{case}: successful transaction performed a rollback")
    if not succeeds:
        if len(rollback_lines) != 1:
            note(f"{case}: pre-transaction SSH state was not restored exactly once")
        if expected["prior_dropin"] is not None and dropin.exists():
            if regular_metadata(dropin) != expected_prior_metadata:
                note(f"{case}: rollback did not restore complete regular-file metadata")

    detected_unit = "ssh.service" if case == "success-replaces-prior-dropin" else "sshd.service"
    other_unit = "sshd.service" if detected_unit == "ssh.service" else "ssh.service"
    reload_lines = [call_lines[index] for index in reloads]
    if reload_lines and any(line != f"systemctl reload {detected_unit} " for line in reload_lines):
        note(f"{case}: reloaded a unit other than detected {detected_unit}")
    if any(line == f"systemctl reload {other_unit} " for line in call_lines):
        note(f"{case}: guessed {other_unit} after reload failure")
    if detected_unit == "sshd.service":
        if "systemctl cat sshd.service " not in call_lines:
            note(f"{case}: did not detect sshd.service")
        if "systemctl cat ssh.service " in call_lines:
            note(f"{case}: probed ssh.service after finding sshd.service")
    elif not (
        "systemctl cat sshd.service " in call_lines
        and "systemctl cat ssh.service " in call_lines
        and call_lines.index("systemctl cat sshd.service ")
        < call_lines.index("systemctl cat ssh.service ")
    ):
        note(f"{case}: did not fall back from absent sshd.service to ssh.service")

    artifacts = list(sshd_dir.glob(".00-panama.*"))
    rollback_fails = bool(expected.get("rollback_fails", False))
    if not rollback_fails and artifacts:
        note(f"{case}: successful or cleanly rolled-back transaction left temporary artifacts")
    if rollback_fails:
        backups = [artifact for artifact in artifacts if artifact.name.endswith(".backup")]
        if expected["prior_dropin"] is None:
            if backups:
                note(f"{case}: no-prior-file recovery retained a nonexistent backup")
            expected_remove = f"rm -f -- {dropin.resolve()}"
            if expected_remove not in output or "cp -a --" in output:
                note(f"{case}: no-prior-file recovery did not instruct removal of the installed drop-in")
        elif len(backups) != 1:
            note(f"{case}: rollback failure did not retain exactly one backup")
        else:
            backup = backups[0]
            if backup.parent != sshd_dir or backup.read_bytes() != prior_dropin:
                note(f"{case}: retained backup was not a same-directory copy")
            if regular_metadata(backup) != expected_prior_metadata:
                note(f"{case}: retained backup did not preserve complete regular-file metadata")
            if str(backup.resolve()) not in output:
                note(f"{case}: recovery output omitted the absolute backup path")
        if "sshd -t" not in output or f"systemctl reload {detected_unit}" not in output:
            note(f"{case}: recovery output omitted validation or reload commands")

    artifact_logs = [line for line in call_lines if line.startswith("ssh-artifact ")]
    if expected["prior_dropin"] is not None and not any(
        re.fullmatch(
            rf"ssh-artifact {re.escape(str(sshd_dir))}/\.00-panama\.[A-Za-z0-9]+\.backup 640 ",
            line,
        )
        for line in artifact_logs
    ):
        note(f"{case}: backup was not collision-safe, same-directory, non-.conf, and restrictive")

cleanup_failure_cases = {
    "success-backup-cleanup-fails": {
        "sshd_results": (0,),
        "reload_results": (0,),
        "rm_backup_results": (1,),
        "expected_dropin": desired_dropin,
        "expected_validations": 1,
        "expected_reloads": 1,
    },
    "rollback-backup-cleanup-fails": {
        "sshd_results": (1, 0),
        "reload_results": (),
        "rm_backup_results": (1,),
        "expected_dropin": prior_dropin,
        "expected_validations": 2,
        "expected_reloads": 0,
    },
}

for case, expected in cleanup_failure_cases.items():
    status, output, calls, fixture_root, _ = run_case(
        case,
        sshd_results=expected["sshd_results"],
        reload_results=expected["reload_results"],
        rm_backup_results=expected["rm_backup_results"],
        prior_dropin=prior_dropin,
    )
    dropin = fixture_root / "etc/ssh/sshd_config.d/00-panama.conf"
    backups = list(dropin.parent.glob(".00-panama.*.backup"))
    if status == 0:
        note(f"{case}: cleanup failure returned success")
    if "install-handoff " in calls:
        note(f"{case}: cleanup failure handed off to install")
    if dropin.read_bytes() != expected["expected_dropin"]:
        note(f"{case}: cleanup failure changed the settled SSH drop-in")
    if calls.count("sshd -t \n") != expected["expected_validations"]:
        note(f"{case}: cleanup failure validation count was wrong")
    if calls.count("systemctl reload sshd.service \n") != expected["expected_reloads"]:
        note(f"{case}: cleanup failure reload count was wrong")
    expected_policy_checks = 1 if case == "success-backup-cleanup-fails" else 0
    if calls.count("sshd -T -C user=root\\,host=localhost\\,addr=127.0.0.1 \n") != expected_policy_checks:
        note(f"{case}: cleanup failure root policy validation count was wrong")
    if calls.count("sshd -T -C user=gib\\,host=localhost\\,addr=127.0.0.1 \n") != expected_policy_checks:
        note(f"{case}: cleanup failure target policy validation count was wrong")
    if len(backups) != 1:
        note(f"{case}: failed cleanup did not retain exactly one backup")
    else:
        backup = backups[0]
        if str(backup.resolve()) not in output or "rm -f --" not in output:
            note(f"{case}: retained backup was not reported with an actionable cleanup command")
        if regular_metadata(backup) != expected_prior_metadata:
            note(f"{case}: cleanup failure backup lost regular-file metadata")

status, output, calls, fixture_root, _ = run_case(
    "candidate-cleanup-fails",
    mv_activation_results=(1,),
    rm_candidate_results=(1,),
)
dropin = fixture_root / "etc/ssh/sshd_config.d/00-panama.conf"
candidates = list(dropin.parent.glob(".00-panama.*.tmp"))
if status == 0:
    note("candidate-cleanup-fails: activation cleanup failure returned success")
if dropin.exists():
    note("candidate-cleanup-fails: failed activation changed the final drop-in")
if "install-handoff " in calls:
    note("candidate-cleanup-fails: failed activation reached install handoff")
if len(candidates) != 1:
    note("candidate-cleanup-fails: failed cleanup did not retain exactly one candidate")
else:
    candidate = candidates[0]
    if str(candidate.resolve()) not in output or "rm -f --" not in output:
        note("candidate-cleanup-fails: retained candidate lacked an actionable cleanup command")

signal_cases = {
    "signal-int-restores-prior": {
        "signal": signal.SIGINT,
        "status": 130,
        "prior_dropin": prior_dropin,
    },
    "signal-term-removes-new-dropin": {
        "signal": signal.SIGTERM,
        "status": 143,
        "prior_dropin": None,
    },
}

for case, expected in signal_cases.items():
    status, output, calls, fixture_root, boot_pid = run_case(
        case,
        signal_after_activation=expected["signal"],
        prior_traps=True,
        sshd_results=(0,),
        reload_results=(0,),
        prior_dropin=expected["prior_dropin"],
    )
    dropin = fixture_root / "etc/ssh/sshd_config.d/00-panama.conf"
    actual_dropin = dropin.read_bytes() if dropin.exists() else None
    if status != expected["status"]:
        note(f"{case}: signal returned status {status}, expected {expected['status']}")
    if actual_dropin != expected["prior_dropin"]:
        note(f"{case}: signal did not restore the pre-transaction SSH state")
    if list(dropin.parent.glob(".00-panama.*")):
        note(f"{case}: signal left transaction residue")
    if "install-handoff " in calls:
        note(f"{case}: signal reached install handoff")
    if f"prior-exit {boot_pid}\n" not in calls:
        note(f"{case}: signal suppressed the saved EXIT trap")
    if calls.count("sshd -t \n") != 1:
        note(f"{case}: signal did not validate restored configuration once")
    if calls.count("systemctl reload sshd.service \n") != 1:
        note(f"{case}: signal did not reload restored configuration once")
    call_lines = calls.splitlines()
    rollback_indices = [
        index
        for index, line in enumerate(call_lines)
        if (
            expected["prior_dropin"] is not None
            and re.fullmatch(
                rf"mv -f -- {re.escape(str(dropin.parent))}/\.00-panama\.[A-Za-z0-9]+\.restore "
                rf"{re.escape(str(dropin))} source-mode=640 ",
                line,
            )
        )
        or (expected["prior_dropin"] is None and line == f"rm -f -- {dropin} ")
    ]
    validation_indices = [
        index for index, line in enumerate(call_lines) if line == "sshd -t "
    ]
    reload_indices = [
        index
        for index, line in enumerate(call_lines)
        if line == "systemctl reload sshd.service "
    ]
    if not (
        len(rollback_indices) == len(validation_indices) == len(reload_indices) == 1
        and rollback_indices[0] < validation_indices[0] < reload_indices[0]
    ):
        note(f"{case}: signal did not restore, validate, then reload in order")
    if expected["prior_dropin"] is not None and regular_metadata(dropin) != expected_prior_metadata:
        note(f"{case}: signal rollback did not restore complete regular-file metadata")

pre_activation_signal_cases = {
    "signal-int-before-activation-prior": {
        "signal": signal.SIGINT,
        "status": 130,
        "prior_dropin": prior_dropin,
        "cleanup_fails": False,
    },
    "signal-term-before-activation-no-prior": {
        "signal": signal.SIGTERM,
        "status": 143,
        "prior_dropin": None,
        "cleanup_fails": False,
    },
    "signal-int-before-activation-cleanup-fails": {
        "signal": signal.SIGINT,
        "status": 130,
        "prior_dropin": None,
        "cleanup_fails": True,
    },
}

for case, expected in pre_activation_signal_cases.items():
    rm_candidate_results = (1,) if expected["cleanup_fails"] else ()
    status, output, calls, fixture_root, boot_pid = run_case(
        case,
        signal_before_activation=expected["signal"],
        prior_traps=True,
        prior_dropin=expected["prior_dropin"],
        rm_candidate_results=rm_candidate_results,
    )
    dropin = fixture_root / "etc/ssh/sshd_config.d/00-panama.conf"
    actual_dropin = dropin.read_bytes() if dropin.exists() else None
    candidates = list(dropin.parent.glob(".00-panama.*.tmp"))
    backups = list(dropin.parent.glob(".00-panama.*.backup"))
    if status != expected["status"]:
        note(f"{case}: signal returned status {status}, expected {expected['status']}")
    if actual_dropin != expected["prior_dropin"]:
        note(f"{case}: pre-activation signal changed the final drop-in state")
    if backups:
        note(f"{case}: pre-activation signal left backup residue")
    if expected["cleanup_fails"]:
        if len(candidates) != 1:
            note(f"{case}: injected cleanup failure did not retain exactly one candidate")
        else:
            candidate = candidates[0]
            expected_command = f"rm -f -- {candidate.resolve()}"
            if str(candidate.resolve()) not in output or expected_command not in output:
                note(f"{case}: retained candidate lacked its absolute cleanup command")
    elif candidates:
        note(f"{case}: pre-activation signal left candidate residue")
    if "install-handoff " in calls:
        note(f"{case}: pre-activation signal reached install handoff")
    if f"prior-exit {boot_pid}\n" not in calls:
        note(f"{case}: pre-activation signal suppressed the saved EXIT trap")
    if "sshd -t " in calls or "systemctl reload " in calls:
        note(f"{case}: pre-activation signal validated or reloaded unchanged SSH state")
    if ".restore " in calls or f"rm -f -- {dropin} " in calls:
        note(f"{case}: pre-activation signal rewrote the unchanged final drop-in")

preparation_signal_cases = {
    "signal-int-during-candidate-preparation": {
        "signal": signal.SIGINT,
        "status": 130,
        "prior_dropin": None,
        "phase": "candidate",
    },
    "signal-term-during-backup-preparation": {
        "signal": signal.SIGTERM,
        "status": 143,
        "prior_dropin": prior_dropin,
        "phase": "backup",
    },
}

for case, expected in preparation_signal_cases.items():
    signal_arguments = (
        {"signal_during_candidate_preparation": expected["signal"]}
        if expected["phase"] == "candidate"
        else {"signal_during_backup_preparation": expected["signal"]}
    )
    status, output, calls, fixture_root, boot_pid = run_case(
        case,
        prior_traps=True,
        prior_dropin=expected["prior_dropin"],
        **signal_arguments,
    )
    dropin = fixture_root / "etc/ssh/sshd_config.d/00-panama.conf"
    actual_dropin = dropin.read_bytes() if dropin.exists() else None
    if status != expected["status"]:
        note(f"{case}: signal returned status {status}, expected {expected['status']}")
    if actual_dropin != expected["prior_dropin"]:
        note(f"{case}: preparation signal changed the final drop-in")
    if list(dropin.parent.glob(".00-panama.*")):
        note(f"{case}: preparation signal left candidate or backup residue")
    if "install-handoff " in calls:
        note(f"{case}: preparation signal reached install handoff")
    if f"prior-exit {boot_pid}\n" not in calls:
        note(f"{case}: preparation signal suppressed the saved EXIT trap")
    if "sshd " in calls or "systemctl reload " in calls:
        note(f"{case}: preparation signal validated or reloaded unchanged SSH state")
    if expected["prior_dropin"] is not None and regular_metadata(dropin) != expected_prior_metadata:
        note(f"{case}: preparation signal changed prior regular-file metadata")

if findings:
    print(f"root server bootstrap: {len(findings)} finding(s)", file=sys.stderr)
    for finding in findings:
        print(f"  - {finding}", file=sys.stderr)
    raise SystemExit(1)

print("root server bootstrap: PASS")
PY