1180 lines
49 KiB
Python
1180 lines
49 KiB
Python
#!/usr/bin/env python3
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import hashlib
|
|
import os
|
|
import pathlib
|
|
import secrets
|
|
import shutil
|
|
import signal
|
|
import socket
|
|
import stat
|
|
import subprocess
|
|
import tempfile
|
|
import termios
|
|
import threading
|
|
import time
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
|
|
ROOT = pathlib.Path(__file__).resolve().parents[2]
|
|
PROD = ROOT / "scripts/libexec/slack-webhook-recovery-vault.py"
|
|
KEEPASSXC_CLI = "/usr/bin/keepassxc-cli"
|
|
|
|
|
|
def load_vault_module():
|
|
spec = importlib.util.spec_from_file_location("slack_webhook_recovery_vault", PROD)
|
|
if spec is None or spec.loader is None:
|
|
raise AssertionError("KeePass database classifier is unavailable")
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def synthetic_webhook(suffix: bytes = b"A") -> bytes:
|
|
prefix = b"https://" + b"hooks." + b"slack.com/services/"
|
|
return prefix + b"T01234567/B01234567/" + (b"X" * 23) + suffix
|
|
|
|
|
|
def write_private(path: pathlib.Path, payload: bytes) -> None:
|
|
path.write_bytes(payload)
|
|
os.chmod(path, 0o600)
|
|
|
|
|
|
def private_keepass_environment() -> dict[str, str]:
|
|
return {
|
|
"HOME": os.devnull,
|
|
"XDG_CONFIG_HOME": os.devnull,
|
|
"LC_ALL": "C.UTF-8",
|
|
"PATH": "/usr/bin:/bin",
|
|
}
|
|
|
|
|
|
def create_database(path: pathlib.Path, master: bytearray) -> bytes:
|
|
child_input = bytearray(master)
|
|
child_input.extend(b"\n")
|
|
child_input.extend(master)
|
|
child_input.extend(b"\n")
|
|
try:
|
|
completed = subprocess.run(
|
|
[KEEPASSXC_CLI, "db-create", "-q", "-p", os.fspath(path)],
|
|
input=bytes(child_input),
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
env=private_keepass_environment(),
|
|
timeout=20,
|
|
check=False,
|
|
)
|
|
finally:
|
|
for index in range(len(child_input)):
|
|
child_input[index] = 0
|
|
if completed.returncode != 0:
|
|
raise AssertionError("synthetic KeePass database creation failed")
|
|
if bytes(master) in completed.stdout + completed.stderr:
|
|
raise AssertionError("synthetic master appeared in captured KeePassXC output")
|
|
os.chmod(path, 0o600)
|
|
return completed.stdout + completed.stderr
|
|
|
|
|
|
class VaultClassifierRedTest(unittest.TestCase):
|
|
"""Initial Task 3 gate using a real database with synthetic credentials."""
|
|
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
cls.fixture = tempfile.TemporaryDirectory(
|
|
prefix="slack-webhook-recovery-vault-test."
|
|
)
|
|
cls.fixture_path = pathlib.Path(cls.fixture.name)
|
|
os.chmod(cls.fixture_path, 0o700)
|
|
cls.database = cls.fixture_path / "test.kdbx"
|
|
cls.master = bytearray(secrets.token_bytes(32).hex().encode("ascii"))
|
|
cls.db_create_output = create_database(cls.database, cls.master)
|
|
|
|
@classmethod
|
|
def tearDownClass(cls) -> None:
|
|
for index in range(len(cls.master)):
|
|
cls.master[index] = 0
|
|
cls.fixture.cleanup()
|
|
|
|
def test_01_real_synthetic_kdbx_fixture_is_private_and_output_is_safe(self) -> None:
|
|
# Production break caught: a test harness that substitutes fake database
|
|
# bytes or leaks the synthetic master through KeePassXC output.
|
|
database_stat = self.database.stat()
|
|
self.assertTrue(stat.S_ISREG(database_stat.st_mode))
|
|
self.assertEqual(stat.S_IMODE(database_stat.st_mode), 0o600)
|
|
if bytes(self.master) in self.db_create_output:
|
|
self.fail("synthetic master appeared in captured KeePassXC output")
|
|
|
|
def test_02_vault_classifier_api_is_available(self) -> None:
|
|
# Production break caught: Task 1 can reach execute mode without the
|
|
# long-lived vault classifier boundary being implemented.
|
|
if not PROD.is_file():
|
|
self.fail("KeePass database classifier is unavailable")
|
|
module = load_vault_module()
|
|
required = (
|
|
"open_validated_webhook",
|
|
"read_operator_secrets",
|
|
"run_keepass",
|
|
"classify_database",
|
|
"build_candidate",
|
|
"verify_database",
|
|
"serve_private_socket",
|
|
)
|
|
if any(not callable(getattr(module, name, None)) for name in required):
|
|
self.fail("KeePass database classifier is unavailable")
|
|
|
|
|
|
@unittest.skipUnless(PROD.is_file(), "production vault helper is not implemented yet")
|
|
class WebhookAndInputContractTest(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
cls.vault = load_vault_module()
|
|
|
|
def setUp(self) -> None:
|
|
self.fixture = tempfile.TemporaryDirectory(
|
|
prefix="slack-webhook-recovery-webhook-test."
|
|
)
|
|
self.root = pathlib.Path(self.fixture.name)
|
|
os.chmod(self.root, 0o700)
|
|
|
|
def tearDown(self) -> None:
|
|
self.fixture.cleanup()
|
|
|
|
def test_valid_webhook_retains_one_fd_and_rejects_content_drift(self) -> None:
|
|
# Production break caught: reopening the path or accepting changed bytes
|
|
# after startup instead of binding the retained descriptor and content.
|
|
path = self.root / "webhook"
|
|
write_private(path, synthetic_webhook())
|
|
retained = self.vault.open_validated_webhook(os.fspath(path))
|
|
try:
|
|
self.assertEqual(retained.read_bytes(), synthetic_webhook())
|
|
with path.open("r+b", buffering=0) as stream:
|
|
stream.seek(-1, os.SEEK_END)
|
|
stream.write(b"B")
|
|
with self.assertRaises(self.vault.VaultContractError):
|
|
retained.read_bytes()
|
|
finally:
|
|
retained.close()
|
|
|
|
def test_retained_webhook_uses_mutable_storage_and_wipes_on_close(self) -> None:
|
|
path = self.root / "webhook-wipe"
|
|
write_private(path, synthetic_webhook())
|
|
retained = self.vault.open_validated_webhook(os.fspath(path))
|
|
content_reference = retained._content
|
|
self.assertIsInstance(content_reference, bytearray)
|
|
retained.close()
|
|
self.assertEqual(content_reference, bytearray(len(content_reference)))
|
|
|
|
def test_startup_failure_wipes_retained_webhook_storage(self) -> None:
|
|
path = self.root / "webhook-startup-wipe"
|
|
write_private(path, synthetic_webhook())
|
|
captured = []
|
|
real_open = self.vault.open_validated_webhook
|
|
|
|
def capture_open(webhook_path: str):
|
|
retained = real_open(webhook_path)
|
|
captured.append(retained._content)
|
|
return retained
|
|
|
|
output_fd = os.open(os.devnull, os.O_WRONLY | os.O_CLOEXEC)
|
|
try:
|
|
with mock.patch.object(
|
|
self.vault, "_attest_runtime_root", return_value=os.fspath(self.root)
|
|
), mock.patch.object(
|
|
self.vault, "open_validated_webhook", side_effect=capture_open
|
|
), mock.patch.object(
|
|
self.vault, "read_operator_secrets",
|
|
side_effect=self.vault.VaultContractError("synthetic startup failure"),
|
|
):
|
|
with self.assertRaises(self.vault.VaultContractError):
|
|
self.vault.serve_private_socket(
|
|
os.fspath(self.root), os.fspath(path),
|
|
input_fd=0, output_fd=output_fd,
|
|
)
|
|
finally:
|
|
os.close(output_fd)
|
|
self.assertEqual(len(captured), 1)
|
|
self.assertEqual(captured[0], bytearray(len(captured[0])))
|
|
|
|
def test_socket_constructor_failure_finalizes_retained_resources(self) -> None:
|
|
path = self.root / "webhook-socket-constructor-failure"
|
|
write_private(path, synthetic_webhook())
|
|
captured = []
|
|
wiped_master = []
|
|
real_open = self.vault.open_validated_webhook
|
|
|
|
def capture_open(webhook_path: str):
|
|
retained = real_open(webhook_path)
|
|
captured.append(retained)
|
|
return retained
|
|
|
|
with mock.patch.object(
|
|
self.vault, "_attest_runtime_root", return_value=os.fspath(self.root)
|
|
), mock.patch.object(
|
|
self.vault, "open_validated_webhook", side_effect=capture_open
|
|
), mock.patch.object(
|
|
self.vault.socket, "socket", side_effect=OSError("synthetic socket failure")
|
|
):
|
|
with self.assertRaises(OSError):
|
|
self.vault.serve_private_socket(
|
|
os.fspath(self.root), os.fspath(path),
|
|
wipe_hook=lambda value: wiped_master.append(bytes(value)),
|
|
)
|
|
self.assertEqual(len(captured), 1)
|
|
self.assertTrue(captured[0]._closed)
|
|
self.assertEqual(
|
|
captured[0]._content, bytearray(len(captured[0]._content))
|
|
)
|
|
self.assertEqual(wiped_master, [b""])
|
|
|
|
def test_retained_webhook_rejects_same_size_drift_during_read(self) -> None:
|
|
# Production break caught: checking only size/identity around a read,
|
|
# which misses an in-place same-size credential change during that read.
|
|
path = self.root / "webhook-concurrent"
|
|
write_private(path, synthetic_webhook(b"A"))
|
|
retained = self.vault.open_validated_webhook(os.fspath(path))
|
|
real_read = self.vault.os.read
|
|
mutated = False
|
|
|
|
def drifting_read(fd: int, count: int) -> bytes:
|
|
nonlocal mutated
|
|
chunk = real_read(fd, count)
|
|
if chunk and not mutated:
|
|
mutated = True
|
|
with path.open("r+b", buffering=0) as stream:
|
|
stream.seek(-1, os.SEEK_END)
|
|
stream.write(b"B")
|
|
return chunk
|
|
|
|
try:
|
|
with mock.patch.object(self.vault.os, "read", side_effect=drifting_read):
|
|
with self.assertRaises(self.vault.VaultContractError):
|
|
retained.read_bytes()
|
|
finally:
|
|
retained.close()
|
|
|
|
def test_open_webhook_rejects_same_size_drift_during_startup_binding(self) -> None:
|
|
# Production break caught: startup content binding that performs only
|
|
# one read and misses an in-place same-size change during that read.
|
|
path = self.root / "webhook-open-concurrent"
|
|
write_private(path, synthetic_webhook(b"A"))
|
|
real_read = self.vault.os.read
|
|
mutated = False
|
|
|
|
def drifting_read(fd: int, count: int) -> bytes:
|
|
nonlocal mutated
|
|
chunk = real_read(fd, count)
|
|
if chunk and not mutated:
|
|
mutated = True
|
|
with path.open("r+b", buffering=0) as stream:
|
|
stream.seek(-1, os.SEEK_END)
|
|
stream.write(b"B")
|
|
return chunk
|
|
|
|
with mock.patch.object(self.vault.os, "read", side_effect=drifting_read):
|
|
with self.assertRaises(self.vault.VaultContractError):
|
|
self.vault.open_validated_webhook(os.fspath(path))
|
|
|
|
def test_path_replacement_never_switches_the_retained_webhook(self) -> None:
|
|
# Production break caught: a consumer reopening a replaced pathname and
|
|
# silently consuming the replacement credential.
|
|
path = self.root / "webhook"
|
|
replacement = self.root / "replacement"
|
|
write_private(path, synthetic_webhook(b"A"))
|
|
write_private(replacement, synthetic_webhook(b"B"))
|
|
retained = self.vault.open_validated_webhook(os.fspath(path))
|
|
try:
|
|
os.replace(replacement, path)
|
|
with self.assertRaises(self.vault.VaultContractError):
|
|
retained.read_bytes()
|
|
finally:
|
|
retained.close()
|
|
|
|
def test_webhook_metadata_and_content_rejection_matrix(self) -> None:
|
|
# Production breaks caught: weakening any member of the exact metadata,
|
|
# one-line, bounded-content, or Slack URL grammar matrix.
|
|
valid = synthetic_webhook()
|
|
cases = {
|
|
"empty": b"",
|
|
"oversized": b"A" * 4097,
|
|
"carriage-return": valid + b"\r",
|
|
"line-feed": valid + b"\n",
|
|
"nul": valid + b"\0",
|
|
"leading-space": b" " + valid,
|
|
"trailing-space": valid + b" ",
|
|
"wrong-host": valid.replace(b"hooks.slack.com", b"example.invalid"),
|
|
"two-components": (b"https://" + b"hooks." + b"slack.com/services/T/B"),
|
|
"extra-component": valid + b"/extra",
|
|
"query": valid + b"?x=1",
|
|
}
|
|
for name, payload in cases.items():
|
|
with self.subTest(case=name):
|
|
path = self.root / name
|
|
write_private(path, payload)
|
|
with self.assertRaises(self.vault.VaultContractError):
|
|
self.vault.open_validated_webhook(os.fspath(path))
|
|
|
|
wrong_mode = self.root / "wrong-mode"
|
|
write_private(wrong_mode, valid)
|
|
os.chmod(wrong_mode, 0o640)
|
|
with self.assertRaises(self.vault.VaultContractError):
|
|
self.vault.open_validated_webhook(os.fspath(wrong_mode))
|
|
|
|
original = self.root / "hardlink-original"
|
|
linked = self.root / "hardlink"
|
|
write_private(original, valid)
|
|
os.link(original, linked)
|
|
with self.assertRaises(self.vault.VaultContractError):
|
|
self.vault.open_validated_webhook(os.fspath(linked))
|
|
|
|
target = self.root / "symlink-target"
|
|
symlink = self.root / "symlink"
|
|
write_private(target, valid)
|
|
symlink.symlink_to(target)
|
|
with self.assertRaises((self.vault.VaultContractError, OSError)):
|
|
self.vault.open_validated_webhook(os.fspath(symlink))
|
|
|
|
directory = self.root / "directory"
|
|
directory.mkdir(mode=0o700)
|
|
with self.assertRaises((self.vault.VaultContractError, OSError)):
|
|
self.vault.open_validated_webhook(os.fspath(directory))
|
|
|
|
foreign = self.root / "foreign-owner"
|
|
write_private(foreign, valid)
|
|
with mock.patch.object(self.vault.os, "getuid", return_value=os.getuid() + 1):
|
|
with self.assertRaises(self.vault.VaultContractError):
|
|
self.vault.open_validated_webhook(os.fspath(foreign))
|
|
|
|
def test_operator_input_validates_app_and_keeps_master_mutable(self) -> None:
|
|
# Production break caught: accepting notes delimiters or returning an
|
|
# immutable master value that cannot be best-effort overwritten.
|
|
read_fd, write_fd = os.pipe()
|
|
output_read_fd, output_fd = os.pipe()
|
|
master = secrets.token_bytes(24).hex().encode("ascii")
|
|
os.write(write_fd, b"Synthetic Slack App\n" + master + b"\n")
|
|
os.close(write_fd)
|
|
try:
|
|
app_name, master_buffer = self.vault.read_operator_secrets(
|
|
input_fd=read_fd, output_fd=output_fd
|
|
)
|
|
finally:
|
|
os.close(read_fd)
|
|
os.close(output_fd)
|
|
prompts = os.read(output_read_fd, 4096)
|
|
os.close(output_read_fd)
|
|
self.assertEqual(
|
|
prompts,
|
|
b"Slack app name: KeePassXC master password: ",
|
|
)
|
|
self.assertEqual(app_name, "Synthetic Slack App")
|
|
self.assertIsInstance(master_buffer, bytearray)
|
|
self.assertEqual(bytes(master_buffer), master)
|
|
self.vault.wipe_mutable_buffer(master_buffer)
|
|
self.assertEqual(bytes(master_buffer), b"\0" * len(master))
|
|
|
|
def test_signal_during_hidden_input_restores_terminal_echo(self) -> None:
|
|
# Production break caught: the coordinator's bounded startup abort
|
|
# terminates the vault while ECHO is disabled and leaves the operator's
|
|
# terminal in a secret-hostile state.
|
|
master_fd, slave_fd = os.openpty()
|
|
child_pid = os.fork()
|
|
if child_pid == 0:
|
|
try:
|
|
os.close(master_fd)
|
|
self.vault.install_process_signal_handlers()
|
|
self.vault.read_operator_secrets(
|
|
input_fd=slave_fd, output_fd=slave_fd
|
|
)
|
|
except BaseException:
|
|
os._exit(0)
|
|
os._exit(80)
|
|
os.close(slave_fd)
|
|
try:
|
|
deadline = time.monotonic() + 5.0
|
|
observed = b""
|
|
while b"Slack app name: " not in observed and time.monotonic() < deadline:
|
|
observed += os.read(master_fd, 256)
|
|
os.write(master_fd, b"Synthetic Slack App\n")
|
|
while b"KeePassXC master password: " not in observed and time.monotonic() < deadline:
|
|
observed += os.read(master_fd, 256)
|
|
attrs = termios.tcgetattr(master_fd)
|
|
self.assertFalse(attrs[3] & termios.ECHO)
|
|
os.kill(child_pid, signal.SIGTERM)
|
|
waited_pid, status = os.waitpid(child_pid, 0)
|
|
child_pid = -1
|
|
self.assertEqual(waited_pid > 0, True)
|
|
self.assertTrue(os.WIFEXITED(status))
|
|
self.assertEqual(os.WEXITSTATUS(status), 0)
|
|
restored = termios.tcgetattr(master_fd)
|
|
self.assertTrue(restored[3] & termios.ECHO)
|
|
finally:
|
|
if child_pid > 0:
|
|
os.kill(child_pid, signal.SIGKILL)
|
|
os.waitpid(child_pid, 0)
|
|
os.close(master_fd)
|
|
|
|
def test_socket_is_not_published_until_operator_inputs_are_retained(self) -> None:
|
|
# Production break caught: sudo and the vault racing each other for
|
|
# /dev/tty because socket publication precedes secret retention.
|
|
read_fd, write_fd = os.pipe()
|
|
output_fd = os.open(os.devnull, os.O_WRONLY | os.O_CLOEXEC)
|
|
failures: list[str] = []
|
|
|
|
def target() -> None:
|
|
try:
|
|
self.vault.serve_private_socket(
|
|
os.fspath(self.root),
|
|
os.fspath(self.root / "webhook"),
|
|
input_fd=read_fd,
|
|
output_fd=output_fd,
|
|
accept_timeout=5.0,
|
|
io_timeout=2.0,
|
|
child_timeout=5.0,
|
|
)
|
|
except BaseException as exc:
|
|
failures.append(f"{type(exc).__name__}:{exc}")
|
|
finally:
|
|
os.close(read_fd)
|
|
os.close(output_fd)
|
|
|
|
write_private(self.root / "webhook", synthetic_webhook())
|
|
attest_patch = mock.patch.object(
|
|
self.vault, "_attest_runtime_root", return_value=os.fspath(self.root)
|
|
)
|
|
attest_patch.start()
|
|
thread = threading.Thread(target=target, daemon=True)
|
|
thread.start()
|
|
try:
|
|
time.sleep(0.1)
|
|
self.assertFalse((self.root / "vault.sock").exists())
|
|
self.assertFalse(
|
|
(self.root / "vault-home").exists(),
|
|
"writable KeePass private home was created before input retention",
|
|
)
|
|
os.write(write_fd, b"Synthetic Slack App\nsynthetic-master\n")
|
|
os.close(write_fd)
|
|
write_fd = -1
|
|
deadline = time.monotonic() + 5
|
|
while not (self.root / "vault.sock").exists() and time.monotonic() < deadline:
|
|
time.sleep(0.01)
|
|
self.assertTrue((self.root / "vault.sock").exists())
|
|
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as client:
|
|
client.connect(os.fspath(self.root / "vault.sock"))
|
|
client.sendall(b"SHUTDOWN\n")
|
|
self.assertEqual(client.recv(32), b"stopped\n")
|
|
thread.join(timeout=5)
|
|
self.assertFalse(thread.is_alive())
|
|
self.assertEqual(failures, [])
|
|
finally:
|
|
attest_patch.stop()
|
|
if write_fd >= 0:
|
|
os.close(write_fd)
|
|
|
|
def test_keepass_child_arms_linux_parent_death_signal(self) -> None:
|
|
# Production break caught: a vault killed during communicate leaves its
|
|
# separate-session KeePass child holding secret stdin alive.
|
|
self.assertTrue(callable(getattr(self.vault, "arm_parent_death_signal", None)))
|
|
fake_prctl = mock.Mock(return_value=0)
|
|
fake_libc = mock.Mock(prctl=fake_prctl)
|
|
with mock.patch.object(self.vault.os, "getppid", side_effect=[1234, 1234]), \
|
|
mock.patch.object(self.vault.ctypes, "CDLL", return_value=fake_libc):
|
|
self.vault.arm_parent_death_signal(1234)
|
|
fake_prctl.assert_called_once_with(1, 9, 0, 0, 0)
|
|
with mock.patch.object(self.vault.os, "getppid", return_value=1), \
|
|
mock.patch.object(self.vault.os, "kill") as kill_parent_race:
|
|
self.vault.arm_parent_death_signal(1234)
|
|
kill_parent_race.assert_called_once_with(os.getpid(), 9)
|
|
|
|
def test_parent_death_signal_kills_actual_fake_child(self) -> None:
|
|
script = r'''
|
|
import importlib.util, os, subprocess, sys, time
|
|
spec = importlib.util.spec_from_file_location("vault", sys.argv[1])
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
expected = os.getpid()
|
|
child = subprocess.Popen(
|
|
[sys.executable, "-c", "import time; time.sleep(30)"],
|
|
preexec_fn=lambda: module.arm_parent_death_signal(expected),
|
|
)
|
|
print(child.pid, flush=True)
|
|
'''
|
|
parent = subprocess.Popen(
|
|
[os.fspath(pathlib.Path(os.sys.executable)), "-c", script, os.fspath(PROD)],
|
|
stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
|
|
)
|
|
child_pid = int(parent.stdout.readline().strip())
|
|
parent.communicate(timeout=5)
|
|
deadline = time.monotonic() + 5
|
|
while time.monotonic() < deadline:
|
|
try:
|
|
status = pathlib.Path(f"/proc/{child_pid}/status").read_text()
|
|
except (FileNotFoundError, ProcessLookupError):
|
|
break
|
|
if "State:\tZ" in status:
|
|
break
|
|
time.sleep(0.02)
|
|
else:
|
|
os.kill(child_pid, 9)
|
|
self.fail("PDEATHSIG did not terminate the actual fake child")
|
|
|
|
def test_operator_input_rejects_app_note_delimiters_and_whitespace(self) -> None:
|
|
# Production break caught: app text escaping its single-line Notes key.
|
|
invalid_apps = (b"", b" leading", b"trailing ", b"a;b", b"a=b", b"a\\b", b"a\rb")
|
|
for app in invalid_apps:
|
|
with self.subTest(app_case=repr(app)):
|
|
read_fd, write_fd = os.pipe()
|
|
output_fd = os.open(os.devnull, os.O_WRONLY | os.O_CLOEXEC)
|
|
os.write(write_fd, app + b"\nsynthetic-master\n")
|
|
os.close(write_fd)
|
|
try:
|
|
with self.assertRaises(self.vault.VaultContractError):
|
|
self.vault.read_operator_secrets(
|
|
input_fd=read_fd, output_fd=output_fd
|
|
)
|
|
finally:
|
|
os.close(read_fd)
|
|
os.close(output_fd)
|
|
|
|
def test_protocol_line_rejects_cr_nul_oversize_partial_and_pipeline(self) -> None:
|
|
# Production breaks caught: request framing ambiguity or unbounded reads.
|
|
cases = (
|
|
b"PREPARE\r\n",
|
|
b"PREPARE\0\n",
|
|
b"A" * 4097 + b"\n",
|
|
b"PREPARE",
|
|
b"PREPARE\nSHUTDOWN\n",
|
|
)
|
|
for payload in cases:
|
|
with self.subTest(payload_length=len(payload)):
|
|
server, client = socket.socketpair(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
try:
|
|
client.sendall(payload)
|
|
client.shutdown(socket.SHUT_WR)
|
|
with self.assertRaises(self.vault.VaultProtocolError):
|
|
self.vault.read_protocol_line(server, timeout=0.5)
|
|
finally:
|
|
server.close()
|
|
client.close()
|
|
|
|
def test_parser_accepts_only_exact_commands_and_state(self) -> None:
|
|
# Production break caught: protocol fields selecting candidate/baseline
|
|
# names or bypassing the BUILD state gate.
|
|
valid = {
|
|
"PREPARE": ("PREPARE", None),
|
|
"BUILD ADD": ("BUILD", "ADD"),
|
|
"BUILD UPDATE": ("BUILD", "UPDATE"),
|
|
"VERIFY_CANDIDATE": ("VERIFY_CANDIDATE", None),
|
|
"SHUTDOWN": ("SHUTDOWN", None),
|
|
}
|
|
for line, expected in valid.items():
|
|
self.assertEqual(self.vault.parse_protocol_command(line), expected)
|
|
for line in (
|
|
"",
|
|
"PREPARE extra",
|
|
"BUILD",
|
|
"BUILD ADD extra",
|
|
"BUILD DELETE",
|
|
"SHUTDOWN extra",
|
|
"baseline.kdbx",
|
|
):
|
|
with self.subTest(line=line):
|
|
with self.assertRaises(self.vault.VaultProtocolError):
|
|
self.vault.parse_protocol_command(line)
|
|
|
|
def test_direct_cli_is_closed_and_has_no_environment_input_seam(self) -> None:
|
|
# Production break caught: adding a direct classifier/debug form or an
|
|
# environment-activated replacement for the mandatory /dev/tty input.
|
|
runtime_parent = pathlib.Path(f"/run/user/{os.getuid()}")
|
|
with tempfile.TemporaryDirectory(
|
|
prefix="slack-webhook-recovery-cli.", dir=runtime_parent
|
|
) as raw_runtime:
|
|
runtime = pathlib.Path(raw_runtime)
|
|
os.chmod(runtime, 0o700)
|
|
webhook = runtime / "input-webhook"
|
|
write_private(webhook, synthetic_webhook())
|
|
marker = b"synthetic-stdin-must-not-be-an-input-seam"
|
|
environment = {
|
|
"HOME": "/hostile-home",
|
|
"LC_ALL": "C.UTF-8",
|
|
"PATH": "/usr/bin:/bin",
|
|
"SWR_VAULT_INPUT_FD": "0",
|
|
}
|
|
invalid = subprocess.run(
|
|
[PYTHON := "/usr/bin/python3", os.fspath(PROD), "--serve"],
|
|
input=marker + b"\n",
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
env=environment,
|
|
timeout=5,
|
|
check=False,
|
|
)
|
|
self.assertEqual(invalid.returncode, 2)
|
|
self.assertEqual(invalid.stdout + invalid.stderr, b"")
|
|
|
|
direct = subprocess.run(
|
|
[
|
|
PYTHON, os.fspath(PROD), "--serve", "--runtime-root",
|
|
os.fspath(runtime), "--webhook-file", os.fspath(webhook),
|
|
],
|
|
input=b"Synthetic App\n" + marker + b"\n",
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
env=environment,
|
|
timeout=5,
|
|
check=False,
|
|
)
|
|
self.assertEqual(direct.returncode, 1)
|
|
self.assertEqual(direct.stdout, b"")
|
|
self.assertEqual(direct.stderr, b"vault helper failed\n")
|
|
self.assertNotIn(marker, direct.stdout + direct.stderr)
|
|
|
|
def test_direct_cli_preserves_first_signal_status_after_cleanup(self) -> None:
|
|
# Production break caught: translating the coordinator's forwarded
|
|
# HUP/INT/TERM into generic failure after terminal/resource cleanup.
|
|
for signum in (signal.SIGHUP, signal.SIGINT, signal.SIGTERM):
|
|
with self.subTest(signum=signum), mock.patch.object(
|
|
self.vault, "install_process_signal_handlers"
|
|
), mock.patch.object(
|
|
self.vault, "serve_private_socket",
|
|
side_effect=self.vault.VaultSignal(signum),
|
|
):
|
|
rc = self.vault._main(
|
|
["--serve", "--runtime-root", "/synthetic/runtime",
|
|
"--webhook-file", "/synthetic/webhook"]
|
|
)
|
|
self.assertEqual(rc, 128 + signum)
|
|
|
|
def test_committed_snapshot_validation_matrix(self) -> None:
|
|
# Production breaks caught: verifying a symlink, hardlink, wrong-mode,
|
|
# foreign-owner, or out-of-runtime snapshot.
|
|
runtime_parent = pathlib.Path(f"/run/user/{os.getuid()}")
|
|
with tempfile.TemporaryDirectory(
|
|
prefix="slack-webhook-recovery-snapshot.", dir=runtime_parent
|
|
) as raw_runtime:
|
|
runtime = pathlib.Path(raw_runtime)
|
|
os.chmod(runtime, 0o700)
|
|
valid = runtime / "candidate.kdbx"
|
|
write_private(valid, b"synthetic-encrypted-database")
|
|
self.vault._attest_snapshot(os.fspath(valid), os.fspath(runtime))
|
|
|
|
wrong_mode = runtime / "wrong-mode.kdbx"
|
|
write_private(wrong_mode, b"synthetic")
|
|
os.chmod(wrong_mode, 0o640)
|
|
with self.assertRaises(self.vault.VaultContractError):
|
|
self.vault._attest_snapshot(os.fspath(wrong_mode), os.fspath(runtime))
|
|
|
|
original = runtime / "linked-original.kdbx"
|
|
hardlink = runtime / "hardlink.kdbx"
|
|
write_private(original, b"synthetic")
|
|
os.link(original, hardlink)
|
|
with self.assertRaises(self.vault.VaultContractError):
|
|
self.vault._attest_snapshot(os.fspath(hardlink), os.fspath(runtime))
|
|
|
|
symlink = runtime / "symlink.kdbx"
|
|
symlink.symlink_to(valid)
|
|
with self.assertRaises(self.vault.VaultContractError):
|
|
self.vault._attest_snapshot(os.fspath(symlink), os.fspath(runtime))
|
|
|
|
with mock.patch.object(self.vault.os, "getuid", return_value=os.getuid() + 1):
|
|
with self.assertRaises(self.vault.VaultContractError):
|
|
self.vault._attest_snapshot(os.fspath(valid), os.fspath(runtime))
|
|
|
|
outside = self.root / "outside.kdbx"
|
|
write_private(outside, b"synthetic")
|
|
with self.assertRaises(self.vault.VaultProtocolError):
|
|
self.vault._attest_snapshot(os.fspath(outside), os.fspath(runtime))
|
|
|
|
def test_keepass_child_timeout_status_and_environment_are_sanitized(self) -> None:
|
|
# Production breaks caught: inherited hostile environment, unchecked
|
|
# child status, or an unbounded KeePass child.
|
|
script = self.root / "fake-keepass"
|
|
script.write_text(
|
|
"#!/usr/bin/python3\n"
|
|
"import os,sys,time\n"
|
|
"data=sys.stdin.buffer.read()\n"
|
|
"mode=sys.argv[1]\n"
|
|
"if mode=='timeout': time.sleep(2)\n"
|
|
"if mode=='status': sys.stderr.write('synthetic failure\\n'); raise SystemExit(7)\n"
|
|
"sys.stdout.write('\\n'.join(sorted(os.environ)))\n",
|
|
encoding="utf-8",
|
|
)
|
|
os.chmod(script, 0o700)
|
|
home = self.root / "home"
|
|
home.mkdir(mode=0o700)
|
|
master = bytearray(b"synthetic-master")
|
|
with mock.patch.object(self.vault, "KEEPASSXC_CLI", os.fspath(script)):
|
|
with self.assertRaises(self.vault.VaultContractError):
|
|
self.vault.run_keepass(
|
|
["timeout"], master, os.fspath(home), timeout=0.05
|
|
)
|
|
with self.assertRaises(self.vault.VaultContractError):
|
|
self.vault.run_keepass(["status"], master, os.fspath(home))
|
|
result = self.vault.run_keepass(
|
|
["environment"], master, os.fspath(home), check=True
|
|
)
|
|
self.assertEqual(result.stdout, b"HOME\nLC_ALL\nPATH\nXDG_CONFIG_HOME")
|
|
self.assertNotIn(bytes(master), result.stdout + result.stderr)
|
|
|
|
|
|
@unittest.skipUnless(PROD.is_file(), "production vault helper is not implemented yet")
|
|
class RealSyntheticDatabaseContractTest(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
cls.vault = load_vault_module()
|
|
|
|
def setUp(self) -> None:
|
|
runtime_parent = pathlib.Path(f"/run/user/{os.getuid()}")
|
|
self.fixture = tempfile.TemporaryDirectory(
|
|
prefix="slack-webhook-recovery-vault-kdbx.", dir=runtime_parent
|
|
)
|
|
self.root = pathlib.Path(self.fixture.name)
|
|
os.chmod(self.root, 0o700)
|
|
self.home = pathlib.Path(os.devnull)
|
|
self.master = bytearray(secrets.token_bytes(32).hex().encode("ascii"))
|
|
self.baseline = self.root / "baseline.kdbx"
|
|
create_database(self.baseline, self.master)
|
|
self.webhook_path = self.root / "input-webhook"
|
|
write_private(self.webhook_path, synthetic_webhook())
|
|
self.webhook = self.vault.open_validated_webhook(os.fspath(self.webhook_path))
|
|
self.app = "Synthetic Slack App"
|
|
|
|
def tearDown(self) -> None:
|
|
self.webhook.close()
|
|
self.vault.wipe_mutable_buffer(self.master)
|
|
self.fixture.cleanup()
|
|
|
|
def _classify(self, database: pathlib.Path, app: str | None = None) -> str:
|
|
return self.vault.classify_database(
|
|
os.fspath(database),
|
|
self.master,
|
|
self.webhook,
|
|
self.app if app is None else app,
|
|
os.fspath(self.home),
|
|
)
|
|
|
|
def _build(self, baseline: pathlib.Path, candidate: pathlib.Path, mode: str) -> None:
|
|
self.vault.build_candidate(
|
|
os.fspath(baseline),
|
|
os.fspath(candidate),
|
|
mode,
|
|
self.master,
|
|
self.webhook,
|
|
self.app,
|
|
os.fspath(self.home),
|
|
)
|
|
|
|
def test_keepass_cli_operates_without_a_writable_private_home(self) -> None:
|
|
# Architecture characterization: the installed KeePassXC CLI must not
|
|
# require a writable HOME/XDG config tree for real synthetic KDBX work.
|
|
before = hashlib.sha256(self.baseline.read_bytes()).digest()
|
|
candidate = self.root / "configless-candidate.kdbx"
|
|
shutil.copyfile(self.baseline, candidate)
|
|
os.chmod(candidate, 0o600)
|
|
environment = {
|
|
"HOME": os.devnull,
|
|
"XDG_CONFIG_HOME": os.devnull,
|
|
"LC_ALL": "C.UTF-8",
|
|
"PATH": "/usr/bin:/bin",
|
|
}
|
|
|
|
listed = subprocess.run(
|
|
[KEEPASSXC_CLI, "ls", "-q", "-R", "-f", os.fspath(candidate)],
|
|
input=bytes(self.master) + b"\n",
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
env=environment,
|
|
timeout=20,
|
|
check=False,
|
|
)
|
|
added = subprocess.run(
|
|
[
|
|
KEEPASSXC_CLI, "add", "-q", "-p", "-u", "synthetic-user",
|
|
"--notes", "synthetic-notes", os.fspath(candidate),
|
|
"/SyntheticConfiglessEntry",
|
|
],
|
|
input=bytes(self.master) + b"\nsynthetic-password\n",
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
env=environment,
|
|
timeout=20,
|
|
check=False,
|
|
)
|
|
edited = subprocess.run(
|
|
[
|
|
KEEPASSXC_CLI, "edit", "-q", "-t", "SyntheticConfiglessEdited",
|
|
os.fspath(candidate), "/SyntheticConfiglessEntry",
|
|
],
|
|
input=bytes(self.master) + b"\n",
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
env=environment,
|
|
timeout=20,
|
|
check=False,
|
|
)
|
|
shown = subprocess.run(
|
|
[
|
|
KEEPASSXC_CLI, "show", "-q", "-a", "Title",
|
|
os.fspath(candidate), "/SyntheticConfiglessEdited",
|
|
],
|
|
input=bytes(self.master) + b"\n",
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
env=environment,
|
|
timeout=20,
|
|
check=False,
|
|
)
|
|
for result in (listed, added, edited, shown):
|
|
self.assertEqual(result.returncode, 0, result.stderr.decode("utf-8", "replace"))
|
|
self.assertEqual(result.stderr, b"")
|
|
self.assertNotIn(bytes(self.master), result.stdout + result.stderr)
|
|
self.assertEqual(hashlib.sha256(self.baseline.read_bytes()).digest(), before)
|
|
self.assertEqual(shown.stdout, b"SyntheticConfiglessEdited\n")
|
|
self.assertFalse((self.root / "vault-home").exists())
|
|
|
|
def test_absent_add_exact_noop_and_wrong_master_zero_change(self) -> None:
|
|
# Production breaks caught: incorrect absent/exact classification,
|
|
# mutating the baseline, or treating invalid credentials as absence.
|
|
before = hashlib.sha256(self.baseline.read_bytes()).digest()
|
|
self.assertEqual(self._classify(self.baseline), "absent")
|
|
candidate = self.root / "candidate.kdbx"
|
|
self._build(self.baseline, candidate, "ADD")
|
|
self.assertEqual(hashlib.sha256(self.baseline.read_bytes()).digest(), before)
|
|
self.assertTrue(
|
|
self.vault.verify_database(
|
|
os.fspath(candidate),
|
|
self.master,
|
|
self.webhook,
|
|
self.app,
|
|
os.fspath(self.home),
|
|
)
|
|
)
|
|
self.assertEqual(self._classify(candidate), "exact-noop")
|
|
wrong_master = bytearray(b"wrong-synthetic-master")
|
|
candidate_before = hashlib.sha256(candidate.read_bytes()).digest()
|
|
self.assertEqual(
|
|
self.vault.classify_database(
|
|
os.fspath(candidate),
|
|
wrong_master,
|
|
self.webhook,
|
|
self.app,
|
|
os.fspath(self.home),
|
|
),
|
|
"failed",
|
|
)
|
|
self.assertEqual(hashlib.sha256(candidate.read_bytes()).digest(), candidate_before)
|
|
self.vault.wipe_mutable_buffer(wrong_master)
|
|
|
|
def test_metadata_password_app_workspace_channel_and_title_drift(self) -> None:
|
|
# Production break caught: comparing only the password or silently
|
|
# accepting any visible field drift.
|
|
exact = self.root / "exact.kdbx"
|
|
self._build(self.baseline, exact, "ADD")
|
|
self.assertEqual(self._classify(exact, app="Different App"), "mismatch")
|
|
|
|
entry = self.vault.ENTRY_PATH
|
|
drift_cases = (
|
|
("username", ["edit", "-q", "-u", "other-workspace", os.fspath(exact), entry]),
|
|
(
|
|
"notes",
|
|
["edit", "-q", "--notes", "channel=other;app=Synthetic Slack App;recovery=revoke-and-reissue-in-slack", os.fspath(exact), entry],
|
|
),
|
|
("url", ["edit", "-q", "--url", "https://example.invalid", os.fspath(exact), entry]),
|
|
)
|
|
for name, args in drift_cases:
|
|
with self.subTest(field=name):
|
|
drifted = self.root / f"{name}.kdbx"
|
|
shutil.copyfile(exact, drifted)
|
|
os.chmod(drifted, 0o600)
|
|
args = [value.replace(os.fspath(exact), os.fspath(drifted)) for value in args]
|
|
self.vault.run_keepass(args, self.master, os.fspath(self.home))
|
|
self.assertEqual(self._classify(drifted), "mismatch")
|
|
|
|
password_drift = self.root / "password.kdbx"
|
|
shutil.copyfile(exact, password_drift)
|
|
os.chmod(password_drift, 0o600)
|
|
other_path = self.root / "other-webhook"
|
|
write_private(other_path, synthetic_webhook(b"B"))
|
|
other = self.vault.open_validated_webhook(os.fspath(other_path))
|
|
try:
|
|
self.vault.run_keepass(
|
|
["edit", "-q", "-p", os.fspath(password_drift), entry],
|
|
self.master,
|
|
os.fspath(self.home),
|
|
webhook=other,
|
|
)
|
|
finally:
|
|
other.close()
|
|
self.assertEqual(self._classify(password_drift), "mismatch")
|
|
|
|
title_drift = self.root / "title.kdbx"
|
|
shutil.copyfile(exact, title_drift)
|
|
os.chmod(title_drift, 0o600)
|
|
self.vault.run_keepass(
|
|
["edit", "-q", "-t", "Different title", os.fspath(title_drift), entry],
|
|
self.master,
|
|
os.fspath(self.home),
|
|
)
|
|
self.assertEqual(self._classify(title_drift), "absent")
|
|
|
|
def test_duplicate_title_is_ambiguous_and_update_is_explicit(self) -> None:
|
|
# Production breaks caught: destructive duplicate reconciliation or
|
|
# editing a mismatch before the explicit BUILD UPDATE command.
|
|
exact = self.root / "exact.kdbx"
|
|
self._build(self.baseline, exact, "ADD")
|
|
duplicate = self.root / "duplicate.kdbx"
|
|
shutil.copyfile(exact, duplicate)
|
|
os.chmod(duplicate, 0o600)
|
|
independently_created = self.root / "independent.kdbx"
|
|
self._build(self.baseline, independently_created, "ADD")
|
|
self.vault.run_keepass(
|
|
["merge", "-q", "-s", os.fspath(duplicate), os.fspath(independently_created)],
|
|
self.master,
|
|
os.fspath(self.home),
|
|
)
|
|
self.assertEqual(self._classify(duplicate), "ambiguous")
|
|
|
|
mismatch = self.root / "mismatch.kdbx"
|
|
shutil.copyfile(exact, mismatch)
|
|
os.chmod(mismatch, 0o600)
|
|
self.vault.run_keepass(
|
|
["edit", "-q", "--notes", "channel=wrong;app=wrong;recovery=wrong", os.fspath(mismatch), self.vault.ENTRY_PATH],
|
|
self.master,
|
|
os.fspath(self.home),
|
|
)
|
|
before = hashlib.sha256(mismatch.read_bytes()).digest()
|
|
self.assertEqual(self._classify(mismatch), "mismatch")
|
|
self.assertEqual(hashlib.sha256(mismatch.read_bytes()).digest(), before)
|
|
updated = self.root / "updated.kdbx"
|
|
self._build(mismatch, updated, "UPDATE")
|
|
self.assertEqual(self._classify(updated), "exact-noop")
|
|
|
|
def test_protected_comparison_requires_producer_success_and_exact_final_lf(self) -> None:
|
|
# Production break caught: trimming KeePass output or ignoring producer
|
|
# status before protected byte comparison.
|
|
expected = synthetic_webhook()
|
|
self.assertTrue(self.vault.protected_value_matches(0, expected + b"\n", expected))
|
|
self.assertFalse(self.vault.protected_value_matches(1, expected + b"\n", expected))
|
|
self.assertFalse(self.vault.protected_value_matches(0, expected, expected))
|
|
self.assertFalse(self.vault.protected_value_matches(0, expected + b"\n\n", expected))
|
|
|
|
def test_private_socket_real_database_lifecycle_and_wipe(self) -> None:
|
|
# Production breaks caught: wrong socket mode/peer/protocol, secret-bearing
|
|
# replies, illegal state changes, or failure to wipe on shutdown.
|
|
read_fd, write_fd = os.pipe()
|
|
output_fd = os.open(os.devnull, os.O_WRONLY | os.O_CLOEXEC)
|
|
os.write(write_fd, self.app.encode("utf-8") + b"\n" + bytes(self.master) + b"\n")
|
|
os.close(write_fd)
|
|
wiped: list[bytes] = []
|
|
webhook_storage: list[bytearray] = []
|
|
failures: list[str] = []
|
|
|
|
def wipe_hook(buffer: bytearray) -> None:
|
|
wiped.append(bytes(buffer))
|
|
|
|
def target() -> None:
|
|
try:
|
|
real_open = self.vault.open_validated_webhook
|
|
def capture_open(path: str):
|
|
retained = real_open(path)
|
|
webhook_storage.append(retained._content)
|
|
return retained
|
|
with mock.patch.object(
|
|
self.vault, "open_validated_webhook", side_effect=capture_open
|
|
):
|
|
self.vault.serve_private_socket(
|
|
os.fspath(self.root),
|
|
os.fspath(self.webhook_path),
|
|
input_fd=read_fd,
|
|
output_fd=output_fd,
|
|
accept_timeout=5.0,
|
|
io_timeout=5.0,
|
|
child_timeout=20.0,
|
|
wipe_hook=wipe_hook,
|
|
)
|
|
except BaseException as exc:
|
|
failures.append(f"{type(exc).__name__}:{exc}")
|
|
finally:
|
|
os.close(read_fd)
|
|
os.close(output_fd)
|
|
|
|
thread = threading.Thread(target=target, daemon=True)
|
|
thread.start()
|
|
socket_path = self.root / "vault.sock"
|
|
deadline = time.monotonic() + 5
|
|
while not socket_path.exists() and time.monotonic() < deadline:
|
|
time.sleep(0.01)
|
|
self.assertTrue(socket_path.exists())
|
|
self.assertEqual(stat.S_IMODE(socket_path.stat().st_mode), 0o600)
|
|
|
|
responses = bytearray()
|
|
client = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
client.settimeout(10)
|
|
client.connect(os.fspath(socket_path))
|
|
try:
|
|
for command, expected in (
|
|
(b"PREPARE\n", b"absent\n"),
|
|
(b"BUILD ADD\n", b"candidate-ready\n"),
|
|
(b"VERIFY_CANDIDATE\n", b"verified\n"),
|
|
(b"VERIFY_COMMITTED " + os.fsencode(self.root / "candidate.kdbx") + b"\n", b"verified\n"),
|
|
(b"SHUTDOWN\n", b"stopped\n"),
|
|
):
|
|
client.sendall(command)
|
|
reply = b""
|
|
while not reply.endswith(b"\n"):
|
|
reply += client.recv(128)
|
|
responses.extend(reply)
|
|
self.assertEqual(reply, expected)
|
|
finally:
|
|
client.close()
|
|
thread.join(timeout=10)
|
|
self.assertFalse(thread.is_alive())
|
|
self.assertEqual(failures, [])
|
|
self.assertTrue(wiped)
|
|
self.assertEqual(wiped[-1], b"\0" * len(self.master))
|
|
self.assertEqual(len(webhook_storage), 1)
|
|
self.assertEqual(
|
|
webhook_storage[0], bytearray(len(webhook_storage[0]))
|
|
)
|
|
self.assertNotIn(bytes(self.master), responses)
|
|
self.assertNotIn(synthetic_webhook(), responses)
|
|
|
|
def test_foreign_peer_is_rejected(self) -> None:
|
|
# Production break caught: accepting a local AF_UNIX client owned by a
|
|
# UID other than the invoking unprivileged user.
|
|
read_fd, write_fd = os.pipe()
|
|
output_fd = os.open(os.devnull, os.O_WRONLY | os.O_CLOEXEC)
|
|
os.write(write_fd, self.app.encode() + b"\n" + bytes(self.master) + b"\n")
|
|
os.close(write_fd)
|
|
failures: list[str] = []
|
|
|
|
def target() -> None:
|
|
try:
|
|
with mock.patch.object(self.vault, "peer_uid", return_value=os.getuid() + 1):
|
|
self.vault.serve_private_socket(
|
|
os.fspath(self.root),
|
|
os.fspath(self.webhook_path),
|
|
input_fd=read_fd,
|
|
output_fd=output_fd,
|
|
accept_timeout=5.0,
|
|
io_timeout=1.0,
|
|
child_timeout=5.0,
|
|
)
|
|
except self.vault.VaultContractError:
|
|
pass
|
|
except BaseException as exc:
|
|
failures.append(type(exc).__name__)
|
|
finally:
|
|
os.close(read_fd)
|
|
os.close(output_fd)
|
|
|
|
thread = threading.Thread(target=target, daemon=True)
|
|
thread.start()
|
|
socket_path = self.root / "vault.sock"
|
|
deadline = time.monotonic() + 5
|
|
while not socket_path.exists() and time.monotonic() < deadline:
|
|
time.sleep(0.01)
|
|
client = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
client.settimeout(2)
|
|
try:
|
|
client.connect(os.fspath(socket_path))
|
|
client.sendall(b"PREPARE\n")
|
|
try:
|
|
self.assertEqual(client.recv(32), b"")
|
|
except ConnectionResetError:
|
|
pass
|
|
finally:
|
|
client.close()
|
|
thread.join(timeout=5)
|
|
self.assertFalse(thread.is_alive())
|
|
self.assertEqual(failures, [])
|
|
|
|
def test_server_fixed_objects_remain_below_retained_runtime_after_path_replacement(self) -> None:
|
|
# Production break caught: resolving fixed private names through the
|
|
# startup pathname after its directory has been replaced by the same UID.
|
|
runtime = self.root / "server-runtime"
|
|
runtime.mkdir(mode=0o700)
|
|
baseline = runtime / "baseline.kdbx"
|
|
create_database(baseline, self.master)
|
|
webhook_path = runtime / "input-webhook"
|
|
write_private(webhook_path, synthetic_webhook())
|
|
read_fd, write_fd = os.pipe()
|
|
output_fd = os.open(os.devnull, os.O_WRONLY | os.O_CLOEXEC)
|
|
os.write(write_fd, self.app.encode() + b"\n" + bytes(self.master) + b"\n")
|
|
os.close(write_fd)
|
|
failures: list[str] = []
|
|
|
|
def target() -> None:
|
|
try:
|
|
self.vault.serve_private_socket(
|
|
os.fspath(runtime),
|
|
os.fspath(webhook_path),
|
|
input_fd=read_fd,
|
|
output_fd=output_fd,
|
|
accept_timeout=5.0,
|
|
io_timeout=5.0,
|
|
child_timeout=20.0,
|
|
)
|
|
except BaseException as exc:
|
|
failures.append(type(exc).__name__)
|
|
finally:
|
|
os.close(read_fd)
|
|
os.close(output_fd)
|
|
|
|
thread = threading.Thread(target=target, daemon=True)
|
|
thread.start()
|
|
socket_path = runtime / "vault.sock"
|
|
deadline = time.monotonic() + 5
|
|
while not socket_path.exists() and time.monotonic() < deadline:
|
|
time.sleep(0.01)
|
|
self.assertTrue(socket_path.exists())
|
|
|
|
anchored = self.root / "server-runtime-anchored"
|
|
runtime.rename(anchored)
|
|
runtime.mkdir(mode=0o700)
|
|
foreign_candidate = runtime / "candidate.kdbx"
|
|
foreign_socket = runtime / "vault.sock"
|
|
write_private(foreign_candidate, b"foreign-candidate-must-survive")
|
|
write_private(foreign_socket, b"foreign-socket-name-must-survive")
|
|
|
|
responses = bytearray()
|
|
client = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
client.settimeout(10)
|
|
client.connect(os.fspath(anchored / "vault.sock"))
|
|
try:
|
|
for command, expected in (
|
|
(b"PREPARE\n", b"absent\n"),
|
|
(b"BUILD ADD\n", b"candidate-ready\n"),
|
|
(b"VERIFY_CANDIDATE\n", b"verified\n"),
|
|
(
|
|
b"VERIFY_COMMITTED "
|
|
+ os.fsencode(runtime / "candidate.kdbx")
|
|
+ b"\n",
|
|
b"verified\n",
|
|
),
|
|
(b"SHUTDOWN\n", b"stopped\n"),
|
|
):
|
|
client.sendall(command)
|
|
reply = b""
|
|
while not reply.endswith(b"\n"):
|
|
reply += client.recv(128)
|
|
responses.extend(reply)
|
|
self.assertEqual(reply, expected)
|
|
finally:
|
|
client.close()
|
|
thread.join(timeout=10)
|
|
self.assertFalse(thread.is_alive())
|
|
self.assertEqual(failures, [])
|
|
self.assertTrue((anchored / "candidate.kdbx").is_file())
|
|
self.assertFalse((anchored / "vault.sock").exists())
|
|
self.assertEqual(foreign_candidate.read_bytes(), b"foreign-candidate-must-survive")
|
|
self.assertEqual(foreign_socket.read_bytes(), b"foreign-socket-name-must-survive")
|
|
self.assertNotIn(bytes(self.master), responses)
|
|
shutil.rmtree(anchored)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|