"""Safely extract the approved Hypersim final-holdout archives. The archive bytes and ZIP central directories are treated as input metadata. Each member is extracted with an explicit path and CRC check; ``ZipFile.extractall`` is intentionally not used. The operation is resumable: an existing regular file is validated before it is skipped, and incomplete writes use a same-directory temp file followed by an atomic rename. """ from __future__ import annotations import argparse import hashlib import json import os import re import shutil import stat import time import zipfile import zlib from collections.abc import Iterator from contextlib import contextmanager from dataclasses import dataclass from pathlib import Path, PurePosixPath from typing import Any from mcss.file_lock import acquire_file_lock, release_file_lock PROJECT = Path(__file__).resolve().parents[1] SOURCE = PROJECT / "outputs/dataset_access_check_20260918/hypersim_holdout_download_sizes.json" ARCHIVE_ROOT = PROJECT / "data/hypersim_final_holdout_archives" TARGET_ROOT = PROJECT / "data/hypersim_final_holdout_raw" LOG_ROOT = PROJECT / "outputs/hypersim_holdout_extraction_20260919" RESERVE_BYTES = 10 * 1024**3 COPY_CHUNK_BYTES = 8 * 1024**2 SCENE_RE = re.compile(r"^ai_\d{3}_\d{3}$") DRIVE_RE = re.compile(r"^[A-Za-z]:") REPARSE_POINT = 0x0400 @dataclass(frozen=True) class ArchivePlan: scene: str path: Path expected_bytes: int archive_bytes: int infos: tuple[zipfile.ZipInfo, ...] member_count: int uncompressed_bytes: int top_level: tuple[str, ...] def _read_json_object(path: Path, description: str) -> dict[str, Any]: try: value = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as error: raise ValueError(f"Unreadable {description}: {path}") from error if not isinstance(value, dict): raise ValueError(f"Invalid {description}: expected an object: {path}") return value def _is_reparse(path: Path) -> bool: try: mode = path.lstat().st_mode attributes = getattr(path.lstat(), "st_file_attributes", 0) except FileNotFoundError: return False return stat.S_ISLNK(mode) or bool(attributes & REPARSE_POINT) def _lexists(path: Path) -> bool: return os.path.lexists(path) def _assert_project_child(path: Path, parent: Path, description: str) -> Path: parent_resolved = parent.resolve() resolved = path.resolve(strict=False) try: relative = resolved.relative_to(parent_resolved) except ValueError as error: raise ValueError(f"{description} escapes its allowed root: {path}") from error if not relative.parts: raise ValueError(f"{description} may not be the allowed root: {path}") return resolved def _assert_safe_chain(root: Path, path: Path) -> None: root_resolved = root.resolve() resolved = path.resolve(strict=False) try: relative = resolved.relative_to(root_resolved) except ValueError as error: raise ValueError(f"Path escapes extraction root: {path}") from error current = root_resolved if _is_reparse(current): raise ValueError(f"Extraction root is a symlink or reparse point: {current}") for part in relative.parts: current /= part if _lexists(current) and _is_reparse(current): raise ValueError(f"Path contains a symlink or reparse point: {current}") def _ensure_directory(path: Path, root: Path) -> None: _assert_safe_chain(root, path) if _lexists(path): if not path.is_dir(): raise ValueError(f"Expected directory, found non-directory: {path}") return path.mkdir(parents=True, exist_ok=True) _assert_safe_chain(root, path) def _safe_member_parts(name: str) -> tuple[str, ...]: if not isinstance(name, str) or not name or "\x00" in name: raise ValueError(f"Invalid ZIP member name: {name!r}") normalized = name.replace("\\", "/") if normalized.startswith("/") or DRIVE_RE.match(normalized): raise ValueError(f"Absolute or drive-qualified ZIP member: {name!r}") parts = PurePosixPath(normalized).parts if any(part == ".." for part in parts): raise ValueError(f"Parent traversal in ZIP member: {name!r}") safe_parts = tuple(part for part in parts if part not in ("", ".")) if not safe_parts: raise ValueError(f"Empty ZIP member path: {name!r}") return safe_parts def _member_is_symlink(info: zipfile.ZipInfo) -> bool: mode = (info.external_attr >> 16) & 0xFFFF return stat.S_ISLNK(mode) def _member_target(target_root: Path, info: zipfile.ZipInfo) -> Path: parts = _safe_member_parts(info.filename) target = target_root.joinpath(*parts) _assert_safe_chain(target_root, target) return target def _load_rows(source: Path) -> tuple[list[dict[str, Any]], str]: source_bytes = source.read_bytes() payload = json.loads(source_bytes) if not isinstance(payload, dict): raise ValueError("Approved manifest must be a JSON object") rows = payload.get("results") if not isinstance(rows, list) or len(rows) != 26: raise ValueError("Expected exactly 26 approved archive rows") if payload.get("verified_scenes") != 26: raise ValueError("Approved manifest does not contain 26 verified scenes") scenes: list[str] = [] normalized: list[dict[str, Any]] = [] for row in rows: if not isinstance(row, dict): raise ValueError("Approved manifest row is not an object") scene = row.get("scene") if not isinstance(scene, str) or not SCENE_RE.fullmatch(scene): raise ValueError(f"Unsafe scene ID in approved manifest: {scene!r}") if scene in scenes: raise ValueError(f"Duplicate scene in approved manifest: {scene}") if row.get("status") != "index_read_ok": raise ValueError(f"Scene was not index-verified: {scene}") try: expected = int(row["zip_bytes"]) except (KeyError, TypeError, ValueError) as error: raise ValueError(f"Invalid expected archive size: {scene}") from error if expected <= 0: raise ValueError(f"Non-positive expected archive size: {scene}") scenes.append(scene) normalized.append({**row, "zip_bytes": expected}) return normalized, hashlib.sha256(source_bytes).hexdigest() def _check_archive_directory(archive_root: Path, scenes: set[str]) -> None: _assert_project_child(archive_root, PROJECT / "data", "Archive root") if _is_reparse(archive_root): raise ValueError(f"Archive root is a symlink or reparse point: {archive_root}") entries = list(archive_root.iterdir()) if archive_root.exists() else [] expected = {f"{scene}.zip" for scene in scenes} actual = {entry.name for entry in entries} if actual != expected: extra = sorted(actual - expected) missing = sorted(expected - actual) raise ValueError(f"Archive directory mismatch; extra={extra}, missing={missing}") for entry in entries: if not entry.is_file() or _is_reparse(entry): raise ValueError(f"Archive entry is not a regular file: {entry}") def _inspect_archive( scene: str, archive_path: Path, expected_bytes: int, target_root: Path ) -> ArchivePlan: actual_bytes = archive_path.stat().st_size if actual_bytes != expected_bytes: raise ValueError( f"Archive byte count mismatch for {scene}: {actual_bytes} != {expected_bytes}" ) try: with zipfile.ZipFile(archive_path) as archive: infos = tuple(archive.infolist()) except (OSError, zipfile.BadZipFile) as error: raise ValueError(f"Unreadable ZIP archive: {archive_path}") from error if not infos: raise ValueError(f"Empty ZIP archive: {archive_path}") names = [info.filename for info in infos] if len(set(names)) != len(names): raise ValueError(f"Duplicate ZIP member in {archive_path}") top_level: set[str] = set() uncompressed = 0 for info in infos: if _member_is_symlink(info): raise ValueError(f"ZIP symlink member is forbidden: {scene}/{info.filename}") parts = _safe_member_parts(info.filename) top_level.add(parts[0]) _member_target(target_root, info) uncompressed += info.file_size if top_level != {scene}: raise ValueError(f"ZIP top-level scene mismatch for {scene}: {sorted(top_level)}") return ArchivePlan( scene=scene, path=archive_path, expected_bytes=expected_bytes, archive_bytes=actual_bytes, infos=infos, member_count=len(infos), uncompressed_bytes=uncompressed, top_level=tuple(sorted(top_level)), ) def _existing_matches(path: Path, info: zipfile.ZipInfo, target_root: Path) -> bool: if not _lexists(path): return False _assert_safe_chain(target_root, path) if not path.is_file(): raise ValueError(f"Expected regular file for ZIP member: {path}") if path.stat().st_size != info.file_size: return False crc = 0 size = 0 with path.open("rb") as handle: while True: chunk = handle.read(COPY_CHUNK_BYTES) if not chunk: break size += len(chunk) crc = zlib.crc32(chunk, crc) return size == info.file_size and (crc & 0xFFFFFFFF) == info.CRC def _extract_member( archive: zipfile.ZipFile, info: zipfile.ZipInfo, target_root: Path, scene: str, ) -> int: target = _member_target(target_root, info) if info.is_dir(): _ensure_directory(target, target_root) return 0 _ensure_directory(target.parent, target_root) temporary = target.with_name(f".{target.name}.mcss-{scene}.part") _assert_safe_chain(target_root, temporary) if _lexists(temporary): if _is_reparse(temporary): raise ValueError( "Temporary extraction path is a symlink or reparse point: " f"{temporary}" ) if not temporary.is_file(): raise ValueError(f"Temporary extraction path is not a file: {temporary}") temporary.unlink() crc = 0 written = 0 try: with archive.open(info, "r") as source, temporary.open("xb") as destination: while True: chunk = source.read(COPY_CHUNK_BYTES) if not chunk: break destination.write(chunk) written += len(chunk) crc = zlib.crc32(chunk, crc) destination.flush() os.fsync(destination.fileno()) if written != info.file_size or (crc & 0xFFFFFFFF) != info.CRC: raise ValueError( f"CRC/size mismatch for {scene}/{info.filename}: " f"size={written}/{info.file_size}, crc={crc & 0xFFFFFFFF}/{info.CRC}" ) _assert_safe_chain(target_root, target) os.replace(temporary, target) return written except Exception: if _lexists(temporary) and not _is_reparse(temporary): temporary.unlink() raise def _atomic_json(path: Path, value: dict[str, Any]) -> None: if _is_reparse(path): raise ValueError(f"Output path is a symlink or reparse point: {path}") temporary = path.with_suffix(path.suffix + ".tmp") if _is_reparse(temporary): raise ValueError(f"Temporary output path is a symlink or reparse point: {temporary}") temporary.write_text(json.dumps(value, indent=2) + "\n", encoding="utf-8") os.replace(temporary, path) @contextmanager def _process_lock(path: Path) -> Iterator[None]: path.parent.mkdir(parents=True, exist_ok=True) handle = path.open("a+b") try: handle.seek(0, os.SEEK_END) if handle.tell() == 0: handle.write(b"0") handle.flush() handle.seek(0) try: acquire_file_lock(handle) except OSError as error: raise RuntimeError(f"Another extraction process holds the lock: {path}") from error try: yield finally: handle.seek(0) release_file_lock(handle) finally: handle.close() def _append_event(path: Path, event: dict[str, Any]) -> None: with path.open("a", encoding="utf-8", newline="\n") as handle: handle.write(json.dumps(event, separators=(",", ":")) + "\n") handle.flush() def _target_root(target: Path, scenes: set[str]) -> None: _assert_project_child(target, PROJECT / "data", "Target root") if _lexists(target): if _is_reparse(target) or not target.is_dir(): raise ValueError(f"Target root is not a normal directory: {target}") else: target.mkdir(parents=True, exist_ok=True) _assert_safe_chain(PROJECT / "data", target) unexpected = sorted(child.name for child in target.iterdir() if child.name not in scenes) if unexpected: raise ValueError(f"Target contains unexpected top-level entries: {unexpected}") def _estimate_required_bytes(plans: list[ArchivePlan], target_root: Path) -> int: required = 0 for plan in plans: for info in plan.infos: if info.is_dir(): continue target = _member_target(target_root, info) if not _existing_matches(target, info, target_root): required += info.file_size return required def _initial_state( plans: list[ArchivePlan], source: Path, source_sha256: str, target_root: Path, reserve_bytes: int, required_bytes: int, ) -> dict[str, Any]: return { "schema": "mcss.hypersim.holdout.extraction.v1", "status": "preflight", "pid": os.getpid(), "started_unix": time.time(), "updated_unix": time.time(), "source_manifest": str(source.resolve()), "source_manifest_sha256": source_sha256, "archive_root": str(ARCHIVE_ROOT.resolve()), "target_root": str(target_root.resolve()), "reserve_bytes": reserve_bytes, "required_bytes_after_resume": required_bytes, "expected_members": sum(plan.member_count for plan in plans), "expected_uncompressed_bytes": sum(plan.uncompressed_bytes for plan in plans), "extracted_members": 0, "skipped_members": 0, "extracted_bytes": 0, "skipped_bytes": 0, "extraction_performed": False, "scientific_evaluation_performed": False, "archives": [ { "scene": plan.scene, "archive_bytes": plan.archive_bytes, "expected_archive_bytes": plan.expected_bytes, "members": plan.member_count, "uncompressed_bytes": plan.uncompressed_bytes, "status": "queued", "processed_members": 0, "extracted_members": 0, "skipped_members": 0, "extracted_bytes": 0, "skipped_bytes": 0, } for plan in plans ], } def run( *, source: Path = SOURCE, archive_root: Path = ARCHIVE_ROOT, target_root: Path = TARGET_ROOT, log_root: Path = LOG_ROOT, reserve_bytes: int = RESERVE_BYTES, dry_run: bool = False, ) -> dict[str, Any]: rows, source_sha256 = _load_rows(source) scenes = {str(row["scene"]) for row in rows} _check_archive_directory(archive_root, scenes) _target_root(target_root, scenes) plans = [ _inspect_archive( str(row["scene"]), archive_root / f"{row['scene']}.zip", int(row["zip_bytes"]), target_root, ) for row in rows ] inventory = { "schema": "mcss.hypersim.holdout.extraction.inventory.v1", "source_manifest": str(source.resolve()), "source_manifest_sha256": source_sha256, "target_root": str(target_root.resolve()), "archives": [ { "scene": plan.scene, "archive_bytes": plan.archive_bytes, "expected_archive_bytes": plan.expected_bytes, "members": plan.member_count, "uncompressed_bytes": plan.uncompressed_bytes, "top_level": list(plan.top_level), } for plan in plans ], "total_members": sum(plan.member_count for plan in plans), "total_uncompressed_bytes": sum(plan.uncompressed_bytes for plan in plans), "extraction_performed": False, "scientific_evaluation_performed": False, } log_root_resolved = _assert_project_child(log_root, PROJECT / "outputs", "Log root") if _is_reparse(log_root_resolved): raise ValueError(f"Log root is a symlink or reparse point: {log_root_resolved}") log_root_resolved.mkdir(parents=True, exist_ok=True) _atomic_json(log_root_resolved / "inventory.json", inventory) required_bytes = _estimate_required_bytes(plans, target_root) free_bytes = shutil.disk_usage(target_root.parent).free state = _initial_state( plans, source, source_sha256, target_root, reserve_bytes, required_bytes, ) state["free_bytes_at_preflight"] = free_bytes state["space_sufficient"] = free_bytes >= required_bytes + reserve_bytes _atomic_json(log_root_resolved / "status.json", state) if not state["space_sufficient"]: state["status"] = "failed" state["error"] = ( f"Insufficient disk space: free={free_bytes}, " f"required={required_bytes}, reserve={reserve_bytes}" ) state["updated_unix"] = time.time() _atomic_json(log_root_resolved / "status.json", state) raise OSError(state["error"]) if dry_run: state["status"] = "dry_run" state["updated_unix"] = time.time() _atomic_json(log_root_resolved / "status.json", state) return state progress_path = log_root_resolved / "progress.jsonl" with _process_lock(log_root_resolved / "extract.lock"): state["status"] = "running" state["updated_unix"] = time.time() _atomic_json(log_root_resolved / "status.json", state) _append_event(progress_path, {"event": "start", "pid": os.getpid()}) last_status = time.monotonic() for archive_index, plan in enumerate(plans): archive_state = state["archives"][archive_index] archive_state["status"] = "running" _append_event( progress_path, { "event": "archive_start", "scene": plan.scene, "members": plan.member_count, "uncompressed_bytes": plan.uncompressed_bytes, }, ) try: with zipfile.ZipFile(plan.path) as archive: for info in plan.infos: target = _member_target(target_root, info) archive_state["processed_members"] += 1 if info.is_dir(): _ensure_directory(target, target_root) elif _existing_matches(target, info, target_root): archive_state["skipped_members"] += 1 archive_state["skipped_bytes"] += info.file_size state["skipped_members"] += 1 state["skipped_bytes"] += info.file_size else: written = _extract_member(archive, info, target_root, plan.scene) archive_state["extracted_members"] += 1 archive_state["extracted_bytes"] += written state["extracted_members"] += 1 state["extracted_bytes"] += written if time.monotonic() - last_status >= 5: state["updated_unix"] = time.time() _atomic_json(log_root_resolved / "status.json", state) last_status = time.monotonic() except Exception as error: archive_state["status"] = "failed" archive_state["error"] = str(error) state["status"] = "failed" state["error"] = f"{plan.scene}: {error}" state["updated_unix"] = time.time() _atomic_json(log_root_resolved / "status.json", state) _append_event( progress_path, {"event": "archive_failed", "scene": plan.scene, "error": str(error)}, ) raise archive_state["status"] = "complete" _append_event(progress_path, {"event": "archive_complete", "scene": plan.scene}) state["updated_unix"] = time.time() _atomic_json(log_root_resolved / "status.json", state) expected_scenes = scenes actual_scenes = {child.name for child in target_root.iterdir() if child.is_dir()} if actual_scenes != expected_scenes: raise ValueError( f"Extracted scene roots mismatch; extra={sorted(actual_scenes - expected_scenes)}, " f"missing={sorted(expected_scenes - actual_scenes)}" ) state["status"] = "complete" state["extraction_performed"] = True state["updated_unix"] = time.time() _atomic_json(log_root_resolved / "status.json", state) receipt = { "schema": "mcss.hypersim.holdout.extraction.receipt.v1", "status": "complete", "source_manifest": str(source.resolve()), "source_manifest_sha256": source_sha256, "archive_root": str(archive_root.resolve()), "target_root": str(target_root.resolve()), "scene_count": len(plans), "member_count": state["expected_members"], "uncompressed_bytes": state["expected_uncompressed_bytes"], "extracted_members": state["extracted_members"], "skipped_members": state["skipped_members"], "extracted_bytes": state["extracted_bytes"], "skipped_bytes": state["skipped_bytes"], "crc_verified": True, "extraction_performed": True, "scientific_evaluation_performed": False, "archives": state["archives"], } _atomic_json(log_root_resolved / "receipt.json", receipt) _append_event(progress_path, {"event": "complete", "scene_count": len(plans)}) return receipt def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--source", type=Path, default=SOURCE) parser.add_argument("--archive-root", type=Path, default=ARCHIVE_ROOT) parser.add_argument("--target-root", type=Path, default=TARGET_ROOT) parser.add_argument("--log-root", type=Path, default=LOG_ROOT) parser.add_argument("--reserve-gib", type=float, default=10.0) parser.add_argument("--dry-run", action="store_true") args = parser.parse_args() try: result = run( source=args.source, archive_root=args.archive_root, target_root=args.target_root, log_root=args.log_root, reserve_bytes=int(args.reserve_gib * 1024**3), dry_run=args.dry_run, ) except Exception as error: print(json.dumps({"status": "failed", "error": str(error)}), flush=True) return 1 print( json.dumps({"status": result["status"], "target_root": result["target_root"]}), flush=True, ) return 0 if __name__ == "__main__": raise SystemExit(main())