1398 lines
62 KiB
Python
1398 lines
62 KiB
Python
#!/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)
|