atlas-iac/services/hermes/scripts/cli_lane_quarantine.py

#!/usr/bin/env python3
"""Bounded diagnostic quarantine for malformed terminal journals."""

from __future__ import annotations

import contextlib
import hashlib
import json
import os
import re
import stat
import sys
from pathlib import Path

from cli_lane_config import QUARANTINE_HASH_BYTES, STATE_ROOT, TerminalIdentity, TerminalRecoverySnapshot
from cli_lane_evidence import _retire_terminal_entry
from cli_lane_records import _read_bounded


def _quarantine_terminal(
    path: Path,
    identity: TerminalIdentity | None,
    reason: str,
    *,
    snapshot: TerminalRecoverySnapshot | None = None,
) -> Path:
    """Record bounded metadata, then retire only the inspected journal inode."""
    board_dir = path.parent
    try:
        board_stat = board_dir.stat(follow_symlinks=False)
    except OSError:
        board_stat = None
    if board_stat is None or not stat.S_ISDIR(board_stat.st_mode):
        print(
            f"refused terminal journal path through unsafe board directory: {path}",
            file=sys.stderr,
            flush=True,
        )
        return path
    board_flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)
    board_flags |= getattr(os, "O_NOFOLLOW", 0)
    file_flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
    create_flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
    create_flags |= getattr(os, "O_NOFOLLOW", 0)
    board_descriptor = None
    quarantine_descriptor = None
    source_descriptor = None
    source_stat = None
    destination = path
    retirement = "not-attempted"
    owns_source = snapshot is None
    try:
        relative = path.relative_to(STATE_ROOT)
    except ValueError:
        relative = Path()
    already_quarantined = (
        len(relative.parts) == 3 and relative.parts[1] == "quarantine"
    )
    try:
        if snapshot is None:
            board_descriptor = os.open(board_dir, board_flags)
            source_stat = os.stat(
                path.name,
                dir_fd=board_descriptor,
                follow_symlinks=False,
            )
        else:
            board_descriptor = snapshot.directory_descriptor
            source_descriptor = snapshot.descriptor
            source_stat = snapshot.file_stat
            pinned_board = os.fstat(board_descriptor)
            if (
                pinned_board.st_dev != board_stat.st_dev
                or pinned_board.st_ino != board_stat.st_ino
            ):
                raise OSError("terminal recovery board identity changed")
        safe_source = stat.S_ISREG(source_stat.st_mode) and source_stat.st_nlink == 1
        digest = None
        hashed_bytes = 0
        hash_complete = False
        if safe_source and snapshot is None:
            source_descriptor = os.open(
                path.name,
                file_flags,
                dir_fd=board_descriptor,
            )
            opened_stat = os.fstat(source_descriptor)
            if (
                opened_stat.st_dev != source_stat.st_dev
                or opened_stat.st_ino != source_stat.st_ino
                or opened_stat.st_nlink != 1
            ):
                raise OSError("terminal journal identity changed while opening")
            # Keep this descriptor open through retirement. The staging entry
            # must match the exact fstat-verified inode, not merely a path stat
            # captured before a concurrent replacement.
            source_stat = opened_stat
            prefix = _read_bounded(source_descriptor, QUARANTINE_HASH_BYTES)
            hashed_bytes = len(prefix)
            hash_complete = opened_stat.st_size <= QUARANTINE_HASH_BYTES
            digest = hashlib.sha256(prefix).hexdigest()
        elif safe_source and source_descriptor is not None:
            prefix = snapshot.prefix
            hashed_bytes = len(prefix)
            hash_complete = (
                source_stat.st_size <= QUARANTINE_HASH_BYTES
                and hashed_bytes == source_stat.st_size
            )
            digest = hashlib.sha256(prefix).hexdigest()
        source_kind = (
            "regular"
            if safe_source
            else "hardlink"
            if stat.S_ISREG(source_stat.st_mode) and source_stat.st_nlink != 1
            else "non-regular"
        )
        diagnostic = {
            "hash_complete": hash_complete,
            "hashed_bytes": hashed_bytes,
            "original_name": path.name,
            "reason": reason,
            "sha256": digest,
            "size": source_stat.st_size,
            "source_kind": source_kind,
        }
        payload = (json.dumps(diagnostic, sort_keys=True) + "\n").encode("utf-8")
        fingerprint = (
            (digest or hashlib.sha256(path.name.encode("utf-8")).hexdigest())[:16]
        )
        safe_reason = (
            re.sub(r"[^a-zA-Z0-9_.-]+", "-", reason).strip("-") or "invalid"
        )
        if already_quarantined:
            # A staged artifact already lives in the board's one bounded
            # quarantine directory. Reuse it so adversarial replacements can
            # never drive recursive quarantine/quarantine/... authority.
            quarantine_descriptor = os.dup(board_descriptor)
            quarantine_dir = board_dir
        else:
            with contextlib.suppress(FileExistsError):
                os.mkdir("quarantine", 0o700, dir_fd=board_descriptor)
            quarantine_descriptor = os.open(
                "quarantine",
                board_flags,
                dir_fd=board_descriptor,
            )
            quarantine_dir = board_dir / "quarantine"
        for sequence in range(32):
            name = f"{path.name}.{safe_reason}.{fingerprint}.{sequence}.quarantine"
            try:
                destination_descriptor = os.open(
                    name,
                    create_flags,
                    0o600,
                    dir_fd=quarantine_descriptor,
                )
            except FileExistsError:
                continue
            try:
                with os.fdopen(destination_descriptor, "wb") as target:
                    target.write(payload)
                    target.flush()
                    os.fsync(target.fileno())
                    target_stat = os.fstat(target.fileno())
                    if (
                        stat.S_IMODE(target_stat.st_mode) != 0o600
                        or target_stat.st_nlink != 1
                    ):
                        raise OSError("quarantine destination is not private and singly linked")
            except Exception:
                os.unlink(name, dir_fd=quarantine_descriptor)
                raise
            destination = quarantine_dir / name
            break
        else:
            raise OSError("could not reserve a unique quarantine destination")
        os.fsync(quarantine_descriptor)
        retirement = _retire_terminal_entry(
            path,
            source_stat,
            board_descriptor=board_descriptor,
            quarantine_descriptor=quarantine_descriptor,
            source_descriptor=source_descriptor,
        )
    except OSError as error:
        # A bad quarantine target must not make an attacker-controlled pending
        # path eligible for replay forever. The same identity-safe staging
        # protocol works in the board directory when quarantine is unusable.
        if board_descriptor is not None and source_stat is not None:
            try:
                retirement = _retire_terminal_entry(
                    path,
                    source_stat,
                    board_descriptor=board_descriptor,
                    quarantine_descriptor=(
                        quarantine_descriptor
                        if quarantine_descriptor is not None
                        else board_descriptor
                    ),
                    source_descriptor=source_descriptor,
                )
            except OSError:
                retirement = "deferred"
        print(
            f"terminal journal quarantine degraded safely: {type(error).__name__}: {error}",
            file=sys.stderr,
            flush=True,
        )
    finally:
        if owns_source and source_descriptor is not None:
            os.close(source_descriptor)
        if quarantine_descriptor is not None:
            os.close(quarantine_descriptor)
        if owns_source and board_descriptor is not None:
            os.close(board_descriptor)
    print(
        "quarantined terminal journal "
        f"{path.name}: {reason}; identity={identity or 'unparseable'}; "
        f"retirement={retirement}",
        file=sys.stderr,
        flush=True,
    )
    return destination