#!/usr/bin/env python3

from __future__ import annotations

import argparse
import ctypes
import hashlib
import json
import os
import pathlib
import signal
import stat
import tempfile
import uuid
from contextlib import ExitStack
from typing import Any, Callable, Sequence


SCHEMA = "platform-slack-keepass-fs-v1"
VAULT_PARTS = ("HyeonworksRecovery", "vault")
BACKUPS_PARTS = VAULT_PARTS + ("backups",)
MAIN_NAME = "hyeonworks-recovery.kdbx"
MAIN_RELATIVE = VAULT_PARTS + (MAIN_NAME,)

OPEN_DIR_FLAGS = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC
OPEN_FILE_FLAGS = os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC
CREATE_FILE_FLAGS = os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW | os.O_CLOEXEC
COPY_CHUNK_SIZE = 1024 * 1024
MAX_BASELINE_JSON_BYTES = 64 * 1024
BLOCKED_SIGNALS = {signal.SIGHUP, signal.SIGINT, signal.SIGTERM}
MOUNT_UID = 1000
MOUNT_GID = 1000


class FsContractError(RuntimeError):
    pass


def _require_absolute(path: str) -> str:
    if not os.path.isabs(path):
        raise FsContractError(f"absolute path required: {path!r}")
    return path


def _validate_component(part: str) -> None:
    if not part or part in {".", ".."} or "/" in part:
        raise FsContractError(f"unsafe path component: {part!r}")


def _close_fd(fd: int) -> None:
    try:
        os.close(fd)
    except OSError:
        pass


def _mode_string(mode: int) -> str:
    return f"{stat.S_IMODE(mode):04o}"


def open_physical_absolute(path: str, flags: int, mode: int | None = None) -> int:
    absolute = _require_absolute(path)
    if mode is None:
        return os.open(absolute, flags)
    return os.open(absolute, flags, mode)


def attest_directory_fd(
    fd: int,
    *,
    expected_uid: int | None = None,
    expected_gid: int | None = None,
) -> os.stat_result:
    if expected_uid is None:
        expected_uid = os.getuid()
    if expected_gid is None:
        expected_gid = os.getgid()
    st = os.fstat(fd)
    if not stat.S_ISDIR(st.st_mode):
        raise FsContractError("expected directory")
    if st.st_uid != expected_uid:
        raise FsContractError("directory owner mismatch")
    if st.st_gid != expected_gid:
        raise FsContractError("directory group mismatch")
    if stat.S_IMODE(st.st_mode) != 0o700:
        raise FsContractError("directory mode mismatch")
    return st


def _attest_regular_fd(
    fd: int,
    *,
    expected_uid: int | None = None,
    expected_gid: int | None = None,
) -> os.stat_result:
    if expected_uid is None:
        expected_uid = os.getuid()
    if expected_gid is None:
        expected_gid = os.getgid()
    st = os.fstat(fd)
    if not stat.S_ISREG(st.st_mode):
        raise FsContractError("expected regular file")
    if st.st_uid != expected_uid:
        raise FsContractError("file owner mismatch")
    if st.st_gid != expected_gid:
        raise FsContractError("file group mismatch")
    if stat.S_IMODE(st.st_mode) != 0o600:
        raise FsContractError("file mode mismatch")
    if st.st_nlink != 1:
        raise FsContractError("file link count mismatch")
    return st


def open_directory_chain(root_fd: int, parts: Sequence[str]) -> int:
    current = os.dup(root_fd)
    try:
        for part in parts:
            _validate_component(part)
            next_fd = os.open(part, OPEN_DIR_FLAGS, dir_fd=current)
            os.close(current)
            current = next_fd
        return current
    except BaseException:
        os.close(current)
        raise


def _attest_retained_directory_entry(parent_fd: int, name: str, child_fd: int) -> None:
    _validate_component(name)
    retained = attest_directory_fd(
        child_fd,
        expected_uid=MOUNT_UID,
        expected_gid=MOUNT_GID,
    )
    current = os.stat(name, dir_fd=parent_fd, follow_symlinks=False)
    if not stat.S_ISDIR(current.st_mode):
        raise FsContractError("mount directory entry type mismatch")
    if (current.st_dev, current.st_ino) != (retained.st_dev, retained.st_ino):
        raise FsContractError("mount directory entry identity mismatch")
    if current.st_uid != MOUNT_UID or current.st_gid != MOUNT_GID:
        raise FsContractError("mount directory entry owner mismatch")
    if stat.S_IMODE(current.st_mode) != 0o700:
        raise FsContractError("mount directory entry mode mismatch")


def _attest_mount_bindings(
    mount_fd: int,
    recovery_fd: int,
    vault_fd: int,
    backups_fd: int | None = None,
) -> None:
    attest_directory_fd(
        mount_fd,
        expected_uid=MOUNT_UID,
        expected_gid=MOUNT_GID,
    )
    _attest_retained_directory_entry(mount_fd, VAULT_PARTS[0], recovery_fd)
    _attest_retained_directory_entry(recovery_fd, VAULT_PARTS[1], vault_fd)
    if backups_fd is not None:
        _attest_retained_directory_entry(vault_fd, "backups", backups_fd)


def _open_mount_tree(
    stack: ExitStack,
    mount_root: str,
    *,
    include_backups: bool,
) -> tuple[int, int, int, int | None]:
    mount_fd = stack.enter_context(_fd_context(_open_absolute_directory(mount_root)))
    attest_directory_fd(
        mount_fd,
        expected_uid=MOUNT_UID,
        expected_gid=MOUNT_GID,
    )
    recovery_fd = stack.enter_context(
        _fd_context(open_directory_chain(mount_fd, (VAULT_PARTS[0],)))
    )
    attest_directory_fd(
        recovery_fd,
        expected_uid=MOUNT_UID,
        expected_gid=MOUNT_GID,
    )
    vault_fd = stack.enter_context(
        _fd_context(open_directory_chain(recovery_fd, (VAULT_PARTS[1],)))
    )
    attest_directory_fd(
        vault_fd,
        expected_uid=MOUNT_UID,
        expected_gid=MOUNT_GID,
    )
    backups_fd = None
    if include_backups:
        backups_fd = stack.enter_context(
            _fd_context(open_directory_chain(vault_fd, ("backups",)))
        )
        attest_directory_fd(
            backups_fd,
            expected_uid=MOUNT_UID,
            expected_gid=MOUNT_GID,
        )
    _attest_mount_bindings(mount_fd, recovery_fd, vault_fd, backups_fd)
    return mount_fd, recovery_fd, vault_fd, backups_fd


def _require_same_identity(left_fd: int, right_fd: int, description: str) -> None:
    left = os.fstat(left_fd)
    right = os.fstat(right_fd)
    if (left.st_dev, left.st_ino) != (right.st_dev, right.st_ino):
        raise FsContractError(f"{description} identity drift")


def _attest_canonical_mount_tree(
    mount_root: str,
    mount_fd: int,
    recovery_fd: int,
    vault_fd: int,
    backups_fd: int | None = None,
    main_fd: int | None = None,
) -> None:
    with ExitStack() as canonical_stack:
        canonical_mount_fd, canonical_recovery_fd, canonical_vault_fd, canonical_backups_fd = (
            _open_mount_tree(
                canonical_stack,
                mount_root,
                include_backups=backups_fd is not None,
            )
        )
        _require_same_identity(mount_fd, canonical_mount_fd, "mount root")
        _require_same_identity(recovery_fd, canonical_recovery_fd, "recovery directory")
        _require_same_identity(vault_fd, canonical_vault_fd, "vault directory")
        if backups_fd is not None:
            if canonical_backups_fd is None:
                raise FsContractError("canonical backups directory unavailable")
            _require_same_identity(backups_fd, canonical_backups_fd, "backups directory")
        if main_fd is not None:
            canonical_main_fd = canonical_stack.enter_context(
                _fd_context(
                    open_regular_at(
                        canonical_vault_fd,
                        MAIN_NAME,
                        expected_uid=MOUNT_UID,
                        expected_gid=MOUNT_GID,
                    )
                )
            )
            _attest_retained_regular_entry(
                canonical_vault_fd,
                MAIN_NAME,
                canonical_main_fd,
                expected_uid=MOUNT_UID,
                expected_gid=MOUNT_GID,
            )
            _require_same_identity(main_fd, canonical_main_fd, "canonical main")
            if not files_equal_fd(main_fd, canonical_main_fd):
                raise FsContractError("canonical main content drift")
        _attest_mount_bindings(
            canonical_mount_fd,
            canonical_recovery_fd,
            canonical_vault_fd,
            canonical_backups_fd,
        )
        _require_same_identity(mount_fd, canonical_mount_fd, "mount root")
        _require_same_identity(recovery_fd, canonical_recovery_fd, "recovery directory")
        _require_same_identity(vault_fd, canonical_vault_fd, "vault directory")
        if backups_fd is not None and canonical_backups_fd is not None:
            _require_same_identity(backups_fd, canonical_backups_fd, "backups directory")


def _absolute_parts(path: str) -> tuple[str, ...]:
    pure = pathlib.PurePosixPath(_require_absolute(path))
    return tuple(pure.parts[1:])


def _open_absolute_directory(path: str) -> int:
    root_fd = open_physical_absolute("/", OPEN_DIR_FLAGS)
    try:
        return open_directory_chain(root_fd, _absolute_parts(path))
    finally:
        _close_fd(root_fd)


def _open_absolute_parent(path: str) -> tuple[int, str]:
    pure = pathlib.PurePosixPath(_require_absolute(path))
    if pure.name in {"", "/", ".", ".."}:
        raise FsContractError("path must name a file")
    root_fd = open_physical_absolute("/", OPEN_DIR_FLAGS)
    try:
        parent_fd = open_directory_chain(root_fd, tuple(pure.parts[1:-1]))
    finally:
        _close_fd(root_fd)
    return parent_fd, pure.name


def open_regular_at(
    dir_fd: int,
    name: str,
    *,
    expected_uid: int | None = None,
    expected_gid: int | None = None,
) -> int:
    _validate_component(name)
    fd = os.open(name, OPEN_FILE_FLAGS, dir_fd=dir_fd)
    try:
        _attest_regular_fd(
            fd,
            expected_uid=expected_uid,
            expected_gid=expected_gid,
        )
        return fd
    except BaseException:
        _close_fd(fd)
        raise


def _open_absolute_regular(path: str) -> int:
    parent_fd, name = _open_absolute_parent(path)
    try:
        attest_directory_fd(parent_fd)
        return open_regular_at(parent_fd, name)
    finally:
        _close_fd(parent_fd)


def _sha256_fd(fd: int) -> str:
    os.lseek(fd, 0, os.SEEK_SET)
    digest = hashlib.sha256()
    while True:
        chunk = os.read(fd, COPY_CHUNK_SIZE)
        if not chunk:
            break
        digest.update(chunk)
    return digest.hexdigest()


def stable_copy_fd(
    src_fd: int,
    dst_fd: int,
    *,
    expected_uid: int | None = None,
    expected_gid: int | None = None,
) -> dict[str, Any]:
    os.lseek(src_fd, 0, os.SEEK_SET)
    os.lseek(dst_fd, 0, os.SEEK_SET)
    digest = hashlib.sha256()
    size = 0
    while True:
        chunk = os.read(src_fd, COPY_CHUNK_SIZE)
        if not chunk:
            break
        digest.update(chunk)
        size += len(chunk)
        view = memoryview(chunk)
        while view:
            written = os.write(dst_fd, view)
            view = view[written:]
    os.fsync(dst_fd)
    dst_stat = _attest_regular_fd(
        dst_fd,
        expected_uid=expected_uid,
        expected_gid=expected_gid,
    )
    if dst_stat.st_size != size:
        raise FsContractError("copied size mismatch")
    return {"size": size, "sha256": digest.hexdigest()}


def files_equal_fd(left_fd: int, right_fd: int) -> bool:
    left_stat = os.fstat(left_fd)
    right_stat = os.fstat(right_fd)
    if left_stat.st_size != right_stat.st_size:
        return False
    os.lseek(left_fd, 0, os.SEEK_SET)
    os.lseek(right_fd, 0, os.SEEK_SET)
    while True:
        left_chunk = os.read(left_fd, COPY_CHUNK_SIZE)
        right_chunk = os.read(right_fd, COPY_CHUNK_SIZE)
        if left_chunk != right_chunk:
            return False
        if not left_chunk:
            return True


def sync_filesystem_fd(fd: int) -> None:
    libc = ctypes.CDLL(None, use_errno=True)
    syncfs = libc.syncfs
    syncfs.argtypes = [ctypes.c_int]
    syncfs.restype = ctypes.c_int
    rc = syncfs(fd)
    if rc != 0:
        err = ctypes.get_errno()
        raise OSError(err, os.strerror(err))


def _build_state(operation: str, mount_root: str, source_stat: os.stat_result, source_sha256: str) -> dict[str, Any]:
    return {
        "schema": SCHEMA,
        "operation": operation,
        "mount": {
            "realpath": os.path.realpath(mount_root),
            "relative_vault": "/".join(VAULT_PARTS),
        },
        "source": {
            "relative_path": "/".join(MAIN_RELATIVE),
            "size": source_stat.st_size,
            "mode": _mode_string(source_stat.st_mode),
            "uid": source_stat.st_uid,
            "gid": source_stat.st_gid,
            "inode": source_stat.st_ino,
            "sha256": source_sha256,
        },
    }


def _read_json_fd(fd: int, *, max_bytes: int) -> dict[str, Any]:
    st = os.fstat(fd)
    if not stat.S_ISREG(st.st_mode):
        raise FsContractError("expected regular file")
    if st.st_size > max_bytes:
        raise FsContractError("baseline json too large")
    os.lseek(fd, 0, os.SEEK_SET)
    chunks: list[bytes] = []
    total = 0
    while True:
        chunk = os.read(fd, min(COPY_CHUNK_SIZE, max_bytes - total + 1))
        if not chunk:
            break
        total += len(chunk)
        if total > max_bytes:
            raise FsContractError("baseline json too large")
        chunks.append(chunk)
    try:
        parsed = json.loads(b"".join(chunks).decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise FsContractError("baseline json invalid") from exc
    if not isinstance(parsed, dict):
        raise FsContractError("baseline json payload mismatch")
    return parsed


def _read_result(path: str) -> dict[str, Any]:
    parent_fd, name = _open_absolute_parent(path)
    try:
        attest_directory_fd(parent_fd)
        baseline_fd = open_regular_at(parent_fd, name)
        try:
            return _read_json_fd(baseline_fd, max_bytes=MAX_BASELINE_JSON_BYTES)
        finally:
            _close_fd(baseline_fd)
    finally:
        _close_fd(parent_fd)


def _validate_snapshot_state(state: dict[str, Any], mount_root: str) -> None:
    if set(state) != {"schema", "operation", "mount", "source"}:
        raise FsContractError("unexpected baseline keys")
    if state["schema"] != SCHEMA:
        raise FsContractError("baseline schema mismatch")
    if state["operation"] != "snapshot":
        raise FsContractError("baseline operation mismatch")
    mount = state["mount"]
    source = state["source"]
    if not isinstance(mount, dict) or set(mount) != {"realpath", "relative_vault"}:
        raise FsContractError("baseline mount payload mismatch")
    if not isinstance(source, dict) or set(source) != {
        "relative_path",
        "size",
        "mode",
        "uid",
        "gid",
        "inode",
        "sha256",
    }:
        raise FsContractError("baseline source payload mismatch")
    if mount["realpath"] != os.path.realpath(mount_root):
        raise FsContractError("baseline mount root mismatch")
    if mount["relative_vault"] != "/".join(VAULT_PARTS):
        raise FsContractError("baseline vault mismatch")
    if source["relative_path"] != "/".join(MAIN_RELATIVE):
        raise FsContractError("baseline source path mismatch")
    if source["mode"] != "0600":
        raise FsContractError("baseline source mode mismatch")
    if source["uid"] != MOUNT_UID:
        raise FsContractError("baseline source owner mismatch")
    if source["gid"] != MOUNT_GID:
        raise FsContractError("baseline source group mismatch")
    if not isinstance(source["size"], int) or source["size"] < 0:
        raise FsContractError("baseline size mismatch")
    if not isinstance(source["inode"], int) or source["inode"] <= 0:
        raise FsContractError("baseline inode mismatch")
    if not isinstance(source["sha256"], str) or len(source["sha256"]) != 64:
        raise FsContractError("baseline hash mismatch")


def _assert_matches_baseline(main_fd: int, baseline_state: dict[str, Any]) -> os.stat_result:
    source = baseline_state["source"]
    main_stat = _attest_regular_fd(
        main_fd,
        expected_uid=MOUNT_UID,
        expected_gid=MOUNT_GID,
    )
    if main_stat.st_size != source["size"]:
        raise FsContractError("baseline size drift")
    if _mode_string(main_stat.st_mode) != source["mode"]:
        raise FsContractError("baseline mode drift")
    if main_stat.st_uid != source["uid"]:
        raise FsContractError("baseline owner drift")
    if main_stat.st_gid != source["gid"]:
        raise FsContractError("baseline group drift")
    if main_stat.st_ino != source["inode"]:
        raise FsContractError("baseline inode drift")
    if _sha256_fd(main_fd) != source["sha256"]:
        raise FsContractError("baseline content drift")
    return main_stat


def _create_private_file(
    dir_fd: int,
    name: str,
    *,
    expected_uid: int | None = None,
    expected_gid: int | None = None,
) -> int:
    _validate_component(name)
    fd = os.open(name, CREATE_FILE_FLAGS, 0o600, dir_fd=dir_fd)
    try:
        _attest_regular_fd(
            fd,
            expected_uid=expected_uid,
            expected_gid=expected_gid,
        )
        return fd
    except BaseException:
        _close_fd(fd)
        raise


def _attest_retained_regular_entry(
    dir_fd: int,
    name: str,
    file_fd: int,
    *,
    expected_uid: int,
    expected_gid: int,
) -> os.stat_result:
    _validate_component(name)
    retained = _attest_regular_fd(
        file_fd,
        expected_uid=expected_uid,
        expected_gid=expected_gid,
    )
    current = os.stat(name, dir_fd=dir_fd, follow_symlinks=False)
    if not stat.S_ISREG(current.st_mode):
        raise FsContractError("file entry type mismatch")
    if (current.st_dev, current.st_ino) != (retained.st_dev, retained.st_ino):
        raise FsContractError("file entry identity mismatch")
    if current.st_uid != expected_uid or current.st_gid != expected_gid:
        raise FsContractError("file entry owner mismatch")
    if stat.S_IMODE(current.st_mode) != 0o600:
        raise FsContractError("file entry mode mismatch")
    if current.st_nlink != 1:
        raise FsContractError("file entry link count mismatch")
    return retained


def _unlink_if_same_file(dir_fd: int, name: str, expected_stat: os.stat_result) -> None:
    try:
        current_fd = os.open(name, OPEN_FILE_FLAGS, dir_fd=dir_fd)
    except FileNotFoundError:
        return
    try:
        current_stat = os.fstat(current_fd)
        if (
            current_stat.st_dev == expected_stat.st_dev
            and current_stat.st_ino == expected_stat.st_ino
        ):
            os.unlink(name, dir_fd=dir_fd)
    finally:
        _close_fd(current_fd)


def _snapshot_from_main(operation: str, mount_root: str, destination: str, candidate: str | None = None) -> dict[str, Any]:
    with ExitStack() as stack:
        mount_fd, recovery_fd, vault_fd, _ = _open_mount_tree(
            stack,
            mount_root,
            include_backups=False,
        )
        main_fd = stack.enter_context(
            _fd_context(
                open_regular_at(
                    vault_fd,
                    MAIN_NAME,
                    expected_uid=MOUNT_UID,
                    expected_gid=MOUNT_GID,
                )
            )
        )
        _attest_retained_regular_entry(
            vault_fd,
            MAIN_NAME,
            main_fd,
            expected_uid=MOUNT_UID,
            expected_gid=MOUNT_GID,
        )
        if candidate is not None:
            candidate_fd = stack.enter_context(_fd_context(_open_absolute_regular(candidate)))
            if not files_equal_fd(main_fd, candidate_fd):
                raise FsContractError("candidate does not match installed database")
        dest_parent_fd, dest_name = _open_absolute_parent(destination)
        stack.callback(_close_fd, dest_parent_fd)
        attest_directory_fd(dest_parent_fd)
        dest_fd = stack.enter_context(_fd_context(_create_private_file(dest_parent_fd, dest_name)))
        dest_stat = os.fstat(dest_fd)
        try:
            _attest_mount_bindings(mount_fd, recovery_fd, vault_fd)
            _attest_canonical_mount_tree(
                mount_root,
                mount_fd,
                recovery_fd,
                vault_fd,
                main_fd=main_fd,
            )
            copied = stable_copy_fd(
                main_fd,
                dest_fd,
                expected_uid=os.getuid(),
                expected_gid=os.getgid(),
            )
            _attest_retained_regular_entry(
                vault_fd,
                MAIN_NAME,
                main_fd,
                expected_uid=MOUNT_UID,
                expected_gid=MOUNT_GID,
            )
            _attest_canonical_mount_tree(
                mount_root,
                mount_fd,
                recovery_fd,
                vault_fd,
                main_fd=main_fd,
            )
            state = _build_state(operation, mount_root, os.fstat(main_fd), copied["sha256"])
            _attest_canonical_mount_tree(
                mount_root,
                mount_fd,
                recovery_fd,
                vault_fd,
                main_fd=main_fd,
            )
            return state
        except BaseException:
            _unlink_if_same_file(dest_parent_fd, dest_name, dest_stat)
            raise


def snapshot_database(mount_root: str, destination: str) -> dict[str, Any]:
    return _snapshot_from_main("snapshot", mount_root, destination)


def snapshot_committed_database(mount_root: str, candidate: str, destination: str) -> dict[str, Any]:
    return _snapshot_from_main("snapshot-current", mount_root, destination, candidate=candidate)


def commit_database(
    mount_root: str,
    baseline: str,
    candidate: str,
    backup_name: str,
    *,
    status_hook: Callable[[str], None] | None = None,
) -> dict[str, Any]:
    _validate_component(backup_name)
    baseline_state = _read_result(baseline)
    _validate_snapshot_state(baseline_state, mount_root)
    result = {
        "schema": SCHEMA,
        "operation": "commit",
        "mount": {
            "realpath": os.path.realpath(mount_root),
            "relative_vault": "/".join(VAULT_PARTS),
        },
        "backup_name": backup_name,
    }
    with ExitStack() as stack:
        mount_fd, recovery_fd, vault_fd, backups_fd = _open_mount_tree(
            stack,
            mount_root,
            include_backups=True,
        )
        if backups_fd is None:
            raise FsContractError("backups directory unavailable")
        original_fd = stack.enter_context(
            _fd_context(
                open_regular_at(
                    vault_fd,
                    MAIN_NAME,
                    expected_uid=MOUNT_UID,
                    expected_gid=MOUNT_GID,
                )
            )
        )
        _assert_matches_baseline(original_fd, baseline_state)
        candidate_fd = stack.enter_context(_fd_context(_open_absolute_regular(candidate)))

        _attest_mount_bindings(mount_fd, recovery_fd, vault_fd, backups_fd)
        backup_fd = stack.enter_context(
            _fd_context(
                _create_private_file(
                    backups_fd,
                    backup_name,
                    expected_uid=MOUNT_UID,
                    expected_gid=MOUNT_GID,
                )
            )
        )
        stable_copy_fd(
            original_fd,
            backup_fd,
            expected_uid=MOUNT_UID,
            expected_gid=MOUNT_GID,
        )
        backup_stat = _attest_retained_regular_entry(
            backups_fd,
            backup_name,
            backup_fd,
            expected_uid=MOUNT_UID,
            expected_gid=MOUNT_GID,
        )
        os.fsync(backups_fd)
        sync_filesystem_fd(backups_fd)

        reopened_backup_fd = stack.enter_context(
            _fd_context(
                open_regular_at(
                    backups_fd,
                    backup_name,
                    expected_uid=MOUNT_UID,
                    expected_gid=MOUNT_GID,
                )
            )
        )
        reopened_stat = _attest_retained_regular_entry(
            backups_fd,
            backup_name,
            reopened_backup_fd,
            expected_uid=MOUNT_UID,
            expected_gid=MOUNT_GID,
        )
        if (reopened_stat.st_dev, reopened_stat.st_ino) != (
            backup_stat.st_dev,
            backup_stat.st_ino,
        ):
            raise FsContractError("backup identity proof failed")
        if not files_equal_fd(original_fd, reopened_backup_fd):
            raise FsContractError("backup proof failed")

        stage_name = f".{MAIN_NAME}.stage.{os.getpid()}.{uuid.uuid4().hex}"
        _attest_mount_bindings(mount_fd, recovery_fd, vault_fd, backups_fd)
        _attest_retained_regular_entry(
            backups_fd,
            backup_name,
            backup_fd,
            expected_uid=MOUNT_UID,
            expected_gid=MOUNT_GID,
        )
        stage_fd = stack.enter_context(
            _fd_context(
                _create_private_file(
                    vault_fd,
                    stage_name,
                    expected_uid=MOUNT_UID,
                    expected_gid=MOUNT_GID,
                )
            )
        )
        stage_stat = os.fstat(stage_fd)
        stage_replaced = False
        try:
            stable_copy_fd(
                candidate_fd,
                stage_fd,
                expected_uid=MOUNT_UID,
                expected_gid=MOUNT_GID,
            )
            os.fsync(vault_fd)
            sync_filesystem_fd(vault_fd)

            current_main_fd = stack.enter_context(
                _fd_context(
                    open_regular_at(
                        vault_fd,
                        MAIN_NAME,
                        expected_uid=MOUNT_UID,
                        expected_gid=MOUNT_GID,
                    )
                )
            )
            _assert_matches_baseline(current_main_fd, baseline_state)
            _attest_retained_regular_entry(
                vault_fd,
                MAIN_NAME,
                current_main_fd,
                expected_uid=MOUNT_UID,
                expected_gid=MOUNT_GID,
            )
            _attest_mount_bindings(mount_fd, recovery_fd, vault_fd, backups_fd)
            _attest_retained_regular_entry(
                backups_fd,
                backup_name,
                backup_fd,
                expected_uid=MOUNT_UID,
                expected_gid=MOUNT_GID,
            )
            _attest_retained_regular_entry(
                vault_fd,
                stage_name,
                stage_fd,
                expected_uid=MOUNT_UID,
                expected_gid=MOUNT_GID,
            )
            _attest_canonical_mount_tree(
                mount_root,
                mount_fd,
                recovery_fd,
                vault_fd,
                backups_fd,
                current_main_fd,
            )

            if status_hook is not None:
                status_hook("rename-attempted")
            previous_mask = signal.pthread_sigmask(signal.SIG_BLOCK, BLOCKED_SIGNALS)
            try:
                os.replace(stage_name, MAIN_NAME, src_dir_fd=vault_fd, dst_dir_fd=vault_fd)
                stage_replaced = True
                installed_fd = stack.enter_context(
                    _fd_context(
                        open_regular_at(
                            vault_fd,
                            MAIN_NAME,
                            expected_uid=MOUNT_UID,
                            expected_gid=MOUNT_GID,
                        )
                    )
                )
                _attest_retained_regular_entry(
                    vault_fd,
                    MAIN_NAME,
                    installed_fd,
                    expected_uid=MOUNT_UID,
                    expected_gid=MOUNT_GID,
                )
                if not files_equal_fd(installed_fd, candidate_fd):
                    raise FsContractError("installed file mismatch")
                os.fsync(installed_fd)
                os.fsync(vault_fd)
                sync_filesystem_fd(vault_fd)
                _attest_canonical_mount_tree(
                    mount_root,
                    mount_fd,
                    recovery_fd,
                    vault_fd,
                    backups_fd,
                    installed_fd,
                )
                _attest_canonical_mount_tree(
                    mount_root,
                    mount_fd,
                    recovery_fd,
                    vault_fd,
                    backups_fd,
                    installed_fd,
                )
                result["fs_commit"] = "verified-commit"
                return result
            except BaseException:
                result["fs_commit"] = "committed-but-uncertain"
                return result
            finally:
                signal.pthread_sigmask(signal.SIG_SETMASK, previous_mask)
        except BaseException:
            if not stage_replaced:
                _unlink_if_same_file(vault_fd, stage_name, stage_stat)
            raise

        return result


def write_result(path: str, state: dict[str, Any]) -> None:
    parent_fd, name = _open_absolute_parent(path)
    payload = json.dumps(state, sort_keys=True, separators=(",", ":")).encode("utf-8") + b"\n"
    temp_name = f".{name}.tmp.{os.getpid()}.{uuid.uuid4().hex}"
    temp_stat = None
    replaced = False
    try:
        attest_directory_fd(parent_fd)
        temp_fd = _create_private_file(parent_fd, temp_name)
        try:
            temp_stat = os.fstat(temp_fd)
            view = memoryview(payload)
            while view:
                written = os.write(temp_fd, view)
                view = view[written:]
            os.fsync(temp_fd)
            _attest_retained_regular_entry(
                parent_fd,
                temp_name,
                temp_fd,
                expected_uid=os.getuid(),
                expected_gid=os.getgid(),
            )
        finally:
            _close_fd(temp_fd)
        os.replace(temp_name, name, src_dir_fd=parent_fd, dst_dir_fd=parent_fd)
        replaced = True
        os.fsync(parent_fd)
    finally:
        if not replaced and temp_stat is not None:
            _unlink_if_same_file(parent_fd, temp_name, temp_stat)
        _close_fd(parent_fd)


class _fd_context:
    def __init__(self, fd: int):
        self.fd = fd

    def __enter__(self) -> int:
        return self.fd

    def __exit__(self, exc_type, exc, tb) -> None:
        _close_fd(self.fd)


def _cmd_snapshot(args: argparse.Namespace) -> dict[str, Any]:
    state = snapshot_database(args.mount_root, args.destination)
    write_result(args.result, state)
    return state


def _cmd_commit(args: argparse.Namespace) -> dict[str, Any]:
    def persist_status(fs_commit: str) -> None:
        if fs_commit not in {"precommit-failure", "rename-attempted"}:
            raise FsContractError("invalid commit prearm status")
        write_result(
            args.result,
            {
                "schema": SCHEMA,
                "operation": "commit-status",
                "fs_commit": fs_commit,
            },
        )

    persist_status("precommit-failure")
    state = commit_database(
        args.mount_root,
        args.baseline,
        args.candidate,
        args.backup_name,
        status_hook=persist_status,
    )
    write_result(args.result, state)
    return state


def _cmd_snapshot_current(args: argparse.Namespace) -> dict[str, Any]:
    state = snapshot_committed_database(args.mount_root, args.candidate, args.destination)
    write_result(args.result, state)
    return state


def read_commit_status(path: str) -> str:
    state = _read_result(path)
    if state.get("schema") != SCHEMA:
        raise FsContractError("commit status schema mismatch")
    operation = state.get("operation")
    fs_commit = state.get("fs_commit")
    if operation == "commit-status":
        if set(state) != {"schema", "operation", "fs_commit"} or fs_commit not in {
            "precommit-failure",
            "rename-attempted",
        }:
            raise FsContractError("commit prearm payload mismatch")
    elif operation == "commit":
        if set(state) != {"schema", "operation", "mount", "backup_name", "fs_commit"} or fs_commit not in {
            "verified-commit",
            "committed-but-uncertain",
        }:
            raise FsContractError("commit result payload mismatch")
    else:
        raise FsContractError("commit status operation mismatch")
    return fs_commit


def _cmd_status(args: argparse.Namespace) -> dict[str, Any]:
    status = read_commit_status(args.result)
    os.write(1, status.encode("ascii") + b"\n")
    return {"fs_commit": status}


def audit_runtime_processes(runtime_root: str) -> None:
    canonical = os.path.realpath(_require_absolute(runtime_root))
    runtime_info = os.stat(canonical, follow_symlinks=False)
    for entry in os.listdir("/proc"):
        if not entry.isdigit() or int(entry) == os.getpid():
            continue
        try:
            process_info = os.stat(f"/proc/{entry}", follow_symlinks=False)
            command_line = pathlib.Path(f"/proc/{entry}/cmdline").read_bytes()
        except OSError:
            continue
        if process_info.st_uid != os.getuid() or not any(
            marker in command_line
            for marker in (
                b"slack-webhook-recovery-vault.py",
                b"slack-webhook-recovery-dirfd.py",
                b"keepassxc-cli",
                b"socat",
            )
        ):
            continue
        retained_runtime = canonical.encode() in command_line
        try:
            fd_names = os.listdir(f"/proc/{entry}/fd")
        except OSError:
            fd_names = ()
        for fd_name in fd_names:
            try:
                fd_info = os.stat(f"/proc/{entry}/fd/{fd_name}")
                fd_target = os.readlink(f"/proc/{entry}/fd/{fd_name}")
            except OSError:
                continue
            if (fd_info.st_dev, fd_info.st_ino) == (
                runtime_info.st_dev, runtime_info.st_ino
            ) or fd_target == canonical or fd_target.startswith(canonical + "/"):
                retained_runtime = True
                break
        if retained_runtime:
            raise FsContractError("runtime-owned process remains")


def _cmd_audit_processes(args: argparse.Namespace) -> dict[str, Any]:
    audit_runtime_processes(args.runtime_root)
    return {"processes": "absent"}


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser()
    subparsers = parser.add_subparsers(dest="command", required=True)

    snapshot = subparsers.add_parser("snapshot")
    snapshot.add_argument("--mount-root", required=True)
    snapshot.add_argument("--destination", required=True)
    snapshot.add_argument("--result", required=True)
    snapshot.set_defaults(func=_cmd_snapshot)

    commit = subparsers.add_parser("commit")
    commit.add_argument("--mount-root", required=True)
    commit.add_argument("--baseline", required=True)
    commit.add_argument("--candidate", required=True)
    commit.add_argument("--backup-name", required=True)
    commit.add_argument("--result", required=True)
    commit.set_defaults(func=_cmd_commit)

    snapshot_current = subparsers.add_parser("snapshot-current")
    snapshot_current.add_argument("--mount-root", required=True)
    snapshot_current.add_argument("--candidate", required=True)
    snapshot_current.add_argument("--destination", required=True)
    snapshot_current.add_argument("--result", required=True)
    snapshot_current.set_defaults(func=_cmd_snapshot_current)

    status_parser = subparsers.add_parser("status")
    status_parser.add_argument("--result", required=True)
    status_parser.set_defaults(func=_cmd_status)

    audit_processes = subparsers.add_parser("audit-processes")
    audit_processes.add_argument("--runtime-root", required=True)
    audit_processes.set_defaults(func=_cmd_audit_processes)

    return parser


def main(argv: Sequence[str] | None = None) -> int:
    parser = _build_parser()
    args = parser.parse_args(argv)
    try:
        args.func(args)
        return 0
    except (OSError, FsContractError, ValueError, KeyError, TypeError):
        os.write(2, b"dirfd helper failed\n")
        return 1


if __name__ == "__main__":
    raise SystemExit(main())