#!/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