"""Private filesystem and journal primitives for customer cleanup.""" from __future__ import annotations import hashlib import json import os import stat from collections.abc import Iterable from pathlib import Path from checkin_cli.customer_cleanup_models import CleanupIntegrityError PHASES = ("prepared", "copied_verified", "source_pruned", "committed") def canonical(value: object) -> bytes: return json.dumps(value, ensure_ascii=True, sort_keys=True, separators=(",", ":")).encode() def digest_bytes(value: bytes) -> str: return hashlib.sha256(value).hexdigest() def ensure_private_dir(path: Path) -> None: path.mkdir(parents=True, exist_ok=True, mode=0o700) path.chmod(0o700) info = path.lstat() if stat.S_ISLNK(info.st_mode) or not stat.S_ISDIR(info.st_mode): raise CleanupIntegrityError("cleanup directory is not a regular directory") if info.st_uid != os.geteuid() or stat.S_IMODE(info.st_mode) != 0o700: raise CleanupIntegrityError("cleanup directory is not owner-only") def validate_file(path: Path) -> os.stat_result: info = path.lstat() if stat.S_ISLNK(info.st_mode): raise CleanupIntegrityError(f"cleanup source symlink is forbidden: {path}") if not stat.S_ISREG(info.st_mode): raise CleanupIntegrityError(f"cleanup source is not regular: {path}") if info.st_nlink != 1: raise CleanupIntegrityError(f"cleanup source is hard-linked: {path}") if info.st_uid != os.geteuid() or stat.S_IMODE(info.st_mode) & 0o077: raise CleanupIntegrityError(f"cleanup source must be owner-only: {path}") return info def safe_relative(path: Path, root: Path, customer_key: str) -> str: lexical = path.absolute() try: relative = lexical.relative_to(root) except ValueError as exc: raise CleanupIntegrityError("cleanup path traversal escapes profile") from exc current = root for part in relative.parts: current /= part if current.is_symlink(): raise CleanupIntegrityError(f"cleanup source symlink is forbidden: {current}") parts = relative.parts for index, part in enumerate(parts[:-1]): if part == "customers" and parts[index + 1] != customer_key: raise CleanupIntegrityError("cross-customer cleanup path is forbidden") return relative.as_posix() def source_inventory( root: Path, customer_key: str, candidates: Iterable[Path] ) -> list[dict[str, object]]: customer_root = root / "data" / "customers" / customer_key paths: set[Path] = set() if customer_root.exists() or customer_root.is_symlink(): if customer_root.is_symlink(): raise CleanupIntegrityError("cleanup customer root symlink is forbidden") for current, directories, files in os.walk(customer_root, followlinks=False): base = Path(current) for name in directories: child = base / name if child.is_symlink(): raise CleanupIntegrityError(f"cleanup source symlink is forbidden: {child}") paths.update(base / name for name in files) for candidate in candidates: path = candidate if candidate.is_absolute() else root / candidate relative = safe_relative(path, root, customer_key) if f"/customers/{customer_key}/" not in f"/{relative}" and customer_key not in Path(relative).name: raise CleanupIntegrityError("candidate path is not customer-scoped") paths.add(path) rows: list[dict[str, object]] = [] for path in paths: relative = safe_relative(path, root, customer_key) info = validate_file(path) payload = path.read_bytes() after = path.stat() if (info.st_dev, info.st_ino, info.st_size) != (after.st_dev, after.st_ino, after.st_size): raise CleanupIntegrityError("cleanup source changed while inventoried") rows.append({ "relative_path": relative, "mode": stat.S_IMODE(info.st_mode), "size": len(payload), "sha256": digest_bytes(payload), }) return sorted(rows, key=lambda row: str(row["relative_path"])) def write_private(path: Path, payload: bytes, *, exclusive: bool = False) -> None: ensure_private_dir(path.parent) flags = os.O_WRONLY | os.O_CREAT | os.O_CLOEXEC | os.O_NOFOLLOW flags |= os.O_EXCL if exclusive else os.O_TRUNC descriptor = os.open(path, flags, 0o600) try: os.fchmod(descriptor, 0o600) remaining = memoryview(payload) while remaining: remaining = remaining[os.write(descriptor, remaining) :] os.fsync(descriptor) finally: os.close(descriptor) def append_phase( journal: Path, phase: str, operation_id: str, *, manifest_sha256: str | None = None, ) -> None: rows = read_journal(journal) expected = PHASES[len(rows)] if len(rows) < len(PHASES) else None if phase != expected: raise CleanupIntegrityError("cleanup journal phase is not forward-only") previous = digest_bytes(canonical(rows[-1])) if rows else "0" * 64 row = {"operation_id": operation_id, "phase": phase, "previous_digest": previous} if manifest_sha256 is not None: row["manifest_sha256"] = manifest_sha256 ensure_private_dir(journal.parent) descriptor = os.open(journal, os.O_APPEND | os.O_CREAT | os.O_WRONLY | os.O_NOFOLLOW, 0o600) try: os.fchmod(descriptor, 0o600) remaining = memoryview(canonical(row) + b"\n") while remaining: remaining = remaining[os.write(descriptor, remaining) :] os.fsync(descriptor) finally: os.close(descriptor) def read_journal(path: Path) -> list[dict[str, object]]: if not path.exists(): return [] validate_file(path) try: rows = [json.loads(line) for line in path.read_text().splitlines() if line] except (OSError, json.JSONDecodeError) as exc: raise CleanupIntegrityError("cleanup journal is invalid") from exc if [row.get("phase") for row in rows] != list(PHASES[: len(rows)]): raise CleanupIntegrityError("cleanup journal phase is not forward-only") operation_ids = {row.get("operation_id") for row in rows} if len(operation_ids) > 1: raise CleanupIntegrityError("cleanup journal operation changed") for index, row in enumerate(rows): expected = "0" * 64 if index == 0 else digest_bytes(canonical(rows[index - 1])) if row.get("previous_digest") != expected: raise CleanupIntegrityError("cleanup journal digest chain is invalid") return rows def copy_inventory(root: Path, archive: Path, rows: list[dict[str, object]]) -> None: files_root = archive / "files" ensure_private_dir(files_root) for row in rows: source = root / str(row["relative_path"]) validate_file(source) payload = source.read_bytes() if len(payload) != row["size"] or digest_bytes(payload) != row["sha256"]: raise CleanupIntegrityError("cleanup source changed before archive copy") destination = files_root / str(row["relative_path"]) write_private(destination, payload, exclusive=not destination.exists()) def verify_archive(archive: Path, rows: list[dict[str, object]]) -> None: for row in rows: path = archive / "files" / str(row["relative_path"]) try: validate_file(path) payload = path.read_bytes() except (OSError, CleanupIntegrityError) as exc: raise CleanupIntegrityError("archive verification failed") from exc if len(payload) != row["size"] or digest_bytes(payload) != row["sha256"]: raise CleanupIntegrityError("archive verification failed") def prune_sources(root: Path, rows: list[dict[str, object]], customer_key: str) -> None: for row in rows: path = root / str(row["relative_path"]) if not path.exists(): continue validate_file(path) payload = path.read_bytes() if len(payload) != row["size"] or digest_bytes(payload) != row["sha256"]: raise CleanupIntegrityError("cleanup source changed before pruning") path.unlink() customer_root = root / "data" / "customers" / customer_key if customer_root.exists(): for path in sorted(customer_root.rglob("*"), key=lambda item: len(item.parts), reverse=True): if path.is_dir(): path.rmdir() customer_root.rmdir() def freeze_archive(archive: Path) -> None: for path in archive.rglob("*"): path.chmod(0o500 if path.is_dir() else 0o400) archive.chmod(0o500)