# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/inside-what-filesystem-and-process-quarantine-gates-must-an-instance-confine-untrusted-archive-extractions-to-prevent-symlink-and-zip-slip-breakouts.md # Remnant Warrant: https://corpusfetch.com/c/inside-what-filesystem-and-process-quarantine-gates-must-an-instance-confine-untrusted-archive-extractions-to-prevent-symlink-and-zip-slip-breakouts.warrant.json # Markdown SHA-256: 73c4edd0e69abc894dc96471afda44de1eac2f324b390db951055e64de617829 # Code AST SHA-256: 2838a031d4090e780b337ac2112c8e0eb043fb04c268b6a66cb518768a3c3281 # Verified Classes: ArchiveQuarantineBreakoutError, ArchiveMemberHeader, ArchiveQuarantineExtractionGate # Defensive Guards: 8 fail-closed invariants (zero external deps) from dataclasses import dataclass from pathlib import Path, PurePosixPath from typing import Final class ArchiveQuarantineBreakoutError(RuntimeError): """Raised when an archive member attempts path traversal, symlink escape, or compression bomb exhaustion.""" pass @dataclass(frozen=True) class ArchiveMemberHeader: filename: str uncompressed_size: int is_symlink: bool = False is_hardlink: bool = False is_device_or_fifo: bool = False class ArchiveQuarantineExtractionGate: """ Confines untrusted archive extraction (Zip-Slip / Symlink Escape / Bomb Defense): 1. Rejects symlinks, hardlinks, FIFOs, and special device nodes outright. 2. Canonicalizes target paths inside a dedicated quarantine root and verifies `.is_relative_to()`. 3. Enforces per-file and cumulative byte budgets plus maximum member count. """ MAX_MEMBER_COUNT: Final[int] = 1_000 MAX_SINGLE_FILE_BYTES: Final[int] = 50 * 1024 * 1024 MAX_TOTAL_EXTRACTED_BYTES: Final[int] = 200 * 1024 * 1024 @classmethod def validate_manifest(cls, quarantine_root: Path, members: list[ArchiveMemberHeader]) -> list[Path]: if not quarantine_root.is_absolute(): raise ArchiveQuarantineBreakoutError("quarantine_root must be an absolute path.") resolved_root = quarantine_root.resolve() if len(members) > cls.MAX_MEMBER_COUNT: raise ArchiveQuarantineBreakoutError( f"Archive member count ({len(members)}) exceeds cap ({cls.MAX_MEMBER_COUNT})." ) cumulative_bytes = 0 safe_destinations: list[Path] = [] for m in members: if "\x00" in m.filename or not m.filename.strip(): raise ArchiveQuarantineBreakoutError("Null byte or empty filename in archive header.") if m.is_symlink or m.is_hardlink or m.is_device_or_fifo: raise ArchiveQuarantineBreakoutError( f"Forbidden special file or link in archive: {m.filename!r}" ) if m.uncompressed_size < 0 or m.uncompressed_size > cls.MAX_SINGLE_FILE_BYTES: raise ArchiveQuarantineBreakoutError( f"Member {m.filename!r} size ({m.uncompressed_size}) exceeds single-file cap." ) posix_rel = PurePosixPath(m.filename) if posix_rel.is_absolute() or ".." in posix_rel.parts: raise ArchiveQuarantineBreakoutError( f"Zip-Slip path traversal blocked in member: {m.filename!r}" ) candidate = (resolved_root / Path(*posix_rel.parts)).resolve() if not candidate.is_relative_to(resolved_root): raise ArchiveQuarantineBreakoutError( f"Resolved path escaped quarantine root: {candidate}" ) cumulative_bytes += m.uncompressed_size if cumulative_bytes > cls.MAX_TOTAL_EXTRACTED_BYTES: raise ArchiveQuarantineBreakoutError( f"Cumulative extraction size ({cumulative_bytes}) exceeds quarantine quota." ) safe_destinations.append(candidate) return safe_destinations