from __future__ import annotations import hashlib import importlib import json import os import re import secrets import stat import threading import time from pathlib import Path from typing import IO, Any, Literal, cast from k1link.artifacts import utc_now_iso CameraArchiveKind = Literal["init", "media"] CameraArchiveStatus = Literal["complete", "interrupted", "failed"] CAMERA_ARCHIVE_SCHEMA = "missioncore.camera-recording/v1" CAMERA_INDEX_SCHEMA = "missioncore.camera-recording-index/v1" CAMERA_COMMIT_POLICY = "per-segment-fsync" _SAFE_COMPONENT = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$") _EPOCH_DIRECTORY = re.compile(r"^epoch-([1-9][0-9]*)$") _SEGMENT_FILE = re.compile(r"^([1-9][0-9]*)\.m4s$") _DEFAULT_COMMIT_INTERVAL_SECONDS = 0.25 _DEFAULT_COMMIT_BYTES = 4 * 1024 * 1024 _MAX_RECOVERY_SUMMARY_BYTES = 2 * 1024 * 1024 _MAX_RECOVERY_INDEX_LINE_BYTES = 64 * 1024 _MAX_RECOVERY_SEGMENT_BYTES = 8 * 1024 * 1024 _ACTIVE_ARCHIVES_LOCK = threading.Lock() _ACTIVE_ARCHIVES: set[Path] = set() class CameraArchiveError(RuntimeError): """A durable camera artifact could not be committed or recovered.""" class CameraArchiveWriter: """Write a canonical, seekable fMP4 camera epoch. ``init.mp4`` and every ``segments/N.m4s`` file are individually fsynced and atomically published before the matching JSONL index record. The index and an interrupted checkpoint summary are then fsynced before :meth:`append` returns. This deliberately uses a per-segment durability boundary: the old group-commit tuning arguments remain source-compatible but never weaken the segment-bound RPO. A process crash can interrupt the small multi-file commit window. The module-level :func:`recover_incomplete_camera_archives` reconciles that window from the independently durable fragments without discarding bytes. """ def __init__( self, session_dir: Path, source_id: str, generation: int, *, commit_interval_seconds: float = _DEFAULT_COMMIT_INTERVAL_SECONDS, commit_bytes: int = _DEFAULT_COMMIT_BYTES, ) -> None: if not _SAFE_COMPONENT.fullmatch(source_id): raise ValueError("camera source id is not a safe storage identifier") if generation < 1: raise ValueError("camera generation must be positive") if commit_interval_seconds < 0: raise ValueError("camera commit interval must be non-negative") if commit_bytes < 1: raise ValueError("camera commit byte threshold must be positive") root = session_dir.expanduser().resolve() if not root.is_dir(): raise ValueError("observation session directory does not exist") media_root = root / "media" source_dir = media_root / source_id archive_dir = source_dir / f"epoch-{generation}" root_fd = _open_directory_fd(root) if root_fd is None: raise CameraArchiveError("camera session root failed no-follow validation") media_fd: int | None = None source_fd: int | None = None archive_fd: int | None = None segments_fd: int | None = None index_stream: IO[bytes] | None = None storage_ready = False try: media_fd = _create_or_open_private_directory_at(root_fd, "media") source_fd = _create_or_open_private_directory_at(media_fd, source_id) archive_fd = _create_or_open_private_directory_at( source_fd, f"epoch-{generation}", exclusive=True, ) segments_fd = _create_or_open_private_directory_at( archive_fd, "segments", exclusive=True, ) index_stream = _open_private_binary_at(archive_fd, "index.jsonl") storage_ready = True finally: for descriptor in (source_fd, media_fd, root_fd): if descriptor is not None: os.close(descriptor) if not storage_ready: if index_stream is not None: index_stream.close() for descriptor in (segments_fd, archive_fd): if descriptor is not None: os.close(descriptor) if index_stream is None or archive_fd is None or segments_fd is None: raise CameraArchiveError("camera index could not be created") self.session_dir = root self.source_id = source_id self.generation = generation self.archive_dir = archive_dir self.init_path = self.archive_dir / "init.mp4" self.segments_dir = self.archive_dir / "segments" self.index_path = self.archive_dir / "index.jsonl" self.summary_path = self.archive_dir / "summary.json" self._archive_fd = archive_fd self._segments_fd = segments_fd self._index = index_stream self._lock = threading.Lock() self._sequence = 0 self._committed_media_segments = 0 self._committed_bytes = 0 self._init_written = False self._closed = False self._closed_summary: dict[str, Any] | None = None self._failed = False self._started_at_utc = utc_now_iso() self._started_monotonic_ns = time.monotonic_ns() self._stream_sha256 = hashlib.sha256() self._index_sha256 = hashlib.sha256() self._init_sha256: str | None = None self._legacy_commit_interval_seconds = commit_interval_seconds self._legacy_commit_bytes = commit_bytes with _ACTIVE_ARCHIVES_LOCK: _ACTIVE_ARCHIVES.add(self.archive_dir) def append( self, kind: CameraArchiveKind, payload: bytes, *, host_epoch_ns: int | None = None, host_monotonic_ns: int | None = None, ) -> dict[str, Any]: if kind not in {"init", "media"}: raise ValueError("camera archive kind must be init or media") if not payload: raise ValueError("camera archive payload must not be empty") epoch_ns = host_epoch_ns if host_epoch_ns is not None else time.time_ns() monotonic_ns = ( host_monotonic_ns if host_monotonic_ns is not None else time.monotonic_ns() ) if epoch_ns < 0 or monotonic_ns < 0: raise ValueError("camera archive timestamps must be non-negative") with self._lock: self._require_open_locked() if kind == "init": return self._append_init_locked(payload, epoch_ns, monotonic_ns) if not self._init_written: raise CameraArchiveError("camera media arrived before its init segment") return self._append_media_locked(payload, epoch_ns, monotonic_ns) def flush(self) -> None: """Reassert the current segment-bound checkpoint durability.""" with self._lock: self._require_open_locked() try: self._index.flush() os.fsync(self._index.fileno()) self._write_summary_locked( status="interrupted", failure_code="process-crash-before-clean-close", ) except (OSError, CameraArchiveError) as exc: self._failed = True raise CameraArchiveError("camera archive durable commit failed") from exc def close( self, *, status: CameraArchiveStatus = "complete", failure_code: str | None = None, ) -> dict[str, Any]: with self._lock: if self._closed: if self._closed_summary is None: raise CameraArchiveError("camera archive close previously failed") return dict(self._closed_summary) effective_status: CameraArchiveStatus = "failed" if self._failed else status close_error: OSError | None = None try: self._index.flush() os.fsync(self._index.fileno()) except OSError as exc: self._failed = True effective_status = "failed" close_error = exc finally: try: self._index.close() except OSError as exc: close_error = close_error or exc summary = self._summary_locked( status=effective_status, failure_code=( failure_code or ("storage-commit-failed" if close_error is not None else None) ), ) try: _write_json_atomic_at(self._archive_fd, "summary.json", summary) except (OSError, CameraArchiveError) as exc: self._failed = True raise CameraArchiveError("camera archive summary commit failed") from exc finally: self._closed = True os.close(self._segments_fd) os.close(self._archive_fd) with _ACTIVE_ARCHIVES_LOCK: _ACTIVE_ARCHIVES.discard(self.archive_dir) self._closed_summary = dict(summary) if close_error is not None: raise CameraArchiveError("camera archive final commit failed") from close_error return summary def _append_init_locked( self, payload: bytes, epoch_ns: int, monotonic_ns: int, ) -> dict[str, Any]: if self._init_written: raise CameraArchiveError("camera archive already contains an init segment") digest = hashlib.sha256(payload).hexdigest() try: _write_bytes_atomic_at(self._archive_fd, "init.mp4", payload) self._init_written = True self._init_sha256 = digest self._committed_bytes = len(payload) self._stream_sha256.update(payload) self._write_summary_locked( status="interrupted", failure_code="process-crash-before-clean-close", ) except (OSError, CameraArchiveError) as exc: self._failed = True raise CameraArchiveError("camera archive durable commit failed") from exc return { "schema_version": CAMERA_INDEX_SCHEMA, "sequence": 0, "kind": "init", "path": "init.mp4", "length": len(payload), "sha256": digest, "host_epoch_ns": epoch_ns, "host_monotonic_ns": monotonic_ns, "session_monotonic_ns": max(0, monotonic_ns - self._started_monotonic_ns), } def _append_media_locked( self, payload: bytes, epoch_ns: int, monotonic_ns: int, ) -> dict[str, Any]: sequence = self._sequence + 1 relative_path = f"segments/{sequence}.m4s" segment_name = f"{sequence}.m4s" digest = hashlib.sha256(payload).hexdigest() entry = { "schema_version": CAMERA_INDEX_SCHEMA, "sequence": sequence, "kind": "media", "path": relative_path, "length": len(payload), "sha256": digest, "host_epoch_ns": epoch_ns, "host_monotonic_ns": monotonic_ns, "session_monotonic_ns": max(0, monotonic_ns - self._started_monotonic_ns), } encoded = _encode_index_entry(entry) index_offset = self._index.tell() published_segment = False try: _write_bytes_atomic_at(self._segments_fd, segment_name, payload) published_segment = True self._index.write(encoded) self._index.flush() os.fsync(self._index.fileno()) except (OSError, CameraArchiveError) as exc: self._failed = True # Best-effort rollback keeps the last fully checkpointed prefix # structurally valid. Recovery handles a crash before this block. with _ignore_os_error(): self._index.seek(index_offset) self._index.truncate() self._index.flush() os.fsync(self._index.fileno()) if published_segment: with _ignore_os_error(): os.unlink(segment_name, dir_fd=self._segments_fd) os.fsync(self._segments_fd) raise CameraArchiveError("camera archive durable commit failed") from exc self._sequence = sequence self._committed_media_segments += 1 self._committed_bytes += len(payload) self._stream_sha256.update(payload) self._index_sha256.update(encoded) try: self._write_summary_locked( status="interrupted", failure_code="process-crash-before-clean-close", ) except (OSError, CameraArchiveError) as exc: self._failed = True raise CameraArchiveError("camera archive durable checkpoint failed") from exc return entry def _write_summary_locked( self, *, status: CameraArchiveStatus, failure_code: str | None, ) -> None: _write_json_atomic_at( self._archive_fd, "summary.json", self._summary_locked(status=status, failure_code=failure_code), ) def _summary_locked( self, *, status: CameraArchiveStatus, failure_code: str | None, ) -> dict[str, Any]: return { "schema_version": CAMERA_ARCHIVE_SCHEMA, "source_id": self.source_id, "codec_epoch": self.generation, "status": status, "started_at_utc": self._started_at_utc, "completed_at_utc": utc_now_iso(), "segment_count": self._committed_media_segments, "entry_count": self._committed_media_segments, "media_segment_count": self._committed_media_segments, "valid_bytes": self._committed_bytes, "init_sha256": self._init_sha256, "stream_sha256": self._stream_sha256.hexdigest(), "index_sha256": self._index_sha256.hexdigest(), "synchronization": "host-arrival-best-effort", "commit_policy": CAMERA_COMMIT_POLICY, "failure_code": failure_code, "artifacts": { "init": "init.mp4", "segments": "segments", "index": "index.jsonl", }, } def _require_open_locked(self) -> None: if self._closed: raise CameraArchiveError("camera archive is already closed") if self._failed: raise CameraArchiveError("camera archive is in a failed state") def recover_incomplete_camera_archives( sessions_root: Path, ) -> tuple[dict[str, Any], ...]: """Seal incomplete camera epochs beneath a trusted sessions root. The scan is confined to direct, safe session children and canonical ``media//epoch-N`` trees. In-process active writers are skipped. Complete, structurally valid summaries are idempotent no-ops. A no-follow advisory lease at ``sessions_root/.camera-recovery.lock`` serializes worker startup recovery across processes. For an incomplete epoch, independently durable, contiguous ``N.m4s`` fragments are re-indexed; non-contiguous/orphan evidence is moved (never deleted) to ``recovery-orphans``; and an ``interrupted`` summary is atomically written. This is intentionally a startup/catalog-refresh operation, not a hot-path operation. A clean sealed epoch is validated from its summary, streaming index and segment stat metadata without loading payloads; only an incomplete or damaged epoch enters fragment recovery and hashes candidate payloads. The callable is safe to repeat, but a composition layer should normally execute it once per server process before the first catalog import. """ root = sessions_root.expanduser().resolve() if not root.is_dir(): return () root_fd = _open_directory_fd(root) if root_fd is None: raise CameraArchiveError("camera sessions root failed no-follow validation") lock_fd: int | None = None try: lock_fd = _open_recovery_lock_at(root_fd) _lock_recovery_file(lock_fd) recovered: list[dict[str, Any]] = [] for session_path in sorted(root.iterdir()): if not session_path.is_dir() or not _SAFE_COMPONENT.fullmatch(session_path.name): continue session = _resolved_confined_directory(session_path, root) if session is None: continue media_root = _resolved_confined_directory(session / "media", session) if media_root is None: continue for source_path in sorted(media_root.iterdir()): if not source_path.is_dir() or not _SAFE_COMPONENT.fullmatch(source_path.name): continue source = _resolved_confined_directory(source_path, media_root) if source is None: continue for epoch_path in sorted(source.iterdir()): match = _EPOCH_DIRECTORY.fullmatch(epoch_path.name) if not epoch_path.is_dir() or match is None: continue epoch = _resolved_confined_directory(epoch_path, source) if epoch is None: continue with _ACTIVE_ARCHIVES_LOCK: active = epoch in _ACTIVE_ARCHIVES if active: continue summary = _recover_epoch( epoch, source_id=source.name, generation=int(match.group(1)), ) if summary is not None: recovered.append(summary) return tuple(recovered) finally: if lock_fd is not None: _unlock_recovery_file(lock_fd) os.close(lock_fd) os.close(root_fd) def _recover_epoch( epoch: Path, *, source_id: str, generation: int, ) -> dict[str, Any] | None: epoch_fd = _open_directory_fd(epoch) if epoch_fd is None: return None try: init = _read_regular_at(epoch_fd, "init.mp4", _MAX_RECOVERY_SEGMENT_BYTES) if not init: return None segments_fd = _open_directory_at(epoch_fd, "segments") if segments_fd is None: return None try: if _sealed_epoch_is_valid_on_disk( epoch_fd, segments_fd, source_id=source_id, init=init, ): return None old_index = _read_regular_at_current_size( epoch_fd, "index.jsonl", allow_empty=True, ) old_summary = _read_regular_at( epoch_fd, "summary.json", _MAX_RECOVERY_SUMMARY_BYTES, allow_empty=False, ) segment_timestamps, orphans = _read_recovery_segment_catalog(segments_fd) old_entries = _parse_index_prefix(old_index) old_by_sequence = { int(entry["sequence"]): entry for entry in old_entries if isinstance(entry.get("sequence"), int) } recovered_entries: list[dict[str, Any]] = [] stream_hash = hashlib.sha256(init) valid_bytes = len(init) expected = 1 while True: if expected not in segment_timestamps: break payload = _read_regular_at( segments_fd, f"{expected}.m4s", _MAX_RECOVERY_SEGMENT_BYTES, allow_empty=False, ) if payload is None: orphans.append(f"{expected}.m4s") break digest = hashlib.sha256(payload).hexdigest() previous = old_by_sequence.get(expected) if _index_entry_matches(previous, expected, len(payload), digest): entry = dict(cast(dict[str, Any], previous)) else: entry = { "schema_version": CAMERA_INDEX_SCHEMA, "sequence": expected, "kind": "media", "path": f"segments/{expected}.m4s", "length": len(payload), "sha256": digest, "host_epoch_ns": segment_timestamps[expected], "host_monotonic_ns": None, "session_monotonic_ns": None, "recovered": True, } recovered_entries.append(entry) stream_hash.update(payload) valid_bytes += len(payload) expected += 1 orphans.extend( f"{sequence}.m4s" for sequence in segment_timestamps if sequence >= expected ) orphans = sorted(set(orphans)) if not recovered_entries: return None encoded_index = b"".join( _encode_index_entry(entry) for entry in recovered_entries ) needs_recovery_dir = bool( orphans or (old_index is not None and old_index != encoded_index) or old_summary ) recovery_fd: int | None = None if needs_recovery_dir: recovery_fd = _ensure_recovery_directory_at(epoch_fd) try: if ( recovery_fd is not None and old_index is not None and old_index != encoded_index ): _preserve_recovery_artifact_at( recovery_fd, "index.pre-recovery.jsonl", old_index, ) if recovery_fd is not None and old_summary: _preserve_recovery_artifact_at( recovery_fd, "summary.pre-recovery.json", old_summary, ) if recovery_fd is not None: for orphan in orphans: _quarantine_orphan_at(segments_fd, recovery_fd, orphan) _write_bytes_atomic_at(epoch_fd, "index.jsonl", encoded_index) started_at = _summary_timestamp_bytes( old_summary, "started_at_utc", ) or utc_now_iso() index_hash = hashlib.sha256(encoded_index).hexdigest() summary = { "schema_version": CAMERA_ARCHIVE_SCHEMA, "source_id": source_id, "codec_epoch": generation, "status": "interrupted", "started_at_utc": started_at, "completed_at_utc": utc_now_iso(), "segment_count": len(recovered_entries), "entry_count": len(recovered_entries), "media_segment_count": len(recovered_entries), "valid_bytes": valid_bytes, "init_sha256": hashlib.sha256(init).hexdigest(), "stream_sha256": stream_hash.hexdigest(), "index_sha256": index_hash, "synchronization": "host-arrival-best-effort", "commit_policy": CAMERA_COMMIT_POLICY, "failure_code": "server-process-interrupted", "recovered_at_utc": utc_now_iso(), "artifacts": { "init": "init.mp4", "segments": "segments", "index": "index.jsonl", }, } _write_json_atomic_at(epoch_fd, "summary.json", summary) return summary finally: if recovery_fd is not None: os.close(recovery_fd) finally: os.close(segments_fd) finally: os.close(epoch_fd) def _sealed_epoch_is_valid_on_disk( epoch_fd: int, segments_fd: int, *, source_id: str, init: bytes, ) -> bool: """Recognize a clean seal without loading a multi-hour archive into RAM. Recovery only needs to prove that the durable commit envelope is complete. Full fragment digests and ISO-BMFF timing are revalidated by the recorded media preparation path before browser publication. """ summary_bytes = _read_regular_at( epoch_fd, "summary.json", _MAX_RECOVERY_SUMMARY_BYTES, allow_empty=False, ) if not summary_bytes: return False try: summary = json.loads(summary_bytes) except (UnicodeDecodeError, json.JSONDecodeError): return False segment_count = summary.get("segment_count") if isinstance(summary, dict) else None if ( not isinstance(summary, dict) or summary.get("schema_version") != CAMERA_ARCHIVE_SCHEMA or summary.get("source_id") != source_id or not isinstance(segment_count, int) or isinstance(segment_count, bool) or segment_count < 1 or summary.get("entry_count") != segment_count or summary.get("media_segment_count") != segment_count or summary.get("commit_policy") != CAMERA_COMMIT_POLICY or summary.get("init_sha256") != hashlib.sha256(init).hexdigest() ): return False try: descriptor = os.open( "index.jsonl", os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0), dir_fd=epoch_fd, ) except OSError: return False digest = hashlib.sha256() valid_bytes = len(init) try: before = os.fstat(descriptor) if not stat.S_ISREG(before.st_mode) or before.st_size <= 0: return False with os.fdopen(descriptor, "rb") as stream: descriptor = -1 for sequence in range(1, segment_count + 1): raw_line = stream.readline(_MAX_RECOVERY_INDEX_LINE_BYTES + 1) if ( not raw_line or len(raw_line) > _MAX_RECOVERY_INDEX_LINE_BYTES or not raw_line.endswith(b"\n") ): return False digest.update(raw_line) try: entry = json.loads(raw_line) except (UnicodeDecodeError, json.JSONDecodeError): return False length = entry.get("length") if isinstance(entry, dict) else None if ( not isinstance(length, int) or isinstance(length, bool) or not 0 < length <= _MAX_RECOVERY_SEGMENT_BYTES or not _index_entry_shape_matches(entry, sequence) ): return False try: segment_stat = os.stat( f"{sequence}.m4s", dir_fd=segments_fd, follow_symlinks=False, ) except OSError: return False if not stat.S_ISREG(segment_stat.st_mode) or segment_stat.st_size != length: return False valid_bytes += length if stream.read(1): return False after = os.fstat(stream.fileno()) if ( (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns) != (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns) ): return False except OSError: return False finally: if descriptor >= 0: os.close(descriptor) try: segment_names = [ name for name in os.listdir(segments_fd) if _SEGMENT_FILE.fullmatch(name) is not None ] except OSError: return False if len(segment_names) != segment_count or any( name != f"{sequence}.m4s" for sequence, name in enumerate( sorted(segment_names, key=lambda name: int(name.removesuffix(".m4s"))), start=1, ) ): return False return ( summary.get("index_sha256") == digest.hexdigest() and summary.get("valid_bytes") == valid_bytes ) def _index_entry_shape_matches(entry: object, sequence: int) -> bool: return ( isinstance(entry, dict) and entry.get("schema_version") == CAMERA_INDEX_SCHEMA and entry.get("sequence") == sequence and entry.get("kind") == "media" and entry.get("path") == f"segments/{sequence}.m4s" and isinstance(entry.get("sha256"), str) and re.fullmatch(r"[a-f0-9]{64}", str(entry["sha256"])) is not None ) def _read_recovery_segment_catalog( segments_fd: int, ) -> tuple[dict[int, int], list[str]]: timestamps: dict[int, int] = {} orphans: list[str] = [] try: names = sorted(os.listdir(segments_fd)) except OSError as exc: raise CameraArchiveError("camera segment directory cannot be enumerated") from exc for name in names: match = _SEGMENT_FILE.fullmatch(name) if match is None: continue sequence = int(match.group(1)) if name != f"{sequence}.m4s" or sequence in timestamps: orphans.append(name) continue try: metadata = os.stat(name, dir_fd=segments_fd, follow_symlinks=False) except OSError: orphans.append(name) continue if ( not stat.S_ISREG(metadata.st_mode) or not 0 < metadata.st_size <= _MAX_RECOVERY_SEGMENT_BYTES ): orphans.append(name) continue timestamps[sequence] = metadata.st_mtime_ns return timestamps, orphans def _parse_index_prefix(payload: bytes | None) -> list[dict[str, Any]]: if not payload: return [] result: list[dict[str, Any]] = [] for raw_line in payload.splitlines(keepends=True): if not raw_line.endswith(b"\n"): break try: value = json.loads(raw_line) except (UnicodeDecodeError, json.JSONDecodeError): break if not isinstance(value, dict): break sequence = value.get("sequence") if ( not isinstance(sequence, int) or isinstance(sequence, bool) or sequence != len(result) + 1 ): break result.append(value) return result def _index_entry_matches( entry: object, sequence: int, length: int, digest: str, ) -> bool: return ( isinstance(entry, dict) and entry.get("sequence") == sequence and entry.get("kind") == "media" and entry.get("path") == f"segments/{sequence}.m4s" and entry.get("length") == length and entry.get("sha256") == digest ) def _encode_index_entry(entry: dict[str, Any]) -> bytes: return ( json.dumps(entry, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n" ).encode("utf-8") def _open_private_binary_at(parent_fd: int, name: str) -> IO[bytes]: if "/" in name or name in {"", ".", ".."}: raise CameraArchiveError("camera artifact has an unsafe name") descriptor = os.open( name, os.O_RDWR | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0), 0o600, dir_fd=parent_fd, ) return os.fdopen(descriptor, "w+b") def _resolved_confined_directory(path: Path, parent: Path) -> Path | None: try: metadata = path.lstat() if not stat.S_ISDIR(metadata.st_mode): return None resolved = path.resolve(strict=True) except OSError: return None return resolved if resolved.is_dir() and resolved.is_relative_to(parent) else None def _open_directory_fd(path: Path) -> int | None: try: before = path.lstat() if not stat.S_ISDIR(before.st_mode): return None descriptor = os.open( path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0), ) except OSError: return None opened = os.fstat(descriptor) if (opened.st_dev, opened.st_ino) != (before.st_dev, before.st_ino): os.close(descriptor) return None return descriptor def _open_directory_at(parent_fd: int, name: str) -> int | None: if "/" in name or name in {"", ".", ".."}: return None try: before = os.stat(name, dir_fd=parent_fd, follow_symlinks=False) if not stat.S_ISDIR(before.st_mode): return None descriptor = os.open( name, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0), dir_fd=parent_fd, ) except OSError: return None opened = os.fstat(descriptor) if (opened.st_dev, opened.st_ino) != (before.st_dev, before.st_ino): os.close(descriptor) return None return descriptor def _create_or_open_private_directory_at( parent_fd: int, name: str, *, exclusive: bool = False, ) -> int: if "/" in name or name in {"", ".", ".."}: raise CameraArchiveError("camera storage directory has an unsafe name") try: os.mkdir(name, 0o700, dir_fd=parent_fd) os.fsync(parent_fd) except FileExistsError: if exclusive: raise except OSError as exc: raise CameraArchiveError("camera storage directory cannot be created") from exc descriptor = _open_directory_at(parent_fd, name) if descriptor is None: raise CameraArchiveError("camera storage directory failed no-follow validation") try: fchmod = getattr(os, "fchmod", None) if callable(fchmod): fchmod(descriptor, 0o700) return descriptor except OSError as exc: os.close(descriptor) raise CameraArchiveError("camera storage permissions cannot be enforced") from exc def _open_recovery_lock_at(root_fd: int) -> int: name = ".camera-recovery.lock" descriptor: int | None = None try: try: before = os.stat(name, dir_fd=root_fd, follow_symlinks=False) except FileNotFoundError: before = None if before is not None and not stat.S_ISREG(before.st_mode): raise OSError("recovery lock is not a regular file") descriptor = os.open( name, os.O_RDWR | os.O_CREAT | getattr(os, "O_NOFOLLOW", 0), 0o600, dir_fd=root_fd, ) metadata = os.fstat(descriptor) after = os.stat(name, dir_fd=root_fd, follow_symlinks=False) if ( not stat.S_ISREG(metadata.st_mode) or not stat.S_ISREG(after.st_mode) or (metadata.st_dev, metadata.st_ino) != (after.st_dev, after.st_ino) or ( before is not None and (metadata.st_dev, metadata.st_ino) != (before.st_dev, before.st_ino) ) ): raise OSError("recovery lock is not a regular file") fchmod = getattr(os, "fchmod", None) if callable(fchmod): fchmod(descriptor, 0o600) if metadata.st_size == 0: os.write(descriptor, b"\0") os.fsync(descriptor) return descriptor except OSError as exc: if descriptor is not None: os.close(descriptor) raise CameraArchiveError("camera recovery lock failed no-follow validation") from exc def _lock_recovery_file(descriptor: int) -> None: try: if os.name == "posix": fcntl = importlib.import_module("fcntl") fcntl.flock(descriptor, fcntl.LOCK_EX) return msvcrt = importlib.import_module("msvcrt") os.lseek(descriptor, 0, os.SEEK_SET) msvcrt.locking(descriptor, msvcrt.LK_LOCK, 1) except (ImportError, OSError) as exc: raise CameraArchiveError("camera recovery lease could not be acquired") from exc def _unlock_recovery_file(descriptor: int) -> None: try: if os.name == "posix": fcntl = importlib.import_module("fcntl") fcntl.flock(descriptor, fcntl.LOCK_UN) return msvcrt = importlib.import_module("msvcrt") os.lseek(descriptor, 0, os.SEEK_SET) msvcrt.locking(descriptor, msvcrt.LK_UNLCK, 1) except (ImportError, OSError): # Closing the descriptor below releases an OS advisory lease even if an # explicit unlock fails during shutdown. return def _read_regular_at( parent_fd: int, name: str, limit: int, *, allow_empty: bool = False, ) -> bytes | None: result = _read_regular_with_metadata_at( parent_fd, name, limit, allow_empty=allow_empty, ) return result[0] if result is not None else None def _read_regular_at_current_size( parent_fd: int, name: str, *, allow_empty: bool, ) -> bytes | None: """Read one private recovery artifact without a duration-derived ceiling.""" try: metadata = os.stat(name, dir_fd=parent_fd, follow_symlinks=False) except OSError: return None if not stat.S_ISREG(metadata.st_mode): return None return _read_regular_at( parent_fd, name, max(1, metadata.st_size), allow_empty=allow_empty, ) def _read_regular_with_metadata_at( parent_fd: int, name: str, limit: int, *, allow_empty: bool = False, ) -> tuple[bytes, os.stat_result] | None: if "/" in name or name in {"", ".", ".."}: return None try: before = os.stat(name, dir_fd=parent_fd, follow_symlinks=False) if ( not stat.S_ISREG(before.st_mode) or before.st_size > limit or (before.st_size == 0 and not allow_empty) ): return None descriptor = os.open( name, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0), dir_fd=parent_fd, ) except OSError: return None try: opened = os.fstat(descriptor) if ( not stat.S_ISREG(opened.st_mode) or (opened.st_dev, opened.st_ino) != (before.st_dev, before.st_ino) or opened.st_size > limit ): return None chunks: list[bytes] = [] remaining = opened.st_size while remaining: chunk = os.read(descriptor, min(1024 * 1024, remaining)) if not chunk: return None chunks.append(chunk) remaining -= len(chunk) payload = b"".join(chunks) after = os.stat(name, dir_fd=parent_fd, follow_symlinks=False) if ( (after.st_dev, after.st_ino) != (opened.st_dev, opened.st_ino) or after.st_size != len(payload) ): return None if not payload and not allow_empty: return None return payload, opened except OSError: return None finally: os.close(descriptor) def _ensure_recovery_directory_at(epoch_fd: int) -> int: name = "recovery-orphans" try: metadata = os.stat(name, dir_fd=epoch_fd, follow_symlinks=False) except FileNotFoundError: try: os.mkdir(name, 0o700, dir_fd=epoch_fd) os.fsync(epoch_fd) except OSError as exc: raise CameraArchiveError("camera recovery directory cannot be created") from exc except OSError as exc: raise CameraArchiveError("camera recovery directory cannot be inspected") from exc else: if not stat.S_ISDIR(metadata.st_mode): raise CameraArchiveError("camera recovery directory is not a real directory") descriptor = _open_directory_at(epoch_fd, name) if descriptor is None: raise CameraArchiveError("camera recovery directory failed no-follow validation") return descriptor def _preserve_recovery_artifact_at( recovery_fd: int, name: str, payload: bytes, ) -> None: if not payload: return destination = _unique_name_at(recovery_fd, name) _write_bytes_atomic_at(recovery_fd, destination, payload) def _quarantine_orphan_at( segments_fd: int, recovery_fd: int, name: str, ) -> None: if "/" in name or name in {"", ".", ".."}: raise CameraArchiveError("camera orphan has an unsafe name") destination = _unique_name_at(recovery_fd, name) try: # dir_fd-relative rename moves the directory entry itself. A malicious # symlink is quarantined without ever following or reading its target. os.rename( name, destination, src_dir_fd=segments_fd, dst_dir_fd=recovery_fd, ) os.fsync(segments_fd) os.fsync(recovery_fd) except OSError as exc: raise CameraArchiveError("camera orphan cannot be quarantined safely") from exc def _unique_name_at(directory_fd: int, base: str) -> str: candidate = base suffix = 1 while True: try: os.stat(candidate, dir_fd=directory_fd, follow_symlinks=False) except FileNotFoundError: return candidate except OSError as exc: raise CameraArchiveError("camera recovery destination cannot be inspected") from exc candidate = f"{base}.{suffix}" suffix += 1 def _write_bytes_atomic_at(directory_fd: int, name: str, payload: bytes) -> None: if "/" in name or name in {"", ".", ".."}: raise CameraArchiveError("camera recovery artifact has an unsafe name") temp_name = f".{name}.{secrets.token_hex(8)}.tmp" descriptor: int | None = None try: descriptor = os.open( temp_name, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600, dir_fd=directory_fd, ) view = memoryview(payload) while view: written = os.write(descriptor, view) if written <= 0: raise OSError("short write") view = view[written:] os.fsync(descriptor) os.close(descriptor) descriptor = None os.replace( temp_name, name, src_dir_fd=directory_fd, dst_dir_fd=directory_fd, ) os.fsync(directory_fd) except OSError as exc: raise CameraArchiveError("camera recovery artifact cannot be committed") from exc finally: if descriptor is not None: os.close(descriptor) try: os.unlink(temp_name, dir_fd=directory_fd) except FileNotFoundError: pass except OSError: pass def _write_json_atomic_at(directory_fd: int, name: str, payload: dict[str, Any]) -> None: encoded = (json.dumps(payload, ensure_ascii=False, indent=2) + "\n").encode("utf-8") _write_bytes_atomic_at(directory_fd, name, encoded) def _summary_timestamp_bytes(payload: bytes | None, field: str) -> str | None: if not payload: return None try: summary = json.loads(payload) except (UnicodeDecodeError, json.JSONDecodeError): return None value = summary.get(field) if isinstance(summary, dict) else None return value if isinstance(value, str) and len(value) <= 64 else None class _ignore_os_error: def __enter__(self) -> None: return None def __exit__(self, exc_type: object, exc: object, traceback: object) -> bool: return isinstance(exc, OSError)