#!/usr/bin/env python3 import json import os import stat import subprocess import sys from pathlib import Path import pytest MODULE_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(MODULE_ROOT / "lib")) import seat_authority_installer as installer FILES = { "overdeck-seat-scope-entry": 0o755, "seat_scope_entry.py": 0o644, "seat_common.py": 0o644, "seat_implementer_identity.py": 0o644, "overdeck-seat-implementer-exec": 0o755, "seat_implementer_exec.py": 0o644, "overdeck-seat-tmux-mediator": 0o755, "seat_tmux_mediator.py": 0o644, "seat_execution_broker.py": 0o644, "seat_execution_client.py": 0o644, "overdeck-seat-execution-client": 0o755, "agent-session-reap-close.py": 0o755, "seat-launcher": 0o755, } def make_source(tmp_path: Path) -> Path: source = tmp_path / "source" source.mkdir() for name, mode in FILES.items(): path = source / name path.write_text(f"candidate:{name}\n") path.chmod(mode) return source def run_install(tmp_path: Path, source: Path, fault: str = "") -> installer.InstallResult: root = tmp_path / "root" return installer.install( source_dir=str(source), root=str(root), invoking_uid=os.getuid(), invoking_user="owner", fault=fault, ) def stable_paths(root: Path) -> list[Path]: return [ root / "usr/local/bin/overdeck-seat-scope-entry", root / "usr/local/bin/overdeck-seat-implementer-exec", root / "usr/local/bin/overdeck-seat-tmux-mediator", root / "usr/local/bin/overdeck-seat-execution-client", root / "etc/sudoers.d/overdeck-seat-scope-entry", ] def test_source_replacement_after_open_cannot_change_snapshot(tmp_path, monkeypatch): source = make_source(tmp_path) original = (source / "seat_scope_entry.py").read_bytes() def replace_after_open(name, fd): if name == "seat_scope_entry.py": replacement = source / "replacement" replacement.write_text("replacement\n") os.replace(replacement, source / name) monkeypatch.setattr(installer, "SOURCE_OPENED_HOOK", replace_after_open) with pytest.raises(installer.InstallError, match="source-changed"): run_install(tmp_path, source) assert not (tmp_path / "root/usr/local/lib/overdeck/seat-authority-active").exists() @pytest.mark.parametrize("kind", ["symlink", "directory", "fifo", "hardlink", "writable"]) def test_unsafe_source_type_or_metadata_fails_before_mutation(tmp_path, kind): source = make_source(tmp_path) path = source / "seat_scope_entry.py" path.unlink() if kind == "symlink": path.symlink_to(source / "seat_common.py") elif kind == "directory": path.mkdir() elif kind == "fifo": os.mkfifo(path) elif kind == "hardlink": os.link(source / "seat_common.py", path) else: path.write_text("unsafe\n") path.chmod(0o666) with pytest.raises(installer.InstallError): run_install(tmp_path, source) assert not (tmp_path / "root/usr/local/lib/overdeck/seat-authority-active").exists() @pytest.mark.parametrize("kind", ["symlink", "directory", "fifo", "wrong-owner"]) def test_unexpected_stable_destination_fails_closed(tmp_path, kind, monkeypatch): source = make_source(tmp_path) root = tmp_path / "root" path = stable_paths(root)[0] path.parent.mkdir(parents=True) if kind == "symlink": target = tmp_path / "target"; target.write_text("x"); path.symlink_to(target) elif kind == "directory": path.mkdir() elif kind == "fifo": os.mkfifo(path) else: path.write_text("x") monkeypatch.setattr(installer, "trusted_owner", lambda st: False) with pytest.raises(installer.InstallError): run_install(tmp_path, source) @pytest.mark.parametrize("kind", ["symlink", "writable", "wrong-type", "mismatch"]) def test_unsafe_existing_version_is_rejected(tmp_path, kind): source = make_source(tmp_path) root = tmp_path / "root" digest = installer.source_digest(str(source), os.getuid()) target = root / "usr/local/lib/overdeck/versions" / digest target.parent.mkdir(parents=True) if kind == "symlink": other = tmp_path / "other"; other.mkdir(); target.symlink_to(other) elif kind == "wrong-type": target.write_text("x") else: target.mkdir() if kind == "writable": target.chmod(0o777) else: (target / "seat_scope_entry.py").write_text("mismatch\n") with pytest.raises(installer.InstallError): run_install(tmp_path, source) @pytest.mark.parametrize("name", ["scope", "implementer", "mediator", "client", "sudoers", "activation"]) def test_every_publication_fault_restores_exact_state(tmp_path, name): source = make_source(tmp_path) root = tmp_path / "root" prior = {} for index, path in enumerate(stable_paths(root)): path.parent.mkdir(parents=True, exist_ok=True) path.write_text(f"prior:{index}\n") path.chmod(0o700 + index) prior[path] = (path.read_bytes(), stat.S_IMODE(path.stat().st_mode)) active = root / "usr/local/lib/overdeck/seat-authority-active" active.parent.mkdir(parents=True, exist_ok=True) active.write_text("prior\n"); active.chmod(0o600) with pytest.raises(installer.InstallError): run_install(tmp_path, source, f"after-{name}") assert active.read_text() == "prior\n" for path, expected in prior.items(): assert (path.read_bytes(), stat.S_IMODE(path.stat().st_mode)) == expected def test_interrupted_journal_recovers_before_new_install(tmp_path): source = make_source(tmp_path) with pytest.raises(installer.InstallInterrupted): run_install(tmp_path, source, "interrupt-after-client") root = tmp_path / "root" assert (root / "var/lib/overdeck/seat-install-transaction.json").exists() result = run_install(tmp_path, source) assert Path(result.version_root).is_dir() assert not (root / "var/lib/overdeck/seat-install-transaction.json").exists() def test_rollback_attempts_every_item_and_reports_incomplete(tmp_path, monkeypatch): source = make_source(tmp_path) real_restore = installer.restore_item calls = [] def failing_restore(item): calls.append(item["name"]) if item["name"] == "mediator": raise OSError("injected") return real_restore(item) monkeypatch.setattr(installer, "restore_item", failing_restore) with pytest.raises(installer.RollbackIncomplete): run_install(tmp_path, source, "after-sudoers") assert set(calls) >= {"scope", "implementer", "mediator", "client", "sudoers"} assert (tmp_path / "root/var/lib/overdeck/seat-install-transaction.json").exists() def test_success_leaves_one_activation_and_no_journal_or_temps(tmp_path): source = make_source(tmp_path) result = run_install(tmp_path, source) root = tmp_path / "root" assert (root / "usr/local/lib/overdeck/seat-authority-active").read_text().strip() == result.version assert not (root / "var/lib/overdeck/seat-install-transaction.json").exists() transaction_root = root / "var/lib/overdeck/seat-install-transactions" assert not transaction_root.exists() or not list(transaction_root.iterdir()) for path in stable_paths(root): assert path.is_file() and not path.is_symlink() def test_concurrent_install_lock_fails_closed(tmp_path): source = make_source(tmp_path) root = tmp_path / "root" lock = installer.acquire_lock(str(root)) try: with pytest.raises(installer.InstallError, match="install-busy"): run_install(tmp_path, source) finally: os.close(lock)