#!/usr/bin/python3

from __future__ import annotations

import hmac
import ctypes
import os
import re
import signal
import socket
import stat
import struct
import subprocess
import sys
import termios
from typing import Callable, Sequence


KEEPASSXC_CLI = "/usr/bin/keepassxc-cli"
PYTHON3 = "/usr/bin/python3"
SOCKET_NAME = "vault.sock"
BASELINE_NAME = "baseline.kdbx"
CANDIDATE_NAME = "candidate.kdbx"
ENTRY_GROUP = "/Platform/Observability/Slack"
ENTRY_TITLE = "Alertmanager webhook - desktop-infra-전체"
ENTRY_PATH = ENTRY_GROUP + "/" + ENTRY_TITLE
ENTRY_USERNAME = "desktop-infra"
ENTRY_URL = ""
MAX_WEBHOOK_BYTES = 4096
MAX_INPUT_BYTES = 4096
MAX_PROTOCOL_BYTES = 4096
MAX_CHILD_OUTPUT_BYTES = 64 * 1024
MAX_CHILD_STDERR_BYTES = 8192
DEFAULT_ACCEPT_TIMEOUT = 900.0
DEFAULT_IO_TIMEOUT = 15.0
DEFAULT_CHILD_TIMEOUT = 30.0
WEBHOOK_RE = re.compile(
    rb"https://hooks[.]slack[.]com/services/"
    rb"[A-Za-z0-9_-]+/[A-Za-z0-9_-]+/[A-Za-z0-9_-]+"
)


class VaultContractError(RuntimeError):
    pass


class VaultProtocolError(VaultContractError):
    pass


class VaultSignal(BaseException):
    def __init__(self, signum: int) -> None:
        super().__init__(signum)
        self.signum = signum


HANDLED_SIGNALS = (signal.SIGHUP, signal.SIGINT, signal.SIGTERM)


def install_process_signal_handlers() -> None:
    def raise_first_signal(signum: int, _frame: object) -> None:
        for handled in HANDLED_SIGNALS:
            signal.signal(handled, signal.SIG_IGN)
        raise VaultSignal(signum)

    for signum in HANDLED_SIGNALS:
        signal.signal(signum, raise_first_signal)


class KeepassResult:
    def __init__(self, returncode: int, stdout: bytes, stderr: bytes) -> None:
        self.returncode = returncode
        self.stdout = stdout
        self.stderr = stderr


def wipe_mutable_buffer(buffer: bytearray) -> None:
    for index in range(len(buffer)):
        buffer[index] = 0


def _regular_metadata(fd: int, *, allow_unlinked: bool = False) -> tuple[int, ...]:
    info = os.fstat(fd)
    if not stat.S_ISREG(info.st_mode):
        raise VaultContractError("expected regular file")
    if info.st_uid != os.getuid():
        raise VaultContractError("file owner mismatch")
    if stat.S_IMODE(info.st_mode) != 0o600:
        raise VaultContractError("file mode mismatch")
    if info.st_nlink != 1 and not (allow_unlinked and info.st_nlink == 0):
        raise VaultContractError("file link count mismatch")
    return (
        info.st_dev,
        info.st_ino,
        info.st_uid,
        stat.S_IMODE(info.st_mode),
        info.st_nlink,
        info.st_size,
    )


class RetainedWebhook:
    def __init__(self, fd: int, metadata: tuple[int, ...], content: bytearray) -> None:
        self._fd = fd
        self._metadata = metadata
        self._content = bytearray(content)
        self._closed = False

    @property
    def fd(self) -> int:
        if self._closed:
            raise VaultContractError("webhook descriptor is closed")
        return self._fd

    def read_bytes(self) -> bytes:
        if self._closed:
            raise VaultContractError("webhook descriptor is closed")
        before = _regular_metadata(self._fd)
        if before != self._metadata:
            raise VaultContractError("webhook metadata drift")
        collected = bytearray()
        confirmation = bytearray()
        try:
            for target in (collected, confirmation):
                os.lseek(self._fd, 0, os.SEEK_SET)
                while len(target) <= MAX_WEBHOOK_BYTES:
                    chunk = os.read(
                        self._fd,
                        min(1024, MAX_WEBHOOK_BYTES + 1 - len(target)),
                    )
                    if not chunk:
                        break
                    target.extend(chunk)
            after = _regular_metadata(self._fd)
            if after != before:
                raise VaultContractError("webhook metadata drift")
            if len(collected) > MAX_WEBHOOK_BYTES or len(confirmation) > MAX_WEBHOOK_BYTES:
                raise VaultContractError("webhook content is oversized")
            if not hmac.compare_digest(bytes(collected), bytes(confirmation)) or not hmac.compare_digest(
                bytes(confirmation), self._content
            ):
                raise VaultContractError("webhook content drift")
            return bytes(collected)
        finally:
            wipe_mutable_buffer(collected)
            wipe_mutable_buffer(confirmation)

    def close(self) -> None:
        if not self._closed:
            try:
                os.close(self._fd)
            finally:
                wipe_mutable_buffer(self._content)
                self._closed = True


def _validate_webhook_content(content: bytes) -> None:
    if not content or len(content) > MAX_WEBHOOK_BYTES:
        raise VaultContractError("invalid webhook content")
    if b"\r" in content or b"\n" in content or b"\0" in content:
        raise VaultContractError("invalid webhook content")
    if content.strip() != content:
        raise VaultContractError("invalid webhook content")
    if WEBHOOK_RE.fullmatch(content) is None:
        raise VaultContractError("invalid webhook content")


def open_validated_webhook(path: str) -> RetainedWebhook:
    if not os.path.isabs(path):
        raise VaultContractError("absolute webhook path required")
    flags = os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC
    fd = os.open(path, flags)
    try:
        before = _regular_metadata(fd)
        content_buffer = bytearray()
        confirmation_buffer = bytearray()
        try:
            for target in (content_buffer, confirmation_buffer):
                if _regular_metadata(fd) != before:
                    raise VaultContractError("webhook metadata drift")
                os.lseek(fd, 0, os.SEEK_SET)
                while len(target) <= MAX_WEBHOOK_BYTES:
                    chunk = os.read(
                        fd,
                        min(1024, MAX_WEBHOOK_BYTES + 1 - len(target)),
                    )
                    if not chunk:
                        break
                    target.extend(chunk)
                if _regular_metadata(fd) != before:
                    raise VaultContractError("webhook metadata drift")
            after = _regular_metadata(fd)
            if before != after:
                raise VaultContractError("webhook metadata drift")
            if not hmac.compare_digest(
                bytes(content_buffer), bytes(confirmation_buffer)
            ):
                raise VaultContractError("webhook content drift")
            _validate_webhook_content(confirmation_buffer)
            return RetainedWebhook(fd, before, confirmation_buffer)
        finally:
            wipe_mutable_buffer(content_buffer)
            wipe_mutable_buffer(confirmation_buffer)
    except BaseException:
        os.close(fd)
        raise


def _read_line_fd(fd: int, limit: int) -> bytearray:
    value = bytearray()
    try:
        while len(value) <= limit:
            chunk = os.read(fd, 1)
            if not chunk:
                raise VaultContractError("operator input ended early")
            if chunk == b"\n":
                return value
            if chunk in {b"\r", b"\0"}:
                raise VaultContractError("invalid operator input")
            value.extend(chunk)
        raise VaultContractError("operator input is oversized")
    except BaseException:
        wipe_mutable_buffer(value)
        raise


def _validate_app_name(raw: bytes) -> str:
    try:
        app = raw.decode("utf-8", "strict")
    except UnicodeDecodeError as exc:
        raise VaultContractError("invalid app name") from exc
    if not 1 <= len(app) <= 80 or app.strip() != app:
        raise VaultContractError("invalid app name")
    if any(character in ";=\\\r\n" or not character.isprintable() for character in app):
        raise VaultContractError("invalid app name")
    return app


def read_operator_secrets(
    *, input_fd: int | None = None, output_fd: int | None = None
) -> tuple[str, bytearray]:
    owned_fd = False
    if input_fd is None:
        input_fd = os.open("/dev/tty", os.O_RDWR | os.O_CLOEXEC)
        output_fd = input_fd
        owned_fd = True
    if output_fd is None:
        raise VaultContractError("operator output descriptor required")
    master = bytearray()
    old_termios = None
    try:
        os.write(output_fd, b"Slack app name: ")
        raw_app = _read_line_fd(input_fd, MAX_INPUT_BYTES)
        try:
            app = _validate_app_name(bytes(raw_app))
        finally:
            wipe_mutable_buffer(raw_app)
        if os.isatty(input_fd):
            previous_mask = signal.pthread_sigmask(signal.SIG_BLOCK, HANDLED_SIGNALS)
            try:
                old_termios = termios.tcgetattr(input_fd)
                new_termios = list(old_termios)
                new_termios[3] &= ~termios.ECHO
                termios.tcsetattr(input_fd, termios.TCSAFLUSH, new_termios)
                os.write(output_fd, b"KeePassXC master password: ")
            finally:
                signal.pthread_sigmask(signal.SIG_SETMASK, previous_mask)
        else:
            os.write(output_fd, b"KeePassXC master password: ")
        master = _read_line_fd(input_fd, MAX_INPUT_BYTES)
        if not master:
            raise VaultContractError("empty master password")
        if old_termios is not None:
            previous_mask = signal.pthread_sigmask(signal.SIG_BLOCK, HANDLED_SIGNALS)
            try:
                termios.tcsetattr(input_fd, termios.TCSAFLUSH, old_termios)
                old_termios = None
                os.write(output_fd, b"\n")
            finally:
                signal.pthread_sigmask(signal.SIG_SETMASK, previous_mask)
        return app, master
    except BaseException:
        wipe_mutable_buffer(master)
        raise
    finally:
        if old_termios is not None:
            previous_mask = signal.pthread_sigmask(signal.SIG_BLOCK, HANDLED_SIGNALS)
            try:
                termios.tcsetattr(input_fd, termios.TCSAFLUSH, old_termios)
                os.write(output_fd, b"\n")
            finally:
                signal.pthread_sigmask(signal.SIG_SETMASK, previous_mask)
        if owned_fd:
            os.close(input_fd)


def _validate_child_stderr(stderr: bytes) -> None:
    if len(stderr) > MAX_CHILD_STDERR_BYTES or b"\0" in stderr or b"\r" in stderr:
        raise VaultContractError("invalid KeePass diagnostic")
    try:
        text = stderr.decode("utf-8", "strict")
    except UnicodeDecodeError as exc:
        raise VaultContractError("invalid KeePass diagnostic") from exc
    if any(character not in "\t\n" and not character.isprintable() for character in text):
        raise VaultContractError("invalid KeePass diagnostic")


def run_keepass(
    arguments: Sequence[str],
    master_buffer: bytearray,
    private_home: str,
    *,
    webhook: RetainedWebhook | None = None,
    timeout: float = DEFAULT_CHILD_TIMEOUT,
    check: bool = True,
) -> KeepassResult:
    if not arguments or any(not isinstance(value, str) or "\0" in value for value in arguments):
        raise VaultContractError("invalid KeePass arguments")
    if not os.path.isabs(private_home):
        raise VaultContractError("private home must be absolute")
    child_input = bytearray(master_buffer)
    child_input.extend(b"\n")
    if webhook is not None:
        protected = bytearray(webhook.read_bytes())
        try:
            child_input.extend(protected)
            child_input.extend(b"\n")
        finally:
            wipe_mutable_buffer(protected)
    environment = {
        "HOME": os.devnull,
        "XDG_CONFIG_HOME": os.devnull,
        "LC_ALL": "C.UTF-8",
        "PATH": "/usr/bin:/bin",
    }
    expected_parent_pid = os.getpid()
    def prepare_child() -> None:
        os.umask(0o077)
        arm_parent_death_signal(expected_parent_pid)
    process = subprocess.Popen(
        [KEEPASSXC_CLI, *arguments],
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        env=environment,
        close_fds=True,
        start_new_session=True,
        preexec_fn=prepare_child,
    )
    try:
        try:
            stdout, stderr = process.communicate(bytes(child_input), timeout=timeout)
        except subprocess.TimeoutExpired as exc:
            try:
                os.killpg(process.pid, 15)
            except ProcessLookupError:
                pass
            try:
                process.communicate(timeout=1.0)
            except subprocess.TimeoutExpired:
                try:
                    os.killpg(process.pid, 9)
                except ProcessLookupError:
                    pass
                process.communicate()
            raise VaultContractError("KeePass child timeout") from exc
        if len(stdout) > MAX_CHILD_OUTPUT_BYTES:
            raise VaultContractError("KeePass output is oversized")
        _validate_child_stderr(stderr)
        result = KeepassResult(process.returncode, stdout, stderr)
        if check and process.returncode != 0:
            raise VaultContractError("KeePass command failed")
        return result
    finally:
        wipe_mutable_buffer(child_input)


def arm_parent_death_signal(expected_parent_pid: int) -> None:
    """Ensure a vault death cannot orphan a secret-bearing KeePass child."""
    if expected_parent_pid <= 1:
        raise VaultContractError("invalid expected parent PID")
    libc = ctypes.CDLL(None, use_errno=True)
    prctl = libc.prctl
    rc = prctl(1, 9, 0, 0, 0)  # PR_SET_PDEATHSIG, SIGKILL
    if rc != 0:
        err = ctypes.get_errno()
        raise OSError(err, os.strerror(err))
    if os.getppid() != expected_parent_pid:
        os.kill(os.getpid(), 9)


def expected_notes(app_name: str) -> str:
    _validate_app_name(app_name.encode("utf-8"))
    return (
        "channel=desktop-infra-전체;"
        f"app={app_name};"
        "recovery=revoke-and-reissue-in-slack"
    )


def _entry_count(
    database: str,
    master: bytearray,
    private_home: str,
    child_timeout: float = DEFAULT_CHILD_TIMEOUT,
) -> tuple[int, set[str]]:
    result = run_keepass(
        ["ls", "-q", "-R", "-f", database], master, private_home,
        check=False, timeout=child_timeout,
    )
    if result.returncode != 0:
        raise VaultContractError("KeePass database unlock failed")
    try:
        text = result.stdout.decode("utf-8", "strict")
    except UnicodeDecodeError as exc:
        raise VaultContractError("invalid KeePass listing") from exc
    lines = text.splitlines()
    exact = ENTRY_PATH.lstrip("/")
    return sum(line == exact for line in lines), set(lines)


def _shown_attributes(
    database: str,
    master: bytearray,
    private_home: str,
    child_timeout: float = DEFAULT_CHILD_TIMEOUT,
) -> tuple[str, str, str, str]:
    result = run_keepass(
        [
            "show", "-q", "-a", "Title", "-a", "UserName", "-a", "URL",
            "-a", "Notes", database, ENTRY_PATH,
        ],
        master,
        private_home,
        timeout=child_timeout,
    )
    try:
        text = result.stdout.decode("utf-8", "strict")
    except UnicodeDecodeError as exc:
        raise VaultContractError("invalid KeePass attributes") from exc
    lines = text.split("\n")
    if len(lines) != 5 or lines[-1] != "" or any("\r" in line for line in lines[:-1]):
        raise VaultContractError("invalid KeePass attribute framing")
    return lines[0], lines[1], lines[2], lines[3]


def protected_value_matches(producer_rc: int, shown: bytes, expected: bytes) -> bool:
    comparison = bytearray(expected)
    comparison.extend(b"\n")
    try:
        return producer_rc == 0 and hmac.compare_digest(shown, bytes(comparison))
    finally:
        wipe_mutable_buffer(comparison)


def _password_matches(
    database: str,
    master: bytearray,
    webhook: RetainedWebhook,
    private_home: str,
    child_timeout: float = DEFAULT_CHILD_TIMEOUT,
) -> bool:
    result = run_keepass(
        ["show", "-q", "-s", "-a", "Password", database, ENTRY_PATH],
        master,
        private_home,
        check=False,
        timeout=child_timeout,
    )
    expected = bytearray(webhook.read_bytes())
    shown = bytearray(result.stdout)
    try:
        return protected_value_matches(result.returncode, bytes(shown), bytes(expected))
    finally:
        wipe_mutable_buffer(expected)
        wipe_mutable_buffer(shown)


def classify_database(
    database: str,
    master_buffer: bytearray,
    webhook: RetainedWebhook,
    app_name: str,
    private_home: str,
    child_timeout: float = DEFAULT_CHILD_TIMEOUT,
) -> str:
    try:
        count, _ = _entry_count(database, master_buffer, private_home, child_timeout)
        if count == 0:
            return "absent"
        if count > 1:
            return "ambiguous"
        expected = (ENTRY_TITLE, ENTRY_USERNAME, ENTRY_URL, expected_notes(app_name))
        if _shown_attributes(database, master_buffer, private_home, child_timeout) != expected:
            return "mismatch"
        if not _password_matches(
            database, master_buffer, webhook, private_home, child_timeout
        ):
            return "mismatch"
        return "exact-noop"
    except (OSError, VaultContractError):
        return "failed"


def _copy_private_database(source: str, destination: str) -> None:
    source_fd = os.open(source, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC)
    destination_fd = -1
    try:
        _regular_metadata(source_fd)
        destination_fd = os.open(
            destination,
            os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW | os.O_CLOEXEC,
            0o600,
        )
        while True:
            chunk = os.read(source_fd, 1024 * 1024)
            if not chunk:
                break
            view = memoryview(chunk)
            while view:
                written = os.write(destination_fd, view)
                view = view[written:]
        os.fsync(destination_fd)
        _regular_metadata(destination_fd)
    finally:
        if destination_fd >= 0:
            os.close(destination_fd)
        os.close(source_fd)


def _ensure_groups(
    database: str,
    master: bytearray,
    private_home: str,
    existing_lines: set[str],
    child_timeout: float = DEFAULT_CHILD_TIMEOUT,
) -> None:
    for group in ("/Platform", "/Platform/Observability", ENTRY_GROUP):
        flattened = group.lstrip("/") + "/"
        if flattened not in existing_lines:
            run_keepass(
                ["mkdir", "-q", database, group], master, private_home,
                timeout=child_timeout,
            )
            existing_lines.add(flattened)


def build_candidate(
    baseline: str,
    candidate: str,
    mode: str,
    master_buffer: bytearray,
    webhook: RetainedWebhook,
    app_name: str,
    private_home: str,
    child_timeout: float = DEFAULT_CHILD_TIMEOUT,
) -> None:
    if mode not in {"ADD", "UPDATE"}:
        raise VaultContractError("invalid candidate mode")
    classification = classify_database(
        baseline, master_buffer, webhook, app_name, private_home, child_timeout
    )
    if (mode == "ADD" and classification != "absent") or (
        mode == "UPDATE" and classification != "mismatch"
    ):
        raise VaultContractError("candidate mode does not match classification")
    _copy_private_database(baseline, candidate)
    try:
        _, existing_lines = _entry_count(
            candidate, master_buffer, private_home, child_timeout
        )
        notes = expected_notes(app_name)
        if mode == "ADD":
            _ensure_groups(
                candidate, master_buffer, private_home, existing_lines, child_timeout
            )
            arguments = [
                "add", "-q", "-p", "-u", ENTRY_USERNAME, "--url", ENTRY_URL,
                "--notes", notes, candidate, ENTRY_PATH,
            ]
        else:
            arguments = [
                "edit", "-q", "-p", "-t", ENTRY_TITLE, "-u", ENTRY_USERNAME,
                "--url", ENTRY_URL, "--notes", notes, candidate, ENTRY_PATH,
            ]
        run_keepass(
            arguments, master_buffer, private_home, webhook=webhook,
            timeout=child_timeout,
        )
        if not verify_database(
            candidate, master_buffer, webhook, app_name, private_home, child_timeout
        ):
            raise VaultContractError("candidate verification failed")
    except BaseException:
        try:
            os.unlink(candidate)
        except OSError:
            pass
        raise


def verify_database(
    database: str,
    master_buffer: bytearray,
    webhook: RetainedWebhook,
    app_name: str,
    private_home: str,
    child_timeout: float = DEFAULT_CHILD_TIMEOUT,
) -> bool:
    return classify_database(
        database, master_buffer, webhook, app_name, private_home, child_timeout
    ) == "exact-noop"


def _attest_runtime_root(runtime_root: str) -> str:
    if not os.path.isabs(runtime_root):
        raise VaultContractError("absolute runtime root required")
    canonical = os.path.realpath(runtime_root)
    if canonical != os.path.abspath(runtime_root):
        raise VaultContractError("runtime root must be canonical")
    expected_parent = f"/run/user/{os.getuid()}"
    if os.path.commonpath((canonical, expected_parent)) != expected_parent or canonical == expected_parent:
        raise VaultContractError("runtime root is outside the private runtime")
    fd = os.open(canonical, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC)
    try:
        info = os.fstat(fd)
        if info.st_uid != os.getuid() or stat.S_IMODE(info.st_mode) != 0o700:
            raise VaultContractError("runtime root metadata mismatch")
    finally:
        os.close(fd)
    mount_type = None
    best_length = -1
    with open("/proc/self/mountinfo", "r", encoding="utf-8") as stream:
        for line in stream:
            left, right = line.rstrip("\n").split(" - ", 1)
            mount_point = left.split()[4].replace("\\040", " ")
            if canonical == mount_point or canonical.startswith(mount_point.rstrip("/") + "/"):
                if len(mount_point) > best_length:
                    mount_type = right.split()[0]
                    best_length = len(mount_point)
    if mount_type != "tmpfs":
        raise VaultContractError("runtime root is not on tmpfs")
    return canonical


def _retained_runtime_path(runtime_fd: int) -> str:
    info = os.fstat(runtime_fd)
    if not stat.S_ISDIR(info.st_mode):
        raise VaultContractError("retained runtime is not a directory")
    proc_path = f"/proc/{os.getpid()}/fd/{runtime_fd}"
    link_info = os.lstat(proc_path)
    if not stat.S_ISLNK(link_info.st_mode):
        raise VaultContractError("retained runtime proc anchor is unavailable")
    resolved = os.stat(proc_path)
    if (resolved.st_dev, resolved.st_ino) != (info.st_dev, info.st_ino):
        raise VaultContractError("retained runtime proc anchor mismatch")
    return proc_path


def _open_snapshot_below_runtime(
    path: str, runtime_root: str, runtime_fd: int
) -> int:
    if not os.path.isabs(path) or os.path.normpath(path) != path:
        raise VaultProtocolError("snapshot must be canonical")
    relative = os.path.relpath(path, runtime_root)
    parts = relative.split(os.sep)
    if not parts or any(
        not part or part in {".", ".."} or os.sep in part for part in parts
    ):
        raise VaultProtocolError("snapshot is outside runtime root")
    current_fd = os.dup(runtime_fd)
    try:
        for part in parts[:-1]:
            next_fd = os.open(
                part,
                os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC,
                dir_fd=current_fd,
            )
            os.close(current_fd)
            current_fd = next_fd
            directory_info = os.fstat(current_fd)
            if (
                directory_info.st_uid != os.getuid()
                or stat.S_IMODE(directory_info.st_mode) != 0o700
            ):
                raise VaultProtocolError("snapshot directory metadata mismatch")
        return os.open(
            parts[-1],
            os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC,
            dir_fd=current_fd,
        )
    finally:
        os.close(current_fd)


def _attest_snapshot(
    path: str, runtime_root: str, runtime_fd: int | None = None
) -> None:
    if runtime_fd is not None:
        fd = _open_attested_snapshot(path, runtime_root, runtime_fd)
        try:
            return
        finally:
            os.close(fd)
    if not os.path.isabs(path) or os.path.realpath(path) != os.path.abspath(path):
        raise VaultProtocolError("snapshot must be canonical")
    if os.path.commonpath((path, runtime_root)) != runtime_root or path == runtime_root:
        raise VaultProtocolError("snapshot is outside runtime root")
    fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC)
    try:
        _regular_metadata(fd)
    finally:
        os.close(fd)


def _open_attested_snapshot(path: str, runtime_root: str, runtime_fd: int) -> int:
    fd = _open_snapshot_below_runtime(path, runtime_root, runtime_fd)
    try:
        _regular_metadata(fd)
        return fd
    except BaseException:
        os.close(fd)
        raise


def read_protocol_line(connection: socket.socket, *, timeout: float) -> str:
    connection.settimeout(timeout)
    data = bytearray()
    try:
        while len(data) <= MAX_PROTOCOL_BYTES:
            chunk = connection.recv(MAX_PROTOCOL_BYTES + 2 - len(data))
            if not chunk:
                raise VaultProtocolError("partial protocol EOF")
            data.extend(chunk)
            if b"\n" in data:
                break
        if len(data) > MAX_PROTOCOL_BYTES + 1:
            raise VaultProtocolError("protocol request is oversized")
        if not data.endswith(b"\n") or data.count(b"\n") != 1:
            raise VaultProtocolError("invalid protocol framing")
        line = data[:-1]
        if b"\r" in line or b"\0" in line:
            raise VaultProtocolError("invalid protocol bytes")
        try:
            return line.decode("utf-8", "strict")
        except UnicodeDecodeError as exc:
            raise VaultProtocolError("invalid protocol UTF-8") from exc
    finally:
        wipe_mutable_buffer(data)


def parse_protocol_command(line: str) -> tuple[str, str | None]:
    if line in {"PREPARE", "VERIFY_CANDIDATE", "SHUTDOWN"}:
        return line, None
    if line in {"BUILD ADD", "BUILD UPDATE"}:
        return "BUILD", line.split(" ", 1)[1]
    prefix = "VERIFY_COMMITTED "
    if line.startswith(prefix):
        path = line[len(prefix):]
        if not path or " " in path or not os.path.isabs(path):
            raise VaultProtocolError("invalid committed snapshot command")
        return "VERIFY_COMMITTED", path
    raise VaultProtocolError("unknown protocol command")


def peer_uid(connection: socket.socket) -> int:
    credentials = connection.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12)
    _, uid, _ = struct.unpack("3i", credentials)
    return uid


def _write_reply(connection: socket.socket, reply: str, timeout: float) -> None:
    payload = reply.encode("ascii") + b"\n"
    connection.settimeout(timeout)
    connection.sendall(payload)


def serve_private_socket(
    runtime_root: str,
    webhook_file: str,
    *,
    input_fd: int | None = None,
    output_fd: int | None = None,
    accept_timeout: float = DEFAULT_ACCEPT_TIMEOUT,
    io_timeout: float = DEFAULT_IO_TIMEOUT,
    child_timeout: float = DEFAULT_CHILD_TIMEOUT,
    wipe_hook: Callable[[bytearray], None] | None = None,
) -> None:
    if child_timeout <= 0:
        raise VaultContractError("invalid child timeout")
    runtime = ""
    runtime_fd = -1
    runtime_access = ""
    webhook = None
    master = bytearray()
    listener = None
    socket_identity = None
    try:
        runtime = _attest_runtime_root(runtime_root)
        runtime_fd = os.open(
            runtime, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC
        )
        runtime_access = _retained_runtime_path(runtime_fd)
        webhook = open_validated_webhook(webhook_file)
        listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        socket_path = os.path.join(runtime_access, SOCKET_NAME)
        private_home = os.devnull
        baseline = os.path.join(runtime_access, BASELINE_NAME)
        candidate = os.path.join(runtime_access, CANDIDATE_NAME)
        app_name, master = read_operator_secrets(input_fd=input_fd, output_fd=output_fd)
        previous_mask = signal.pthread_sigmask(signal.SIG_BLOCK, HANDLED_SIGNALS)
        try:
            listener.bind(socket_path)
            os.chmod(SOCKET_NAME, 0o600, dir_fd=runtime_fd, follow_symlinks=False)
            socket_info = os.stat(
                SOCKET_NAME, dir_fd=runtime_fd, follow_symlinks=False
            )
            socket_identity = (socket_info.st_dev, socket_info.st_ino)
            listener.listen(1)
        finally:
            signal.pthread_sigmask(signal.SIG_SETMASK, previous_mask)
        listener.settimeout(accept_timeout)
        connection, _ = listener.accept()
        listener.close()
        if peer_uid(connection) != os.getuid():
            connection.close()
            raise VaultContractError("socket peer UID mismatch")
        state = "initial"
        classification = None
        with connection:
            while True:
                try:
                    command, argument = parse_protocol_command(
                        read_protocol_line(connection, timeout=io_timeout)
                    )
                    if command == "PREPARE":
                        if state != "initial":
                            raise VaultProtocolError("PREPARE out of state")
                        classification = classify_database(
                            baseline, master, webhook, app_name, private_home,
                            child_timeout,
                        )
                        state = "prepared"
                        _write_reply(connection, classification, io_timeout)
                    elif command == "BUILD":
                        if state != "prepared" or (
                            argument == "ADD" and classification != "absent"
                        ) or (
                            argument == "UPDATE" and classification != "mismatch"
                        ):
                            raise VaultProtocolError("BUILD out of state")
                        try:
                            build_candidate(
                                baseline, candidate, argument or "", master,
                                webhook, app_name, private_home, child_timeout,
                            )
                        except (OSError, VaultContractError):
                            _write_reply(connection, "failed", io_timeout)
                            continue
                        state = "candidate-ready"
                        _write_reply(connection, "candidate-ready", io_timeout)
                    elif command == "VERIFY_CANDIDATE":
                        if state != "candidate-ready":
                            raise VaultProtocolError("VERIFY_CANDIDATE out of state")
                        verified = verify_database(
                            candidate, master, webhook, app_name, private_home,
                            child_timeout,
                        )
                        if verified:
                            state = "candidate-verified"
                        _write_reply(connection, "verified" if verified else "failed", io_timeout)
                    elif command == "VERIFY_COMMITTED":
                        if state != "candidate-verified" or argument is None:
                            raise VaultProtocolError("VERIFY_COMMITTED out of state")
                        snapshot_fd = _open_attested_snapshot(
                            argument, runtime, runtime_fd
                        )
                        try:
                            verified = verify_database(
                                f"/proc/{os.getpid()}/fd/{snapshot_fd}",
                                master,
                                webhook,
                                app_name,
                                private_home,
                                child_timeout,
                            )
                        finally:
                            os.close(snapshot_fd)
                        _write_reply(connection, "verified" if verified else "failed", io_timeout)
                    elif command == "SHUTDOWN":
                        _write_reply(connection, "stopped", io_timeout)
                        break
                except VaultProtocolError:
                    try:
                        _write_reply(connection, "failed", io_timeout)
                    except OSError:
                        pass
                    raise
    finally:
        try:
            if listener is not None:
                listener.close()
        finally:
            try:
                if webhook is not None:
                    webhook.close()
            finally:
                try:
                    wipe_mutable_buffer(master)
                    if wipe_hook is not None:
                        wipe_hook(master)
                finally:
                    try:
                        if socket_identity is not None:
                            try:
                                current = os.stat(
                                    SOCKET_NAME, dir_fd=runtime_fd,
                                    follow_symlinks=False,
                                )
                                if (current.st_dev, current.st_ino) == socket_identity:
                                    os.unlink(SOCKET_NAME, dir_fd=runtime_fd)
                            except FileNotFoundError:
                                pass
                    finally:
                        if runtime_fd >= 0:
                            os.close(runtime_fd)


def _main(argv: Sequence[str]) -> int:
    if len(argv) != 5 or argv[0] != "--serve" or argv[1] != "--runtime-root" or argv[3] != "--webhook-file":
        return 2
    runtime_root, webhook_file = argv[2], argv[4]
    if not os.path.isabs(runtime_root) or not os.path.isabs(webhook_file):
        return 2
    install_process_signal_handlers()
    try:
        serve_private_socket(runtime_root, webhook_file)
        return 0
    except VaultSignal as exc:
        return 128 + exc.signum
    except (OSError, VaultContractError):
        os.write(2, b"vault helper failed\n")
        return 1


if __name__ == "__main__":
    raise SystemExit(_main(sys.argv[1:]))