#!/usr/bin/env python3 import importlib.util import json import os import pathlib import signal import stat import subprocess import sys import tempfile import argparse import unittest from unittest import mock ROOT = pathlib.Path(__file__).resolve().parents[2] PROD = ROOT / "scripts/libexec/slack-webhook-recovery-dirfd.py" KDBX_BYTES = ( b"\x03\xd9\xa2\x9a\x67\xfb\x4b\xb5" b"\x01\x00\x04\x00" b"synthetic-keepass-payload-for-task-2\n" ) MAIN_RELATIVE = ("HyeonworksRecovery", "vault", "hyeonworks-recovery.kdbx") BACKUP_NAME = "hyeonworks-recovery.pre-slack-20260813T060000Z.kdbx" def load_helper(): spec = importlib.util.spec_from_file_location("swr_dirfd_helper", PROD) if spec is None or spec.loader is None: raise AssertionError("production dirfd helper is absent") module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module class HelperPresenceTest(unittest.TestCase): def test_production_helper_exists_and_exports_open_directory_chain(self): self.assertTrue(PROD.is_file(), "production dirfd helper is absent") helper = load_helper() self.assertTrue( hasattr(helper, "open_directory_chain"), "open_directory_chain is absent", ) class DirfdHelperTest(unittest.TestCase): @classmethod def setUpClass(cls): if not PROD.is_file(): raise unittest.SkipTest("production dirfd helper is absent") cls.helper = load_helper() if not hasattr(cls.helper, "open_directory_chain"): raise unittest.SkipTest("open_directory_chain is absent") def setUp(self): self.tempdir = tempfile.TemporaryDirectory() self.addCleanup(self.tempdir.cleanup) self.root = pathlib.Path(self.tempdir.name) self.mount_root = self.root / "mount-root" self.runtime = self.root / "runtime" self.destination = self.runtime / "baseline.kdbx" self.baseline_path = self.runtime / "baseline.json" self.candidate_path = self.runtime / "candidate.kdbx" self.snapshot_current_path = self.runtime / "current.kdbx" self.result_path = self.runtime / "result.json" self.runtime.mkdir(mode=0o700) self._build_mount_root(self.mount_root, KDBX_BYTES) def _build_mount_root(self, mount_root: pathlib.Path, main_bytes: bytes): backups = mount_root / "HyeonworksRecovery" / "vault" / "backups" backups.mkdir(parents=True, mode=0o700) for directory in ( mount_root, mount_root / "HyeonworksRecovery", mount_root / "HyeonworksRecovery" / "vault", backups, ): directory.chmod(0o700) main_path = mount_root.joinpath(*MAIN_RELATIVE) main_path.write_bytes(main_bytes) main_path.chmod(0o600) return main_path def _main_path(self, mount_root: pathlib.Path | None = None) -> pathlib.Path: base = self.mount_root if mount_root is None else mount_root return base.joinpath(*MAIN_RELATIVE) def _snapshot_state(self): state = self.helper.snapshot_database( os.fspath(self.mount_root), os.fspath(self.destination), ) self.helper.write_result(os.fspath(self.baseline_path), state) return state def _commit(self, backup_name: str = BACKUP_NAME): return self.helper.commit_database( os.fspath(self.mount_root), os.fspath(self.baseline_path), os.fspath(self.candidate_path), backup_name, ) def _supplementary_gid(self) -> int: gids = [gid for gid in os.getgroups() if gid != 1000] if not gids: self.skipTest("no alternate supplementary gid is available") return gids[0] def test_snapshot_copies_exact_database_and_private_state(self): state = self._snapshot_state() self.assertEqual(self.destination.read_bytes(), KDBX_BYTES) self.assertEqual(state["schema"], "platform-slack-keepass-fs-v1") self.assertEqual(state["operation"], "snapshot") self.assertEqual(state["source"]["relative_path"], "/".join(MAIN_RELATIVE)) self.assertEqual(state["source"]["size"], len(KDBX_BYTES)) self.assertEqual(state["source"]["mode"], "0600") self.assertEqual(state["mount"]["relative_vault"], "HyeonworksRecovery/vault") reloaded = json.loads(self.baseline_path.read_text(encoding="utf-8")) self.assertEqual(reloaded["schema"], state["schema"]) self.assertEqual(reloaded["source"]["sha256"], state["source"]["sha256"]) def test_commit_command_prearms_and_persists_rename_attempt_status(self): # Production break caught: the coordinator loses the helper after the # rename attempt and sees no durable state with which to prohibit retry. args = argparse.Namespace( mount_root=os.fspath(self.mount_root), baseline=os.fspath(self.baseline_path), candidate=os.fspath(self.candidate_path), backup_name=BACKUP_NAME, result=os.fspath(self.result_path), ) def lost_after_rename(*_args, status_hook=None): self.assertEqual( json.loads(self.result_path.read_text(encoding="utf-8"))["fs_commit"], "precommit-failure", ) self.assertIsNotNone(status_hook) status_hook("rename-attempted") raise RuntimeError("synthetic lost helper response") with mock.patch.object(self.helper, "commit_database", side_effect=lost_after_rename): with self.assertRaises(RuntimeError): self.helper._cmd_commit(args) persisted = json.loads(self.result_path.read_text(encoding="utf-8")) self.assertEqual( persisted, { "fs_commit": "rename-attempted", "operation": "commit-status", "schema": "platform-slack-keepass-fs-v1", }, ) def test_snapshot_rejects_symlinks_in_every_path_component(self): for index, part in enumerate(("HyeonworksRecovery", "vault", "hyeonworks-recovery.kdbx")): with self.subTest(component=part): mount_root = self.root / f"mount-symlink-{index}" mount_root.mkdir(mode=0o700) (mount_root / "real-target").mkdir(mode=0o700) (mount_root / "real-target" / "vault").mkdir(mode=0o700) (mount_root / "real-target" / "vault" / "backups").mkdir(mode=0o700) (mount_root / "real-target" / "vault" / "hyeonworks-recovery.kdbx").write_bytes(KDBX_BYTES) (mount_root / "real-target" / "vault" / "hyeonworks-recovery.kdbx").chmod(0o600) target = mount_root / "real-target" if part == "HyeonworksRecovery": (mount_root / "HyeonworksRecovery").symlink_to(target) elif part == "vault": (mount_root / "HyeonworksRecovery").mkdir(mode=0o700) (mount_root / "HyeonworksRecovery" / "vault").symlink_to(target / "vault") else: (mount_root / "HyeonworksRecovery").mkdir(mode=0o700) (mount_root / "HyeonworksRecovery" / "vault").mkdir(mode=0o700) (mount_root / "HyeonworksRecovery" / "vault" / "backups").mkdir(mode=0o700) (mount_root / "HyeonworksRecovery" / "vault" / "hyeonworks-recovery.kdbx").symlink_to( target / "vault" / "hyeonworks-recovery.kdbx" ) with self.assertRaises(Exception): self.helper.snapshot_database( os.fspath(mount_root), os.fspath(self.runtime / f"symlink-{index}.kdbx"), ) def test_snapshot_rejects_symlink_mount_root(self): real_mount = self.root / "real-mount" self._build_mount_root(real_mount, KDBX_BYTES) symlink_mount = self.root / "symlink-mount" symlink_mount.symlink_to(real_mount, target_is_directory=True) with self.assertRaises(Exception): self.helper.snapshot_database( os.fspath(symlink_mount), os.fspath(self.destination), ) def test_snapshot_rejects_nonprivate_mode_on_each_traversed_mount_directory(self): directories = ( self.mount_root, self.mount_root / "HyeonworksRecovery", self.mount_root / "HyeonworksRecovery" / "vault", ) for directory in directories: with self.subTest(directory=directory.name): self.setUp() directory = { "mount-root": self.mount_root, "HyeonworksRecovery": self.mount_root / "HyeonworksRecovery", "vault": self.mount_root / "HyeonworksRecovery" / "vault", }[directory.name] directory.chmod(0o777) with self.assertRaises(Exception): self.helper.snapshot_database( os.fspath(self.mount_root), os.fspath(self.destination), ) def test_snapshot_rejects_alternate_gid_on_each_traversed_mount_directory(self): alternate_gid = self._supplementary_gid() relative_directories = ((), ("HyeonworksRecovery",), ("HyeonworksRecovery", "vault")) for parts in relative_directories: with self.subTest(parts=parts): self.setUp() directory = self.mount_root.joinpath(*parts) os.chown(directory, -1, alternate_gid) with self.assertRaises(Exception): self.helper.snapshot_database( os.fspath(self.mount_root), os.fspath(self.destination), ) def test_commit_rejects_nonprivate_or_alternate_gid_backups_directory(self): alternate_gid = self._supplementary_gid() for case in ("mode", "gid"): with self.subTest(case=case): self.setUp() self._snapshot_state() self.candidate_path.write_bytes(KDBX_BYTES + b"candidate\n") self.candidate_path.chmod(0o600) backups = self.mount_root / "HyeonworksRecovery" / "vault" / "backups" if case == "mode": backups.chmod(0o777) else: os.chown(backups, -1, alternate_gid) with self.assertRaises(Exception): self._commit() def test_snapshot_rejects_main_kdbx_owned_by_supplementary_gid(self): os.chown(self._main_path(), -1, self._supplementary_gid()) with self.assertRaises(Exception): self.helper.snapshot_database( os.fspath(self.mount_root), os.fspath(self.destination), ) def test_snapshot_rejects_kdbx_symlink_and_hardlink(self): other = self.mount_root / "other.kdbx" other.write_bytes(KDBX_BYTES) other.chmod(0o600) main_path = self._main_path() main_path.unlink() main_path.symlink_to(other) with self.assertRaises(Exception): self.helper.snapshot_database(os.fspath(self.mount_root), os.fspath(self.destination)) self.setUp() main_path = self._main_path() hardlink = self.mount_root / "hardlink.kdbx" os.link(main_path, hardlink) with self.assertRaises(Exception): self.helper.snapshot_database(os.fspath(self.mount_root), os.fspath(self.destination)) def test_snapshot_rejects_wrong_type_and_mode(self): with self.subTest(case="wrong-type"): main_path = self._main_path() main_path.unlink() main_path.mkdir(mode=0o700) with self.assertRaises(Exception): self.helper.snapshot_database(os.fspath(self.mount_root), os.fspath(self.destination)) self.setUp() with self.subTest(case="wrong-mode"): self._main_path().chmod(0o644) with self.assertRaises(Exception): self.helper.snapshot_database(os.fspath(self.mount_root), os.fspath(self.destination)) def test_snapshot_staging_requires_exclusive_nofollow_destination(self): self.destination.write_bytes(b"occupied") self.destination.chmod(0o600) with self.assertRaises(Exception): self.helper.snapshot_database(os.fspath(self.mount_root), os.fspath(self.destination)) self.destination.unlink() target = self.runtime / "target.kdbx" target.write_bytes(b"target") target.chmod(0o600) self.destination.symlink_to(target) with self.assertRaises(Exception): self.helper.snapshot_database(os.fspath(self.mount_root), os.fspath(self.destination)) def test_snapshot_commands_reject_main_replacement_after_source_open(self): # Production break caught: a retained main FD is copied after its # canonical pathname has been replaced, publishing a stale snapshot. for operation in ("snapshot", "snapshot-current"): with self.subTest(operation=operation): self.setUp() retained_main = self.runtime / f"{operation}.retained-main.kdbx" drift_bytes = KDBX_BYTES + b"canonical-drift\n" original_open_regular_at = self.helper.open_regular_at replaced = False if operation == "snapshot-current": self.candidate_path.write_bytes(KDBX_BYTES) self.candidate_path.chmod(0o600) def replace_after_source_open(dir_fd, name, **kwargs): nonlocal replaced fd = original_open_regular_at(dir_fd, name, **kwargs) if not replaced and name == MAIN_RELATIVE[-1]: replaced = True self._main_path().rename(retained_main) self._main_path().write_bytes(drift_bytes) self._main_path().chmod(0o600) return fd with mock.patch.object( self.helper, "open_regular_at", side_effect=replace_after_source_open, ): with self.assertRaises(self.helper.FsContractError): if operation == "snapshot": self.helper.snapshot_database( os.fspath(self.mount_root), os.fspath(self.destination), ) else: self.helper.snapshot_committed_database( os.fspath(self.mount_root), os.fspath(self.candidate_path), os.fspath(self.snapshot_current_path), ) self.assertTrue(replaced) destination = ( self.destination if operation == "snapshot" else self.snapshot_current_path ) self.assertFalse(destination.exists()) self.assertEqual(self._main_path().read_bytes(), drift_bytes) def test_snapshot_commands_remove_destination_on_canonical_tree_rebind_after_copy(self): # Production break caught: copying from a retained old tree succeeds # even though the canonical mount tree is rebound before publication. for operation in ("snapshot", "snapshot-current"): with self.subTest(operation=operation): self.setUp() detached_recovery = self.mount_root / f"HyeonworksRecovery.{operation}.detached" drift_bytes = KDBX_BYTES + b"canonical-tree-drift\n" real_stable_copy = self.helper.stable_copy_fd rebound = False if operation == "snapshot-current": self.candidate_path.write_bytes(KDBX_BYTES) self.candidate_path.chmod(0o600) def rebind_after_copy(src_fd, dst_fd, **kwargs): nonlocal rebound copied = real_stable_copy(src_fd, dst_fd, **kwargs) if not rebound: rebound = True recovery = self.mount_root / "HyeonworksRecovery" recovery.rename(detached_recovery) self._build_mount_root(self.mount_root, drift_bytes) return copied with mock.patch.object( self.helper, "stable_copy_fd", side_effect=rebind_after_copy, ): with self.assertRaises(self.helper.FsContractError): if operation == "snapshot": self.helper.snapshot_database( os.fspath(self.mount_root), os.fspath(self.destination), ) else: self.helper.snapshot_committed_database( os.fspath(self.mount_root), os.fspath(self.candidate_path), os.fspath(self.snapshot_current_path), ) self.assertTrue(rebound) destination = ( self.destination if operation == "snapshot" else self.snapshot_current_path ) self.assertFalse(destination.exists()) self.assertEqual(self._main_path().read_bytes(), drift_bytes) def test_snapshot_commands_recheck_canonical_tree_immediately_before_publication(self): # Production break caught: the canonical tree is rebound while the # post-copy canonical traversal is already retained but before return. for operation in ("snapshot", "snapshot-current"): with self.subTest(operation=operation): self.setUp() detached_recovery = self.mount_root / f"HyeonworksRecovery.{operation}.late-detached" drift_bytes = KDBX_BYTES + b"late-canonical-tree-drift\n" original_open_mount_tree = self.helper._open_mount_tree open_count = 0 rebound = False if operation == "snapshot-current": self.candidate_path.write_bytes(KDBX_BYTES) self.candidate_path.chmod(0o600) def rebind_during_postcopy_reopen(stack, mount_root, *, include_backups): nonlocal open_count, rebound opened = original_open_mount_tree( stack, mount_root, include_backups=include_backups, ) open_count += 1 if open_count == 3 and not rebound: rebound = True recovery = self.mount_root / "HyeonworksRecovery" recovery.rename(detached_recovery) self._build_mount_root(self.mount_root, drift_bytes) return opened with mock.patch.object( self.helper, "_open_mount_tree", side_effect=rebind_during_postcopy_reopen, ): with self.assertRaises(self.helper.FsContractError): if operation == "snapshot": self.helper.snapshot_database( os.fspath(self.mount_root), os.fspath(self.destination), ) else: self.helper.snapshot_committed_database( os.fspath(self.mount_root), os.fspath(self.candidate_path), os.fspath(self.snapshot_current_path), ) self.assertTrue(rebound) destination = ( self.destination if operation == "snapshot" else self.snapshot_current_path ) self.assertFalse(destination.exists()) self.assertEqual(self._main_path().read_bytes(), drift_bytes) def test_snapshot_rejects_generated_destination_with_wrong_gid(self): alternate_gid = self._supplementary_gid() real_create = self.helper._create_private_file def create_with_wrong_snapshot_gid(dir_fd, name, **kwargs): fd = real_create(dir_fd, name, **kwargs) if name == self.destination.name: os.fchown(fd, -1, alternate_gid) return fd with mock.patch.object( self.helper, "_create_private_file", side_effect=create_with_wrong_snapshot_gid, ): with self.assertRaises(Exception): self.helper.snapshot_database( os.fspath(self.mount_root), os.fspath(self.destination), ) def test_write_result_rejects_generated_file_with_wrong_gid(self): alternate_gid = self._supplementary_gid() real_create = self.helper._create_private_file def create_with_wrong_result_gid(dir_fd, name, **kwargs): fd = real_create(dir_fd, name, **kwargs) if name.startswith(f".{self.result_path.name}.tmp."): os.fchown(fd, -1, alternate_gid) return fd with mock.patch.object( self.helper, "_create_private_file", side_effect=create_with_wrong_result_gid, ): with self.assertRaises(Exception): self.helper.write_result( os.fspath(self.result_path), {"schema": "synthetic"}, ) self.assertFalse(self.result_path.exists()) def test_commit_rejects_baseline_drift_and_backup_collision(self): self._snapshot_state() self.candidate_path.write_bytes(KDBX_BYTES + b"candidate\n") self.candidate_path.chmod(0o600) self._main_path().write_bytes(KDBX_BYTES + b"drift\n") self._main_path().chmod(0o600) with self.assertRaises(Exception): self._commit() self.setUp() self._snapshot_state() self.candidate_path.write_bytes(KDBX_BYTES + b"candidate\n") self.candidate_path.chmod(0o600) backup = self.mount_root / "HyeonworksRecovery" / "vault" / "backups" / BACKUP_NAME backup.write_bytes(b"collision") backup.chmod(0o600) with self.assertRaises(Exception): self._commit() def test_commit_rejects_candidate_owned_by_supplementary_gid(self): self._snapshot_state() self.candidate_path.write_bytes(KDBX_BYTES + b"candidate\n") self.candidate_path.chmod(0o600) os.chown(self.candidate_path, -1, self._supplementary_gid()) with self.assertRaises(Exception): self._commit() def test_commit_rejects_baseline_owned_by_supplementary_gid(self): self._snapshot_state() self.candidate_path.write_bytes(KDBX_BYTES + b"candidate\n") self.candidate_path.chmod(0o600) os.chown(self.baseline_path, -1, self._supplementary_gid()) with self.assertRaises(Exception): self._commit() def test_status_rejects_result_owned_by_supplementary_gid(self): self.helper.write_result( os.fspath(self.result_path), { "schema": "platform-slack-keepass-fs-v1", "operation": "commit-status", "fs_commit": "precommit-failure", }, ) os.chown(self.result_path, -1, self._supplementary_gid()) with self.assertRaises(Exception): self.helper.read_commit_status(os.fspath(self.result_path)) def test_commit_rejects_generated_backup_with_wrong_gid_before_replace(self): self._snapshot_state() original_bytes = self._main_path().read_bytes() self.candidate_path.write_bytes(KDBX_BYTES + b"candidate\n") self.candidate_path.chmod(0o600) alternate_gid = self._supplementary_gid() real_create = self.helper._create_private_file def create_with_wrong_backup_gid(dir_fd, name, **kwargs): fd = real_create(dir_fd, name, **kwargs) if name == BACKUP_NAME: os.fchown(fd, -1, alternate_gid) return fd with mock.patch.object( self.helper, "_create_private_file", side_effect=create_with_wrong_backup_gid, ): with self.assertRaises(Exception): self._commit() self.assertEqual(self._main_path().read_bytes(), original_bytes) def test_regular_file_uid_and_gid_attestation_targets_each_exact_role(self): # Production break caught: a role-specific ownership check is absent, # while unrelated runtime-parent metadata can remain fully valid. roles = ( "main", "baseline", "candidate", "generated-snapshot", "generated-result", "generated-backup", "reopened-backup", "stage", "installed-main", ) for identity_field in ("uid", "gid"): for role in roles: with self.subTest(identity_field=identity_field, role=role): self.setUp() if role not in {"main", "generated-snapshot", "generated-result"}: self._snapshot_state() self.candidate_path.write_bytes(KDBX_BYTES + b"candidate-role-attestation\n") self.candidate_path.chmod(0o600) real_fstat = self.helper.os.fstat real_replace = self.helper.os.replace path_counts = {} published = False targeted = False def recording_replace(src, dst, *, src_dir_fd=None, dst_dir_fd=None): nonlocal published result = real_replace( src, dst, src_dir_fd=src_dir_fd, dst_dir_fd=dst_dir_fd, ) if dst == MAIN_RELATIVE[-1]: published = True return result def fstat_with_targeted_identity(fd): nonlocal targeted observed = real_fstat(fd) try: fd_path = os.readlink(f"/proc/self/fd/{fd}") except OSError: return observed path_counts[fd_path] = path_counts.get(fd_path, 0) + 1 count = path_counts[fd_path] basename = pathlib.Path(fd_path).name is_target = False if role == "main": is_target = fd_path == os.fspath(self._main_path()) elif role == "baseline": is_target = fd_path == os.fspath(self.baseline_path) elif role == "candidate": is_target = fd_path == os.fspath(self.candidate_path) elif role == "generated-snapshot": is_target = fd_path == os.fspath(self.destination) and count == 3 elif role == "generated-result": is_target = basename.startswith(f".{self.result_path.name}.tmp.") and count == 3 elif role == "generated-backup": is_target = basename == BACKUP_NAME and count == 2 elif role == "reopened-backup": is_target = basename == BACKUP_NAME and count == 4 elif role == "stage": is_target = basename.startswith(f".{MAIN_RELATIVE[-1]}.stage.") and count == 3 elif role == "installed-main": is_target = published and fd_path.endswith(f"/vault/{MAIN_RELATIVE[-1]}") if not is_target or targeted: return observed targeted = True values = list(observed) index = 4 if identity_field == "uid" else 5 values[index] = values[index] + 1 return os.stat_result(values) with mock.patch.object( self.helper.os, "fstat", side_effect=fstat_with_targeted_identity, ), mock.patch.object( self.helper.os, "replace", side_effect=recording_replace, ): if role == "main": with self.assertRaises(self.helper.FsContractError): self.helper.snapshot_database( os.fspath(self.mount_root), os.fspath(self.destination), ) elif role == "generated-snapshot": with self.assertRaises(self.helper.FsContractError): self.helper.snapshot_database( os.fspath(self.mount_root), os.fspath(self.destination), ) self.assertFalse(self.destination.exists()) elif role == "generated-result": with self.assertRaises(self.helper.FsContractError): self.helper.write_result( os.fspath(self.result_path), {"schema": "synthetic"}, ) self.assertFalse(self.result_path.exists()) elif role == "installed-main": result = self._commit() self.assertEqual(result["fs_commit"], "committed-but-uncertain") else: with self.assertRaises(self.helper.FsContractError): self._commit() self.assertTrue(targeted) def test_commit_rejects_backup_path_replacement_with_identical_bytes(self): self._snapshot_state() original_bytes = self._main_path().read_bytes() self.candidate_path.write_bytes(KDBX_BYTES + b"candidate\n") self.candidate_path.chmod(0o600) backups = self.mount_root / "HyeonworksRecovery" / "vault" / "backups" backup_path = backups / BACKUP_NAME retained_name = backups / f"{BACKUP_NAME}.retained" real_fsync = self.helper.os.fsync replaced = False def replace_after_backup_copy(fd): nonlocal replaced fd_path = os.readlink(f"/proc/self/fd/{fd}") result = real_fsync(fd) if not replaced and fd_path.endswith("/backups"): replaced = True backup_path.rename(retained_name) backup_path.write_bytes(original_bytes) backup_path.chmod(0o600) return result with mock.patch.object(self.helper.os, "fsync", side_effect=replace_after_backup_copy): with self.assertRaises(Exception): self._commit() self.assertTrue(replaced) self.assertEqual(self._main_path().read_bytes(), original_bytes) self.assertEqual(backup_path.read_bytes(), original_bytes) self.assertEqual(retained_name.read_bytes(), original_bytes) def test_commit_rejects_detached_vault_tree_before_any_write(self): self._snapshot_state() original_bytes = self._main_path().read_bytes() candidate_bytes = KDBX_BYTES + b"candidate\n" self.candidate_path.write_bytes(candidate_bytes) self.candidate_path.chmod(0o600) original_open_chain = self.helper.open_directory_chain detached_recovery = self.mount_root / "HyeonworksRecovery.detached" renamed = False def rename_before_backups_open(root_fd, parts): nonlocal renamed root_path = pathlib.Path(os.readlink(f"/proc/self/fd/{root_fd}")) if not renamed and tuple(parts) == ("backups",) and root_path.name == "vault": renamed = True recovery = self.mount_root / "HyeonworksRecovery" recovery.rename(detached_recovery) self._build_mount_root(self.mount_root, original_bytes) return original_open_chain(root_fd, parts) with mock.patch.object( self.helper, "open_directory_chain", side_effect=rename_before_backups_open, ): with self.assertRaises(Exception): self._commit() self.assertTrue(renamed) canonical_main = self._main_path() detached_vault = detached_recovery / "vault" detached_main = detached_vault / MAIN_RELATIVE[-1] self.assertEqual(canonical_main.read_bytes(), original_bytes) self.assertEqual(detached_main.read_bytes(), original_bytes) self.assertEqual(list((self.mount_root / "HyeonworksRecovery" / "vault" / "backups").iterdir()), []) self.assertEqual(list((detached_vault / "backups").iterdir()), []) self.assertEqual( [path for path in detached_vault.iterdir() if path.name.startswith(f".{MAIN_RELATIVE[-1]}.stage.")], [], ) def test_commit_backup_proof_failure_stops_before_replace(self): self._snapshot_state() self.candidate_path.write_bytes(KDBX_BYTES + b"candidate\n") self.candidate_path.chmod(0o600) replace_calls = [] with mock.patch.object(self.helper, "files_equal_fd", return_value=False), mock.patch.object( self.helper.os, "replace", side_effect=lambda *args, **kwargs: replace_calls.append((args, kwargs)), ): with self.assertRaises(Exception): self._commit() self.assertEqual(replace_calls, []) def test_commit_never_verifies_when_recovery_tree_is_rebound_inside_replace(self): # Production break caught: stage-to-main replace runs on a newly # detached retained vault and is falsely reported as verified. self._snapshot_state() original_bytes = self._main_path().read_bytes() candidate_bytes = KDBX_BYTES + b"candidate-inside-replace\n" self.candidate_path.write_bytes(candidate_bytes) self.candidate_path.chmod(0o600) detached_recovery = self.mount_root / "HyeonworksRecovery.detached-during-replace" real_replace = self.helper.os.replace replace_calls = [] rebound = False def rebind_inside_replace(src, dst, *, src_dir_fd=None, dst_dir_fd=None): nonlocal rebound replace_calls.append((src, dst, src_dir_fd, dst_dir_fd)) if not rebound and dst == MAIN_RELATIVE[-1] and src.startswith(f".{MAIN_RELATIVE[-1]}.stage."): rebound = True recovery = self.mount_root / "HyeonworksRecovery" recovery.rename(detached_recovery) self._build_mount_root(self.mount_root, original_bytes) return real_replace(src, dst, src_dir_fd=src_dir_fd, dst_dir_fd=dst_dir_fd) with mock.patch.object( self.helper.os, "replace", side_effect=rebind_inside_replace, ): result = self._commit() self.assertTrue(rebound) self.assertEqual(result["fs_commit"], "committed-but-uncertain") self.assertEqual(len(replace_calls), 1) self.assertEqual(self._main_path().read_bytes(), original_bytes) detached_vault = detached_recovery / "vault" self.assertEqual((detached_vault / MAIN_RELATIVE[-1]).read_bytes(), candidate_bytes) self.assertEqual((detached_vault / "backups" / BACKUP_NAME).read_bytes(), original_bytes) self.assertEqual( [path for path in detached_vault.iterdir() if path.name.startswith(f".{MAIN_RELATIVE[-1]}.stage.")], [], ) def test_commit_rejects_canonical_main_replacement_after_final_pre_replace_open(self): # Production break caught: the final baseline check uses a retained FD # but never proves MAIN_NAME still names it immediately before replace. self._snapshot_state() original_bytes = self._main_path().read_bytes() drift_bytes = KDBX_BYTES + b"pre-replace-drift\n" self.candidate_path.write_bytes(KDBX_BYTES + b"candidate\n") self.candidate_path.chmod(0o600) retained_main = self.mount_root / "HyeonworksRecovery" / "vault" / "pre-replace-retained.kdbx" original_open_regular_at = self.helper.open_regular_at real_replace = self.helper.os.replace main_open_count = 0 replace_calls = [] def replace_after_final_main_open(dir_fd, name, **kwargs): nonlocal main_open_count fd = original_open_regular_at(dir_fd, name, **kwargs) if name == MAIN_RELATIVE[-1]: main_open_count += 1 if main_open_count == 2: self._main_path().rename(retained_main) self._main_path().write_bytes(drift_bytes) self._main_path().chmod(0o600) return fd def recording_replace(src, dst, *, src_dir_fd=None, dst_dir_fd=None): replace_calls.append((src, dst, src_dir_fd, dst_dir_fd)) return real_replace(src, dst, src_dir_fd=src_dir_fd, dst_dir_fd=dst_dir_fd) with mock.patch.object( self.helper, "open_regular_at", side_effect=replace_after_final_main_open, ), mock.patch.object(self.helper.os, "replace", side_effect=recording_replace): with self.assertRaises(self.helper.FsContractError): self._commit() self.assertEqual(main_open_count, 2) self.assertEqual(replace_calls, []) self.assertEqual(self._main_path().read_bytes(), drift_bytes) self.assertEqual(retained_main.read_bytes(), original_bytes) backup_path = self.mount_root / "HyeonworksRecovery" / "vault" / "backups" / BACKUP_NAME self.assertEqual(backup_path.read_bytes(), original_bytes) vault = self.mount_root / "HyeonworksRecovery" / "vault" self.assertEqual( [path for path in vault.iterdir() if path.name.startswith(f".{MAIN_RELATIVE[-1]}.stage.")], [], ) def test_commit_never_verifies_canonical_rebind_after_postpublication_sync(self): # Production break caught: canonical ancestry changes after the stage # has been published and synced but before verified status is returned. self._snapshot_state() original_bytes = self._main_path().read_bytes() candidate_bytes = KDBX_BYTES + b"candidate-postpublication\n" self.candidate_path.write_bytes(candidate_bytes) self.candidate_path.chmod(0o600) detached_recovery = self.mount_root / "HyeonworksRecovery.detached-postpublication" real_replace = self.helper.os.replace real_syncfs = self.helper.sync_filesystem_fd replace_calls = [] published = False rebound = False def recording_replace(src, dst, *, src_dir_fd=None, dst_dir_fd=None): nonlocal published replace_calls.append((src, dst, src_dir_fd, dst_dir_fd)) result = real_replace(src, dst, src_dir_fd=src_dir_fd, dst_dir_fd=dst_dir_fd) if dst == MAIN_RELATIVE[-1]: published = True return result def rebind_after_postpublication_sync(fd): nonlocal rebound result = real_syncfs(fd) if published and not rebound: rebound = True recovery = self.mount_root / "HyeonworksRecovery" recovery.rename(detached_recovery) self._build_mount_root(self.mount_root, original_bytes) return result with mock.patch.object( self.helper.os, "replace", side_effect=recording_replace, ), mock.patch.object( self.helper, "sync_filesystem_fd", side_effect=rebind_after_postpublication_sync, ): result = self._commit() self.assertTrue(rebound) self.assertEqual(result["fs_commit"], "committed-but-uncertain") self.assertEqual(len(replace_calls), 1) self.assertEqual(self._main_path().read_bytes(), original_bytes) detached_vault = detached_recovery / "vault" self.assertEqual((detached_vault / MAIN_RELATIVE[-1]).read_bytes(), candidate_bytes) self.assertEqual((detached_vault / "backups" / BACKUP_NAME).read_bytes(), original_bytes) def test_commit_rechecks_canonical_tree_immediately_before_verified_status(self): # Production break caught: a canonical traversal retained after sync # is detached before verified status is selected. self._snapshot_state() original_bytes = self._main_path().read_bytes() candidate_bytes = KDBX_BYTES + b"candidate-late-status\n" self.candidate_path.write_bytes(candidate_bytes) self.candidate_path.chmod(0o600) detached_recovery = self.mount_root / "HyeonworksRecovery.detached-late-status" original_open_mount_tree = self.helper._open_mount_tree real_replace = self.helper.os.replace published = False rebound = False replace_calls = [] def recording_replace(src, dst, *, src_dir_fd=None, dst_dir_fd=None): nonlocal published replace_calls.append((src, dst, src_dir_fd, dst_dir_fd)) result = real_replace(src, dst, src_dir_fd=src_dir_fd, dst_dir_fd=dst_dir_fd) if dst == MAIN_RELATIVE[-1]: published = True return result def rebind_during_first_postpublication_reopen(stack, mount_root, *, include_backups): nonlocal rebound opened = original_open_mount_tree( stack, mount_root, include_backups=include_backups, ) if published and not rebound: rebound = True recovery = self.mount_root / "HyeonworksRecovery" recovery.rename(detached_recovery) self._build_mount_root(self.mount_root, original_bytes) return opened with mock.patch.object( self.helper.os, "replace", side_effect=recording_replace, ), mock.patch.object( self.helper, "_open_mount_tree", side_effect=rebind_during_first_postpublication_reopen, ): result = self._commit() self.assertTrue(rebound) self.assertEqual(result["fs_commit"], "committed-but-uncertain") self.assertEqual(len(replace_calls), 1) self.assertEqual(self._main_path().read_bytes(), original_bytes) detached_vault = detached_recovery / "vault" self.assertEqual((detached_vault / MAIN_RELATIVE[-1]).read_bytes(), candidate_bytes) self.assertEqual((detached_vault / "backups" / BACKUP_NAME).read_bytes(), original_bytes) def test_commit_uses_dirfd_replace_and_defers_signals(self): self._snapshot_state() candidate_bytes = KDBX_BYTES + b"candidate\n" self.candidate_path.write_bytes(candidate_bytes) self.candidate_path.chmod(0o600) replace_calls = [] mask_calls = [] real_replace = self.helper.os.replace real_pthread_sigmask = getattr(signal, "pthread_sigmask") def recording_replace(src, dst, *, src_dir_fd=None, dst_dir_fd=None): replace_calls.append((src, dst, src_dir_fd, dst_dir_fd)) return real_replace(src, dst, src_dir_fd=src_dir_fd, dst_dir_fd=dst_dir_fd) def recording_sigmask(how, mask): mask_calls.append((how, tuple(sorted(mask)))) if len(mask_calls) == 1: return set() return set(mask) with mock.patch.object(self.helper.os, "replace", side_effect=recording_replace), mock.patch.object( self.helper.signal, "pthread_sigmask", side_effect=recording_sigmask, ): result = self._commit() self.assertEqual(result["fs_commit"], "verified-commit") self.assertEqual(self._main_path().read_bytes(), candidate_bytes) self.assertEqual(len(replace_calls), 1) src, dst, src_dir_fd, dst_dir_fd = replace_calls[0] self.assertEqual(dst, MAIN_RELATIVE[-1]) self.assertNotEqual(src, os.fspath(self.candidate_path)) self.assertIsInstance(src_dir_fd, int) self.assertIsInstance(dst_dir_fd, int) self.assertEqual(len(mask_calls), 2) self.assertEqual(mask_calls[0][0], signal.SIG_BLOCK) self.assertEqual(set(mask_calls[0][1]), {signal.SIGHUP, signal.SIGINT, signal.SIGTERM}) self.assertEqual(mask_calls[1][0], signal.SIG_SETMASK) self.assertEqual( self.helper.snapshot_committed_database( os.fspath(self.mount_root), os.fspath(self.candidate_path), os.fspath(self.snapshot_current_path), )["operation"], "snapshot-current", ) self.assertEqual(self.snapshot_current_path.read_bytes(), candidate_bytes) def test_commit_classifies_pre_replace_sync_failures(self): cases = [] def backup_file_failure(fd): return "backups/" in os.readlink(f"/proc/self/fd/{fd}") and os.path.basename( os.readlink(f"/proc/self/fd/{fd}") ) == BACKUP_NAME def backup_dir_failure(fd): return os.readlink(f"/proc/self/fd/{fd}").endswith("/backups") cases.append(("backup-file-fsync", backup_file_failure, None)) cases.append(("backup-dir-fsync", backup_dir_failure, None)) cases.append(("filesystem-sync", None, "syncfs")) for name, fsync_matcher, helper_failure in cases: with self.subTest(case=name): self.setUp() self._snapshot_state() original_bytes = self._main_path().read_bytes() candidate_bytes = KDBX_BYTES + name.encode("utf-8") self.candidate_path.write_bytes(candidate_bytes) self.candidate_path.chmod(0o600) replace_calls = [] real_fsync = self.helper.os.fsync def failing_fsync(fd): if fsync_matcher is not None and fsync_matcher(fd): raise OSError("synthetic fsync failure") return real_fsync(fd) patches = [ mock.patch.object( self.helper.os, "replace", side_effect=lambda *args, **kwargs: replace_calls.append((args, kwargs)), ) ] if fsync_matcher is not None: patches.append(mock.patch.object(self.helper.os, "fsync", side_effect=failing_fsync)) if helper_failure == "syncfs": patches.append( mock.patch.object( self.helper, "sync_filesystem_fd", side_effect=OSError("synthetic syncfs failure"), ) ) with patches[0]: with patches[1] if len(patches) > 1 else mock.patch.object(self.helper, "write_result", self.helper.write_result): with patches[2] if len(patches) > 2 else mock.patch.object(self.helper, "write_result", self.helper.write_result): with self.assertRaises(Exception): self._commit() self.assertEqual(replace_calls, []) self.assertEqual(self._main_path().read_bytes(), original_bytes) def test_commit_cleans_only_owned_stage_after_pre_replace_vault_sync_failure(self): self._snapshot_state() original_bytes = self._main_path().read_bytes() candidate_bytes = KDBX_BYTES + b"candidate\n" self.candidate_path.write_bytes(candidate_bytes) self.candidate_path.chmod(0o600) vault_dir = self.mount_root / "HyeonworksRecovery" / "vault" backup_path = vault_dir / "backups" / BACKUP_NAME foreign_stage = vault_dir / f".{MAIN_RELATIVE[-1]}.stage.foreign" foreign_stage.write_bytes(b"foreign-stage") foreign_stage.chmod(0o600) real_syncfs = self.helper.sync_filesystem_fd def fail_vault_syncfs(fd): fd_path = os.readlink(f"/proc/self/fd/{fd}") if fd_path.endswith("/vault"): raise OSError("synthetic vault syncfs failure") return real_syncfs(fd) with mock.patch.object(self.helper, "sync_filesystem_fd", side_effect=fail_vault_syncfs): with self.assertRaises(Exception): self._commit() owned_stage_prefix = f".{MAIN_RELATIVE[-1]}.stage.{os.getpid()}." self.assertEqual(self._main_path().read_bytes(), original_bytes) self.assertTrue(backup_path.is_file()) self.assertEqual(backup_path.read_bytes(), original_bytes) self.assertTrue(foreign_stage.is_file()) self.assertEqual( sorted(path.name for path in vault_dir.iterdir() if path.name.startswith(owned_stage_prefix)), [], ) def test_commit_classifies_replace_and_post_replace_failures_without_second_replace(self): scenarios = ("replace-failure", "post-file-fsync", "post-dir-fsync", "post-syncfs") for scenario in scenarios: with self.subTest(case=scenario): self.setUp() self._snapshot_state() candidate_bytes = KDBX_BYTES + scenario.encode("utf-8") self.candidate_path.write_bytes(candidate_bytes) self.candidate_path.chmod(0o600) replace_calls = [] real_replace = self.helper.os.replace real_fsync = self.helper.os.fsync real_syncfs = self.helper.sync_filesystem_fd vault_fsync_calls = 0 syncfs_calls = 0 def maybe_fail_replace(src, dst, *, src_dir_fd=None, dst_dir_fd=None): replace_calls.append((src, dst, src_dir_fd, dst_dir_fd)) if scenario == "replace-failure": raise OSError("synthetic replace failure") return real_replace(src, dst, src_dir_fd=src_dir_fd, dst_dir_fd=dst_dir_fd) def maybe_fail_fsync(fd): nonlocal vault_fsync_calls fd_path = os.readlink(f"/proc/self/fd/{fd}") if scenario == "post-file-fsync" and fd_path.endswith(MAIN_RELATIVE[-1]): raise OSError("synthetic installed-main fsync failure") if fd_path.endswith("/vault"): vault_fsync_calls += 1 if scenario == "post-dir-fsync" and fd_path.endswith("/vault") and vault_fsync_calls == 2: raise OSError("synthetic vault-dir fsync failure") return real_fsync(fd) def maybe_fail_syncfs(fd): nonlocal syncfs_calls syncfs_calls += 1 if scenario == "post-syncfs" and syncfs_calls == 3: raise OSError("synthetic post-rename syncfs failure") return real_syncfs(fd) with mock.patch.object(self.helper.os, "replace", side_effect=maybe_fail_replace), mock.patch.object( self.helper.os, "fsync", side_effect=maybe_fail_fsync ), mock.patch.object(self.helper, "sync_filesystem_fd", side_effect=maybe_fail_syncfs): result = self._commit() self.assertEqual(result["fs_commit"], "committed-but-uncertain") self.assertEqual(len(replace_calls), 1) def test_commit_blocks_signals_through_post_replace_classification(self): self._snapshot_state() candidate_bytes = KDBX_BYTES + b"candidate\n" self.candidate_path.write_bytes(candidate_bytes) self.candidate_path.chmod(0o600) real_replace = self.helper.os.replace real_fsync = self.helper.os.fsync real_syncfs = self.helper.sync_filesystem_fd real_files_equal = self.helper.files_equal_fd current_mask = set() replaced = False checkpoints = [] def recording_sigmask(how, mask): nonlocal current_mask mask_set = set(mask) if how == signal.SIG_BLOCK: previous = set(current_mask) current_mask.update(mask_set) return previous if how == signal.SIG_SETMASK: current_mask = set(mask_set) return set() raise AssertionError(f"unexpected sigmask mode: {how}") def recording_replace(src, dst, *, src_dir_fd=None, dst_dir_fd=None): nonlocal replaced replaced = True return real_replace(src, dst, src_dir_fd=src_dir_fd, dst_dir_fd=dst_dir_fd) def recording_files_equal(left_fd, right_fd): if replaced: checkpoints.append(("post-compare", set(current_mask))) return real_files_equal(left_fd, right_fd) def recording_fsync(fd): if replaced: checkpoints.append((os.readlink(f"/proc/self/fd/{fd}"), set(current_mask))) return real_fsync(fd) def recording_syncfs(fd): if replaced: checkpoints.append((f"syncfs:{os.readlink(f'/proc/self/fd/{fd}')}", set(current_mask))) return real_syncfs(fd) with mock.patch.object(self.helper.signal, "pthread_sigmask", side_effect=recording_sigmask), mock.patch.object( self.helper.os, "replace", side_effect=recording_replace ), mock.patch.object(self.helper, "files_equal_fd", side_effect=recording_files_equal), mock.patch.object( self.helper.os, "fsync", side_effect=recording_fsync ), mock.patch.object(self.helper, "sync_filesystem_fd", side_effect=recording_syncfs): result = self._commit() self.assertEqual(result["fs_commit"], "verified-commit") self.assertGreaterEqual(len(checkpoints), 3) for _, observed_mask in checkpoints: self.assertTrue( {signal.SIGHUP, signal.SIGINT, signal.SIGTERM}.issubset(observed_mask), f"post-replace checkpoint observed unblocked mask: {observed_mask!r}", ) self.assertEqual(current_mask, set()) def test_commit_rejects_malformed_extra_and_foreign_mount_state(self): state = self._snapshot_state() self.candidate_path.write_bytes(KDBX_BYTES + b"candidate\n") self.candidate_path.chmod(0o600) malformed = dict(state) malformed.pop("source") self.baseline_path.write_text(json.dumps(malformed), encoding="utf-8") with self.assertRaises(Exception): self._commit() extra = dict(state) extra["unexpected"] = "value" self.helper.write_result(os.fspath(self.baseline_path), extra) with self.assertRaises(Exception): self._commit() foreign_mount = dict(state) foreign_mount["mount"] = dict(foreign_mount["mount"]) foreign_mount["mount"]["realpath"] = os.fspath(self.root / "foreign-mount") self.helper.write_result(os.fspath(self.baseline_path), foreign_mount) with self.assertRaises(Exception): self._commit() def test_commit_rejects_unsafe_baseline_path_metadata(self): with self.subTest(case="symlink-replacement"): self.setUp() self._snapshot_state() self.candidate_path.write_bytes(KDBX_BYTES + b"candidate\n") self.candidate_path.chmod(0o600) baseline_bytes = self.baseline_path.read_bytes() target = self.runtime / "baseline-target.json" target.write_bytes(baseline_bytes) target.chmod(0o600) self.baseline_path.unlink() self.baseline_path.symlink_to(target) with self.assertRaises(Exception): self._commit() with self.subTest(case="hardlink-replacement"): self.setUp() self._snapshot_state() self.candidate_path.write_bytes(KDBX_BYTES + b"candidate\n") self.candidate_path.chmod(0o600) baseline_bytes = self.baseline_path.read_bytes() target = self.runtime / "baseline-target.json" target.write_bytes(baseline_bytes) target.chmod(0o600) self.baseline_path.unlink() os.link(target, self.baseline_path) with self.assertRaises(Exception): self._commit() with self.subTest(case="wrong-mode"): self.setUp() self._snapshot_state() self.candidate_path.write_bytes(KDBX_BYTES + b"candidate\n") self.candidate_path.chmod(0o600) self.baseline_path.chmod(0o644) with self.assertRaises(Exception): self._commit() with self.subTest(case="wrong-type"): self.setUp() self._snapshot_state() self.candidate_path.write_bytes(KDBX_BYTES + b"candidate\n") self.candidate_path.chmod(0o600) self.baseline_path.unlink() self.baseline_path.mkdir(mode=0o700) with self.assertRaises(Exception): self._commit() def test_commit_uses_retained_baseline_fd_if_path_is_replaced_after_open(self): self._snapshot_state() self.candidate_path.write_bytes(KDBX_BYTES + b"candidate\n") self.candidate_path.chmod(0o600) original_open_regular_at = self.helper.open_regular_at baseline_parent = self.baseline_path.parent.resolve() baseline_name = self.baseline_path.name replacement_done = False def replacing_open_regular_at(dir_fd, name, **kwargs): nonlocal replacement_done fd = original_open_regular_at(dir_fd, name, **kwargs) opened_path = pathlib.Path(os.readlink(f"/proc/self/fd/{fd}")).resolve() parent_path = pathlib.Path(os.readlink(f"/proc/self/fd/{dir_fd}")).resolve() if ( not replacement_done and name == baseline_name and parent_path == baseline_parent and opened_path == self.baseline_path.resolve() ): replacement_done = True replacement = self.runtime / "baseline-replacement.json" replacement.write_text('{"schema":"tampered"}', encoding="utf-8") replacement.chmod(0o600) self.baseline_path.unlink() replacement.rename(self.baseline_path) return fd with mock.patch.object(self.helper, "open_regular_at", side_effect=replacing_open_regular_at): result = self._commit() self.assertTrue(replacement_done) self.assertEqual(result["fs_commit"], "verified-commit") self.assertEqual(self._main_path().read_bytes(), self.candidate_path.read_bytes()) self.assertEqual(self.baseline_path.read_text(encoding="utf-8"), '{"schema":"tampered"}') with self.assertRaises(Exception): self._commit() def test_cli_expected_failures_are_payload_free_for_all_mutating_commands(self): marker = f"synthetic-unique-path-{os.getpid()}" missing = os.fspath(self.root / marker) commands = ( ["snapshot", "--mount-root", missing, "--destination", missing, "--result", missing], ["commit", "--mount-root", missing, "--baseline", missing, "--candidate", missing, "--backup-name", BACKUP_NAME, "--result", missing], ["snapshot-current", "--mount-root", missing, "--candidate", missing, "--destination", missing, "--result", missing], ) for arguments in commands: with self.subTest(command=arguments[0]): completed = subprocess.run( [sys.executable, os.fspath(PROD), *arguments], stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=5, check=False, ) self.assertEqual(completed.returncode, 1) self.assertEqual(completed.stdout, b"") self.assertEqual(completed.stderr, b"dirfd helper failed\n") self.assertNotIn(marker.encode(), completed.stderr) self.assertNotIn(b"Traceback", completed.stderr) def test_process_audit_detects_retained_proc_fd_runtime_reference(self): runtime_fd = os.open(self.runtime, os.O_RDONLY | os.O_DIRECTORY) process = subprocess.Popen( [sys.executable, "-c", "import time; time.sleep(30)", "keepassxc-cli", f"/proc/{os.getpid()}/fd/{runtime_fd}"], pass_fds=(runtime_fd,), stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) try: with self.assertRaises(self.helper.FsContractError): self.helper.audit_runtime_processes(os.fspath(self.runtime)) self.assertIsNone(process.poll(), "audit killed a process it does not own") finally: process.terminate() process.wait(timeout=5) os.close(runtime_fd) if __name__ == "__main__": unittest.main(verbosity=2)