from __future__ import annotations import hashlib import json import math import os import re import secrets import stat import threading from collections.abc import Iterable, Iterator from contextlib import suppress from dataclasses import dataclass from pathlib import Path from typing import Any, TypeGuard from .models import RecordedMediaArtifact, ReplayCommand, SessionIntegrityError CAMERA_ARCHIVE_SCHEMA = "missioncore.camera-recording/v1" CAMERA_INDEX_SCHEMA = "missioncore.camera-recording-index/v1" RECORDED_MEDIA_MANIFEST_SCHEMA = "missioncore.observation-recorded-media/v4" RECORDED_MEDIA_PREPARATION_SCHEMA = "missioncore.recorded-media-preparation/v3" MAX_MEDIA_SUMMARY_BYTES = 2 * 1024 * 1024 MAX_MEDIA_INDEX_LINE_BYTES = 64 * 1024 MAX_INIT_BYTES = 8 * 1024 * 1024 MAX_MEDIA_SEGMENT_BYTES = 64 * 1024 * 1024 MAX_SAFE_INTEGER = (1 << 53) - 1 MAX_MP4_BOXES = 100_000 MAX_MP4_SAMPLES_PER_FRAGMENT = 1_000_000 MAX_MP4_FRAGMENT_DURATION_SECONDS = 3_600.0 MEDIA_RECORDING_TIMELINE_TOLERANCE_SECONDS = 0.05 MAX_MEDIA_EPOCHS = 4_096 _EPOCH_PATTERN = re.compile(r"^epoch-([1-9][0-9]*)$") _SHA256_PATTERN = re.compile(r"^[a-f0-9]{64}$") _SIDECAR_NAME_PATTERN = re.compile(r"^[a-f0-9]{64}\.json$") _SIDECAR_TEMP_PATTERN = re.compile(r"^\.tmp-[a-f0-9]{32}$") @dataclass(frozen=True, slots=True) class RecordedMediaSegment: sequence: int path: Path byte_length: int sha256: str random_access: bool end_time_seconds: float @dataclass(frozen=True, slots=True) class RecordedMediaEpoch: ordinal: int path: Path init_path: Path init_byte_length: int init_sha256: str media_type: str timeline_start_seconds: float timeline_end_seconds: float segments: tuple[RecordedMediaSegment, ...] @dataclass(frozen=True, slots=True) class RecordedMediaManifest: session_id: str public_source_id: str artifact_id: str synchronization: str generation_sha256: str timeline_start_seconds: float timeline_end_seconds: float byte_length: int epochs: tuple[RecordedMediaEpoch, ...] @dataclass(frozen=True, slots=True) class RecordedMediaFile: payload: bytes media_type: str byte_length: int sha256: str filename: str @dataclass(frozen=True, slots=True) class _CachedManifest: identity: tuple[tuple[int, int, int, int], ...] manifest: RecordedMediaManifest @dataclass(frozen=True, slots=True) class _Mp4VideoTiming: track_id: int timescale: int default_sample_duration: int | None default_sample_flags: int | None @dataclass(frozen=True, slots=True) class _Mp4VideoFragmentTiming: base_decode_time: int duration_units: int random_access: bool @property def end_decode_time(self) -> int: return self.base_decode_time + self.duration_units @dataclass(slots=True) class _Mp4ParseBudget: boxes_remaining: int = MAX_MP4_BOXES samples_remaining: int = MAX_MP4_SAMPLES_PER_FRAGMENT def consume_box(self) -> None: self.boxes_remaining -= 1 if self.boxes_remaining < 0: raise SessionIntegrityError("recorded media ISO-BMFF box budget was exceeded") def consume_samples(self, count: int) -> None: if count < 0 or count > self.samples_remaining: raise SessionIntegrityError("recorded media ISO-BMFF sample budget was exceeded") self.samples_remaining -= count class RecordedMediaInspector: """Validate canonical camera archives and retain only path-free descriptors. Cache keys are opaque catalog identities. A cache or persisted-sidecar hit rechecks the full native and camera source stat identity without rereading or parsing media; requested payload files are independently stat/hash checked before they are served. """ def __init__(self, cache_root: Path | None = None) -> None: self._lock = threading.Lock() self._cache: dict[tuple[str, str], _CachedManifest] = {} self._cache_root = _prepare_sidecar_root(cache_root) def inspect( self, artifact: RecordedMediaArtifact, replay: ReplayCommand, ) -> RecordedMediaManifest: if artifact.session_id != replay.session_id: raise SessionIntegrityError("recorded media does not belong to replay session") epoch_paths = _epoch_paths(artifact.source_path) key = (artifact.session_id, artifact.artifact_id) with self._lock: cached = self._cache.get(key) if cached is not None: identity = _prepared_source_identity( replay, epoch_paths, cached.manifest.epochs, ) if cached.identity == identity: return cached.manifest prepared = self._load_prepared_sidecar(artifact, replay, epoch_paths) if prepared is not None: manifest, identity = prepared with self._lock: self._cache[key] = _CachedManifest(identity=identity, manifest=manifest) return manifest manifest = _read_manifest( artifact, epoch_paths, origin_epoch_ns=replay.timeline_origin_epoch_ns, origin_monotonic_ns=replay.timeline_origin_monotonic_ns, ) identity = _prepared_source_identity( replay, epoch_paths, manifest.epochs, ) self._publish_prepared_sidecar(manifest, identity) with self._lock: self._cache[key] = _CachedManifest(identity=identity, manifest=manifest) return manifest def delete_prepared(self, session_id: str, artifact_ids: Iterable[str]) -> None: """Forget and remove path-free preparation sidecars for one session.""" unique_artifact_ids = tuple(dict.fromkeys(artifact_ids)) with self._lock: for artifact_id in unique_artifact_ids: self._cache.pop((session_id, artifact_id), None) if self._cache_root is None: return for artifact_id in unique_artifact_ids: sidecar = self._cache_root / _sidecar_name(session_id, artifact_id) try: metadata = sidecar.lstat() except FileNotFoundError: continue except OSError as exc: raise SessionIntegrityError( "recorded media preparation could not be inspected" ) from exc if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISREG(metadata.st_mode): raise SessionIntegrityError("recorded media preparation is not a regular file") try: sidecar.unlink() except OSError as exc: raise SessionIntegrityError( "recorded media preparation could not be deleted" ) from exc def restore_prepared( self, artifact: RecordedMediaArtifact, replay: ReplayCommand, ) -> RecordedMediaManifest | None: """Restore one previously prepared manifest without rebuilding it. A catalog read or process restart must never turn an old camera archive into fresh preparation work. The durable sidecar is accepted only when its complete source stat identity still matches the sealed archive. Missing or stale sidecars remain cold until the operator explicitly requests that session. """ if artifact.session_id != replay.session_id: raise SessionIntegrityError("recorded media does not belong to replay session") epoch_paths = _epoch_paths(artifact.source_path) key = (artifact.session_id, artifact.artifact_id) with self._lock: cached = self._cache.get(key) if cached is not None: identity = _prepared_source_identity( replay, epoch_paths, cached.manifest.epochs, ) if cached.identity == identity: return cached.manifest prepared = self._load_prepared_sidecar(artifact, replay, epoch_paths) if prepared is None: return None manifest, identity = prepared with self._lock: self._cache[key] = _CachedManifest(identity=identity, manifest=manifest) return manifest def _load_prepared_sidecar( self, artifact: RecordedMediaArtifact, replay: ReplayCommand, epoch_paths: tuple[Path, ...], ) -> ( tuple[ RecordedMediaManifest, tuple[tuple[int, int, int, int], ...], ] | None ): root = self._cache_root if root is None: return None path = root / _sidecar_name(artifact.session_id, artifact.artifact_id) try: sidecar_stat = _confined_file_stat(path, root) payload = _read_confined_file(path, root, max(1, sidecar_stat.st_size)) document = _decode_prepared_sidecar(payload) manifest = _manifest_from_sidecar(document, artifact, epoch_paths) identity = _prepared_source_identity( replay, epoch_paths, manifest.epochs, ) if _sidecar_identity(document, expected_length=len(identity)) != identity: return None return manifest, identity except (OSError, SessionIntegrityError, UnicodeDecodeError, json.JSONDecodeError): return None def _publish_prepared_sidecar( self, manifest: RecordedMediaManifest, identity: tuple[tuple[int, int, int, int], ...], ) -> None: root = self._cache_root if root is None: return payload = _encode_prepared_sidecar(manifest, identity) _write_sidecar_atomic( root, _sidecar_name(manifest.session_id, manifest.artifact_id), payload, ) def get_prepared( self, session_id: str, artifact_id: str, ) -> RecordedMediaManifest | None: """Return an already prepared manifest without touching its files.""" with self._lock: cached = self._cache.get((session_id, artifact_id)) return None if cached is None else cached.manifest def get_init(self, manifest: RecordedMediaManifest, ordinal: int) -> RecordedMediaFile: epoch = _epoch_by_ordinal(manifest, ordinal) return _validated_file( epoch.init_path, parent=epoch.path, media_type=epoch.media_type, expected_bytes=epoch.init_byte_length, expected_sha256=epoch.init_sha256, filename=f"recorded-camera-{ordinal}-init.mp4", ) def get_segment( self, manifest: RecordedMediaManifest, ordinal: int, sequence: int, ) -> RecordedMediaFile: epoch = _epoch_by_ordinal(manifest, ordinal) if sequence < 1 or sequence > len(epoch.segments): raise SessionIntegrityError("recorded media segment is outside the manifest") segment = epoch.segments[sequence - 1] if segment.sequence != sequence: raise SessionIntegrityError("recorded media segment sequence is inconsistent") return _validated_file( segment.path, parent=epoch.path / "segments", media_type="video/iso.segment", expected_bytes=segment.byte_length, expected_sha256=segment.sha256, filename=f"recorded-camera-{ordinal}-{sequence}.m4s", ) def inspect_recorded_media_epoch( epoch: Path, *, expected_source_name: str, origin_epoch_ns: int, origin_monotonic_ns: int, ) -> RecordedMediaEpoch: """Validate one canonical camera epoch for a bounded downstream handoff. This is the single-epoch form of :class:`RecordedMediaInspector`. It keeps compute-job packaging on the same digest and ISO-BMFF timing contract as saved-session replay without requiring the worker to see the rest of the observation session. """ match = _EPOCH_PATTERN.fullmatch(epoch.name) if match is None: raise SessionIntegrityError("recorded media epoch name is invalid") try: resolved = epoch.resolve(strict=True) source = resolved.parent.resolve(strict=True) except OSError as exc: raise SessionIntegrityError("recorded media epoch is unavailable") from exc if source.name != expected_source_name or resolved.parent != source: raise SessionIntegrityError("recorded media epoch source is inconsistent") return _read_epoch( resolved, ordinal=int(match.group(1)), expected_source_name=expected_source_name, origin_epoch_ns=origin_epoch_ns, origin_monotonic_ns=origin_monotonic_ns, ) def validate_recorded_media_timeline( manifests: tuple[RecordedMediaManifest, ...], *, recording_start_seconds: float, recording_end_seconds: float, ) -> None: """Require camera coverage to be reachable by the replay-v2 RRD timeline.""" if ( not math.isfinite(recording_start_seconds) or not math.isfinite(recording_end_seconds) or recording_start_seconds < 0 or recording_end_seconds < recording_start_seconds ): raise SessionIntegrityError("recording timeline is invalid") tolerance = MEDIA_RECORDING_TIMELINE_TOLERANCE_SECONDS for manifest in manifests: if ( manifest.timeline_start_seconds < recording_start_seconds - tolerance or manifest.timeline_end_seconds > recording_end_seconds + tolerance ): raise SessionIntegrityError( "recorded media timeline is outside the spatial recording timeline" ) def _prepare_sidecar_root(cache_root: Path | None) -> Path | None: if cache_root is None: return None root = cache_root.expanduser() root.mkdir(mode=0o700, parents=True, exist_ok=True) try: metadata = root.lstat() except OSError as exc: raise ValueError("recorded media cache root is unavailable") from exc if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode): raise ValueError("recorded media cache root must be a real directory") resolved = root.resolve(strict=True) for child in resolved.iterdir(): if _SIDECAR_TEMP_PATTERN.fullmatch(child.name) is None: continue try: child.unlink() except OSError: continue return resolved def _sidecar_name(session_id: str, artifact_id: str) -> str: key = f"{session_id}\0{artifact_id}".encode() return f"{hashlib.sha256(key).hexdigest()}.json" def _encode_prepared_sidecar( manifest: RecordedMediaManifest, identity: tuple[tuple[int, int, int, int], ...], ) -> bytes: body: dict[str, Any] = { "schema_version": RECORDED_MEDIA_PREPARATION_SCHEMA, "session_id": manifest.session_id, "public_source_id": manifest.public_source_id, "artifact_id": manifest.artifact_id, "source_identity": [list(item) for item in identity], "manifest": _sidecar_manifest_document(manifest), } checksum = hashlib.sha256(_canonical_json(body)).hexdigest() return _canonical_json({**body, "checksum_sha256": checksum}) def _decode_prepared_sidecar(payload: bytes) -> dict[str, Any]: value = json.loads(payload) if not isinstance(value, dict): raise SessionIntegrityError("recorded media preparation is not an object") checksum = value.get("checksum_sha256") if not isinstance(checksum, str) or _SHA256_PATTERN.fullmatch(checksum) is None: raise SessionIntegrityError("recorded media preparation checksum is invalid") body = {key: item for key, item in value.items() if key != "checksum_sha256"} if hashlib.sha256(_canonical_json(body)).hexdigest() != checksum: raise SessionIntegrityError("recorded media preparation checksum changed") if value.get("schema_version") != RECORDED_MEDIA_PREPARATION_SCHEMA: raise SessionIntegrityError("recorded media preparation schema is incompatible") return value def _sidecar_manifest_document(manifest: RecordedMediaManifest) -> dict[str, Any]: return { "schema_version": RECORDED_MEDIA_MANIFEST_SCHEMA, "source_id": manifest.public_source_id, "artifact_id": manifest.artifact_id, "generation_sha256": manifest.generation_sha256, "synchronization": manifest.synchronization, "timeline_start_seconds": manifest.timeline_start_seconds, "timeline_end_seconds": manifest.timeline_end_seconds, "byte_length": manifest.byte_length, "epochs": [ { "ordinal": epoch.ordinal, "timeline_start_seconds": epoch.timeline_start_seconds, "timeline_end_seconds": epoch.timeline_end_seconds, "media_type": epoch.media_type, "init_byte_length": epoch.init_byte_length, "init_sha256": epoch.init_sha256, "segments": [ { "sequence": segment.sequence, "byte_length": segment.byte_length, "sha256": segment.sha256, "random_access": segment.random_access, "end_time_seconds": segment.end_time_seconds, } for segment in epoch.segments ], } for epoch in manifest.epochs ], } def _manifest_from_sidecar( document: dict[str, Any], artifact: RecordedMediaArtifact, epoch_paths: tuple[Path, ...], ) -> RecordedMediaManifest: if ( document.get("session_id") != artifact.session_id or document.get("public_source_id") != artifact.public_source_id or document.get("artifact_id") != artifact.artifact_id ): raise SessionIntegrityError("recorded media preparation identity is inconsistent") descriptor = document.get("manifest") if not isinstance(descriptor, dict): raise SessionIntegrityError("recorded media preparation has no manifest") if ( descriptor.get("schema_version") != RECORDED_MEDIA_MANIFEST_SCHEMA or descriptor.get("source_id") != artifact.public_source_id or descriptor.get("artifact_id") != artifact.artifact_id or descriptor.get("synchronization") != "host-arrival-best-effort" ): raise SessionIntegrityError("recorded media prepared manifest is incompatible") epoch_documents = descriptor.get("epochs") if ( not isinstance(epoch_documents, list) or not epoch_documents or len(epoch_documents) != len(epoch_paths) or len(epoch_documents) > MAX_MEDIA_EPOCHS ): raise SessionIntegrityError("recorded media prepared epochs are invalid") epochs = tuple( _epoch_from_sidecar(epoch_document, epoch_path, ordinal) for ordinal, (epoch_document, epoch_path) in enumerate( zip(epoch_documents, epoch_paths, strict=True), start=1, ) ) _validate_epoch_timeline(epochs) byte_length = sum( epoch.init_byte_length + sum(segment.byte_length for segment in epoch.segments) for epoch in epochs ) timeline_start_seconds = min(epoch.timeline_start_seconds for epoch in epochs) timeline_end_seconds = max(epoch.timeline_end_seconds for epoch in epochs) if ( descriptor.get("byte_length") != byte_length or descriptor.get("timeline_start_seconds") != timeline_start_seconds or descriptor.get("timeline_end_seconds") != timeline_end_seconds or not 0 < byte_length <= MAX_SAFE_INTEGER ): raise SessionIntegrityError("recorded media prepared aggregate is inconsistent") generation_sha256 = _manifest_generation_sha256( public_source_id=artifact.public_source_id, artifact_id=artifact.artifact_id, synchronization="host-arrival-best-effort", epochs=epochs, ) if ( descriptor.get("generation_sha256") != generation_sha256 or _SHA256_PATTERN.fullmatch(generation_sha256) is None ): raise SessionIntegrityError("recorded media prepared generation is inconsistent") return RecordedMediaManifest( session_id=artifact.session_id, public_source_id=artifact.public_source_id, artifact_id=artifact.artifact_id, synchronization="host-arrival-best-effort", generation_sha256=generation_sha256, timeline_start_seconds=timeline_start_seconds, timeline_end_seconds=timeline_end_seconds, byte_length=byte_length, epochs=epochs, ) def _epoch_from_sidecar( value: object, epoch_path: Path, ordinal: int, ) -> RecordedMediaEpoch: if not isinstance(value, dict) or value.get("ordinal") != ordinal: raise SessionIntegrityError("recorded media prepared epoch is inconsistent") start = value.get("timeline_start_seconds") end = value.get("timeline_end_seconds") media_type = value.get("media_type") init_byte_length = value.get("init_byte_length") init_sha256 = value.get("init_sha256") if ( not _finite_non_negative_number(start) or not _finite_non_negative_number(end) or float(end) < float(start) or not isinstance(media_type, str) or not media_type.startswith("video/mp4") or len(media_type) > 256 or not _positive_int(init_byte_length) or int(init_byte_length) > MAX_INIT_BYTES or not isinstance(init_sha256, str) or _SHA256_PATTERN.fullmatch(init_sha256) is None ): raise SessionIntegrityError("recorded media prepared epoch descriptor is invalid") segment_documents = value.get("segments") if not isinstance(segment_documents, list) or not segment_documents: raise SessionIntegrityError("recorded media prepared segments are invalid") segments_root = epoch_path / "segments" segments: list[RecordedMediaSegment] = [] previous_end_time_seconds = 0.0 for sequence, segment in enumerate(segment_documents, start=1): if not isinstance(segment, dict) or segment.get("sequence") != sequence: raise SessionIntegrityError("recorded media prepared segment is inconsistent") byte_length = segment.get("byte_length") sha256 = segment.get("sha256") random_access = segment.get("random_access") end_time_seconds = segment.get("end_time_seconds") if ( not _positive_int(byte_length) or int(byte_length) > MAX_MEDIA_SEGMENT_BYTES or not isinstance(sha256, str) or _SHA256_PATTERN.fullmatch(sha256) is None or not isinstance(random_access, bool) or not _finite_non_negative_number(end_time_seconds) or float(end_time_seconds) <= previous_end_time_seconds ): raise SessionIntegrityError("recorded media prepared segment is invalid") previous_end_time_seconds = float(end_time_seconds) segments.append( RecordedMediaSegment( sequence=sequence, path=segments_root / f"{sequence}.m4s", byte_length=int(byte_length), sha256=sha256, random_access=random_access, end_time_seconds=previous_end_time_seconds, ) ) if not math.isclose( previous_end_time_seconds, float(end) - float(start), rel_tol=0.0, abs_tol=1e-9, ): raise SessionIntegrityError("recorded media prepared segment timeline is inconsistent") return RecordedMediaEpoch( ordinal=ordinal, path=epoch_path, init_path=epoch_path / "init.mp4", init_byte_length=int(init_byte_length), init_sha256=init_sha256, media_type=media_type, timeline_start_seconds=float(start), timeline_end_seconds=float(end), segments=tuple(segments), ) def _sidecar_identity( document: dict[str, Any], *, expected_length: int, ) -> tuple[tuple[int, int, int, int], ...]: value = document.get("source_identity") if not isinstance(value, list) or expected_length < 1 or len(value) != expected_length: raise SessionIntegrityError("recorded media preparation source identity is invalid") identity: list[tuple[int, int, int, int]] = [] for item in value: if ( not isinstance(item, list) or len(item) != 4 or any(not _non_negative_int(component) for component in item) ): raise SessionIntegrityError("recorded media preparation source identity is invalid") identity.append((int(item[0]), int(item[1]), int(item[2]), int(item[3]))) return tuple(identity) def _canonical_json(value: object) -> bytes: try: encoded = json.dumps( value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False, ).encode("utf-8") except (TypeError, ValueError) as exc: raise SessionIntegrityError("recorded media preparation cannot be encoded") from exc if not encoded: raise SessionIntegrityError("recorded media preparation is empty") return encoded def _write_sidecar_atomic(root: Path, filename: str, payload: bytes) -> None: if _SIDECAR_NAME_PATTERN.fullmatch(filename) is None: raise SessionIntegrityError("recorded media preparation filename is invalid") directory_fd = os.open( root, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0), ) temporary = f".tmp-{secrets.token_hex(16)}" descriptor: int | None = None try: descriptor = os.open( temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0), 0o600, dir_fd=directory_fd, ) remaining = memoryview(payload) while remaining: written = os.write(descriptor, remaining) if written <= 0: raise OSError("prepared media sidecar write made no progress") remaining = remaining[written:] os.fsync(descriptor) os.close(descriptor) descriptor = None os.replace( temporary, filename, src_dir_fd=directory_fd, dst_dir_fd=directory_fd, ) os.fsync(directory_fd) finally: if descriptor is not None: os.close(descriptor) with suppress(FileNotFoundError): os.unlink(temporary, dir_fd=directory_fd) os.close(directory_fd) def _finite_non_negative_number(value: object) -> TypeGuard[int | float]: return ( isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(float(value)) and float(value) >= 0 ) def _epoch_paths(source_path: Path) -> tuple[Path, ...]: try: source = source_path.resolve(strict=True) except OSError as exc: raise SessionIntegrityError("recorded media source is missing") from exc if not source.is_dir(): raise SessionIntegrityError("recorded media source is not a directory") epochs: list[tuple[int, Path]] = [] for child in source.iterdir(): match = _EPOCH_PATTERN.fullmatch(child.name) if match is None or not child.is_dir(): continue try: resolved = child.resolve(strict=True) except OSError as exc: raise SessionIntegrityError("recorded media epoch is unavailable") from exc if resolved.parent != source: raise SessionIntegrityError("recorded media epoch escapes its source") epochs.append((int(match.group(1)), resolved)) if not epochs: raise SessionIntegrityError("recorded media source has no codec epochs") if len(epochs) > MAX_MEDIA_EPOCHS: raise SessionIntegrityError("recorded media source has too many codec epochs") return tuple(path for _, path in sorted(epochs)) def _prepared_source_identity( replay: ReplayCommand, epoch_paths: tuple[Path, ...], epochs: tuple[RecordedMediaEpoch, ...], ) -> tuple[tuple[int, int, int, int], ...]: if len(epoch_paths) != len(epochs): raise SessionIntegrityError("recorded media epoch identity is inconsistent") identity: list[tuple[int, int, int, int]] = [] for artifact in replay.artifacts: metadata = _session_artifact_stat(artifact.path, replay.session_root) identity.append((metadata.st_dev, metadata.st_ino, metadata.st_size, metadata.st_mtime_ns)) for expected_ordinal, (epoch_path, epoch) in enumerate( zip(epoch_paths, epochs, strict=True), start=1, ): if epoch.ordinal != expected_ordinal or epoch.path != epoch_path: raise SessionIntegrityError("recorded media epoch identity is inconsistent") for filename in ("summary.json", "index.jsonl", "init.mp4"): metadata = _confined_file_stat(epoch_path / filename, epoch_path) identity.append( (metadata.st_dev, metadata.st_ino, metadata.st_size, metadata.st_mtime_ns) ) try: segments_root = (epoch_path / "segments").resolve(strict=True) except OSError as exc: raise SessionIntegrityError("recorded media segment directory is missing") from exc if not segments_root.is_dir() or segments_root.parent != epoch_path: raise SessionIntegrityError("recorded media segment directory is invalid") for segment in epoch.segments: metadata = _confined_file_stat( segments_root / f"{segment.sequence}.m4s", segments_root, ) identity.append( (metadata.st_dev, metadata.st_ino, metadata.st_size, metadata.st_mtime_ns) ) return tuple(identity) def _session_artifact_stat(path: Path, session_root: Path) -> os.stat_result: try: session = session_root.resolve(strict=True) parent = path.parent.resolve(strict=True) except OSError as exc: raise SessionIntegrityError("recording source artifact is missing") from exc if not parent.is_relative_to(session): raise SessionIntegrityError("recording source artifact escapes its session") return _confined_file_stat(path, parent) def _read_manifest( artifact: RecordedMediaArtifact, epoch_paths: tuple[Path, ...], *, origin_epoch_ns: int, origin_monotonic_ns: int, ) -> RecordedMediaManifest: epochs = tuple( _read_epoch( path, ordinal=index, expected_source_name=artifact.source_path.name, origin_epoch_ns=origin_epoch_ns, origin_monotonic_ns=origin_monotonic_ns, ) for index, path in enumerate(epoch_paths, start=1) ) _validate_epoch_timeline(epochs) byte_length = sum( epoch.init_byte_length + sum(segment.byte_length for segment in epoch.segments) for epoch in epochs ) if not 0 < byte_length <= MAX_SAFE_INTEGER: raise SessionIntegrityError("recorded media byte length is outside safe bounds") timeline_start_seconds = min(epoch.timeline_start_seconds for epoch in epochs) timeline_end_seconds = max(epoch.timeline_end_seconds for epoch in epochs) generation_sha256 = _manifest_generation_sha256( public_source_id=artifact.public_source_id, artifact_id=artifact.artifact_id, synchronization="host-arrival-best-effort", epochs=epochs, ) return RecordedMediaManifest( session_id=artifact.session_id, public_source_id=artifact.public_source_id, artifact_id=artifact.artifact_id, synchronization="host-arrival-best-effort", generation_sha256=generation_sha256, timeline_start_seconds=timeline_start_seconds, timeline_end_seconds=timeline_end_seconds, byte_length=byte_length, epochs=epochs, ) def _manifest_generation_sha256( *, public_source_id: str, artifact_id: str, synchronization: str, epochs: tuple[RecordedMediaEpoch, ...], ) -> str: """Digest the immutable, path-free descriptor generation.""" timeline_start_seconds = min(epoch.timeline_start_seconds for epoch in epochs) timeline_end_seconds = max(epoch.timeline_end_seconds for epoch in epochs) byte_length = sum( epoch.init_byte_length + sum(segment.byte_length for segment in epoch.segments) for epoch in epochs ) descriptor = { "schema_version": RECORDED_MEDIA_MANIFEST_SCHEMA, "source_id": public_source_id, "artifact_id": artifact_id, "synchronization": synchronization, "timeline_start_seconds": timeline_start_seconds, "timeline_end_seconds": timeline_end_seconds, "byte_length": byte_length, "epochs": [ { "ordinal": epoch.ordinal, "timeline_start_seconds": epoch.timeline_start_seconds, "timeline_end_seconds": epoch.timeline_end_seconds, "media_type": epoch.media_type, "init_byte_length": epoch.init_byte_length, "init_sha256": epoch.init_sha256, "segments": [ { "sequence": segment.sequence, "byte_length": segment.byte_length, "sha256": segment.sha256, "random_access": segment.random_access, "end_time_seconds": segment.end_time_seconds, } for segment in epoch.segments ], } for epoch in epochs ], } payload = json.dumps( descriptor, ensure_ascii=False, sort_keys=True, separators=(",", ":"), ).encode("utf-8") return hashlib.sha256(payload).hexdigest() def _read_epoch( epoch: Path, *, ordinal: int, expected_source_name: str, origin_epoch_ns: int, origin_monotonic_ns: int, ) -> RecordedMediaEpoch: summary = _read_json_object(epoch / "summary.json", MAX_MEDIA_SUMMARY_BYTES) if ( summary.get("schema_version") != CAMERA_ARCHIVE_SCHEMA or summary.get("source_id") != expected_source_name or summary.get("synchronization") != "host-arrival-best-effort" ): raise SessionIntegrityError("recorded media summary is incompatible") segment_count = summary.get("segment_count") if not _non_negative_int(segment_count) or not 1 <= int(segment_count) <= MAX_SAFE_INTEGER: raise SessionIntegrityError("recorded media segment count is invalid") match = _EPOCH_PATTERN.fullmatch(epoch.name) if ( match is None or summary.get("codec_epoch") != int(match.group(1)) or summary.get("entry_count") != int(segment_count) or summary.get("media_segment_count") != int(segment_count) or summary.get("commit_policy") != "per-segment-fsync" or summary.get("artifacts") != { "init": "init.mp4", "segments": "segments", "index": "index.jsonl", } ): raise SessionIntegrityError("recorded media summary aggregate is inconsistent") entries, index_sha256 = _read_media_index( epoch / "index.jsonl", epoch, expected_count=int(segment_count), ) expected_index_sha256 = summary.get("index_sha256") if not isinstance(expected_index_sha256, str) or index_sha256 != expected_index_sha256: raise SessionIntegrityError("recorded media index digest changed") use_monotonic_clock = all( _non_negative_int(entry.get("host_monotonic_ns")) for entry in entries ) timeline_points = tuple( max( 0.0, _entry_session_seconds( entry, origin_epoch_ns=origin_epoch_ns, origin_monotonic_ns=origin_monotonic_ns, use_monotonic_clock=use_monotonic_clock, ), ) for entry in entries ) if any(not math.isfinite(point) for point in timeline_points): raise SessionIntegrityError("recorded media timeline contains a non-finite value") if any( current >= following for current, following in zip( timeline_points, timeline_points[1:], strict=False, ) ): raise SessionIntegrityError("recorded media segment timeline is not strictly monotonic") segments_root = (epoch / "segments").resolve(strict=True) if not segments_root.is_dir() or segments_root.parent != epoch: raise SessionIntegrityError("recorded media segment directory is invalid") segment_descriptors: list[tuple[int, Path, int, str]] = [] for sequence, entry in enumerate(entries, start=1): byte_length = entry.get("length") digest = entry.get("sha256") if ( not _positive_int(byte_length) or int(byte_length) > MAX_MEDIA_SEGMENT_BYTES or not isinstance(digest, str) or _SHA256_PATTERN.fullmatch(digest) is None ): raise SessionIntegrityError("recorded media segment descriptor is invalid") path = (segments_root / f"{sequence}.m4s").resolve() metadata = _confined_file_stat( segments_root / f"{sequence}.m4s", segments_root, ) if metadata.st_size <= 0 or metadata.st_size > MAX_MEDIA_SEGMENT_BYTES: raise SessionIntegrityError("recorded media segment length is outside bounds") if metadata.st_size != int(byte_length): raise SessionIntegrityError("recorded media segment length changed") segment_descriptors.append((sequence, path, int(byte_length), digest)) init_path = (epoch / "init.mp4").resolve() init_payload = _read_confined_file(init_path, epoch, MAX_INIT_BYTES) init_sha256 = hashlib.sha256(init_payload).hexdigest() expected_init_sha256 = summary.get("init_sha256") if not isinstance(expected_init_sha256, str) or expected_init_sha256 != init_sha256: raise SessionIntegrityError("recorded media init digest changed") media_type = _mp4_media_type(init_payload) timing = _mp4_video_timing(init_payload, _Mp4ParseBudget()) fragment_timings: list[_Mp4VideoFragmentTiming] = [] stream_sha256 = hashlib.sha256(init_payload) valid_bytes = len(init_payload) for _sequence, path, byte_length, digest in segment_descriptors: payload = _read_confined_file( path, segments_root, byte_length, ) if hashlib.sha256(payload).hexdigest() != digest: raise SessionIntegrityError("recorded media segment digest changed") stream_sha256.update(payload) valid_bytes += len(payload) fragment_timing = _mp4_video_fragment_timing( payload, timing, _Mp4ParseBudget(), ) fragment_timings.append(fragment_timing) if ( summary.get("valid_bytes") != valid_bytes or summary.get("stream_sha256") != stream_sha256.hexdigest() ): raise SessionIntegrityError("recorded media stream aggregate changed") if not fragment_timings[0].random_access: raise SessionIntegrityError( "recorded media codec epoch does not begin with a random-access fragment" ) for previous, current in zip( fragment_timings, fragment_timings[1:], strict=False, ): if current.base_decode_time != previous.end_decode_time: raise SessionIntegrityError("recorded media fragment decode timeline is discontinuous") duration_units = [timing.duration_units for timing in fragment_timings] first_duration_seconds = _checked_fragment_duration_seconds( duration_units[0], timing.timescale, ) total_duration_units = sum(duration_units) if total_duration_units > MAX_SAFE_INTEGER: raise SessionIntegrityError("recorded media epoch duration is outside safe bounds") total_duration_seconds = total_duration_units / timing.timescale timeline_start_seconds = max(0.0, timeline_points[0] - first_duration_seconds) timeline_end_seconds = timeline_start_seconds + total_duration_seconds if not math.isfinite(timeline_end_seconds) or timeline_end_seconds < timeline_start_seconds: raise SessionIntegrityError("recorded media epoch timeline is invalid") epoch_duration_seconds = timeline_end_seconds - timeline_start_seconds segments: list[RecordedMediaSegment] = [] cumulative_duration_units = 0 previous_end_time_seconds = 0.0 for index, (descriptor, fragment_timing) in enumerate( zip(segment_descriptors, fragment_timings, strict=True), start=1, ): sequence, path, byte_length, digest = descriptor cumulative_duration_units += fragment_timing.duration_units end_time_seconds = cumulative_duration_units / timing.timescale if index == len(fragment_timings): end_time_seconds = epoch_duration_seconds if not math.isfinite(end_time_seconds) or end_time_seconds <= previous_end_time_seconds: raise SessionIntegrityError("recorded media segment timeline is invalid") previous_end_time_seconds = end_time_seconds segments.append( RecordedMediaSegment( sequence=sequence, path=path, byte_length=byte_length, sha256=digest, random_access=fragment_timing.random_access, end_time_seconds=end_time_seconds, ) ) return RecordedMediaEpoch( ordinal=ordinal, path=epoch, init_path=init_path, init_byte_length=len(init_payload), init_sha256=init_sha256, media_type=media_type, timeline_start_seconds=timeline_start_seconds, timeline_end_seconds=timeline_end_seconds, segments=tuple(segments), ) def _entry_session_seconds( entry: dict[str, Any], *, origin_epoch_ns: int, origin_monotonic_ns: int, use_monotonic_clock: bool, ) -> float: monotonic_ns = entry.get("host_monotonic_ns") if use_monotonic_clock and _non_negative_int(monotonic_ns): try: return (int(monotonic_ns) - origin_monotonic_ns) / 1_000_000_000 except OverflowError as exc: raise SessionIntegrityError("recorded media timeline is outside bounds") from exc epoch_ns = entry.get("host_epoch_ns") if _non_negative_int(epoch_ns): try: return (int(epoch_ns) - origin_epoch_ns) / 1_000_000_000 except OverflowError as exc: raise SessionIntegrityError("recorded media timeline is outside bounds") from exc raise SessionIntegrityError("recorded media has no host-arrival synchronization point") def _validate_epoch_timeline(epochs: tuple[RecordedMediaEpoch, ...]) -> None: previous_end: float | None = None for expected_ordinal, epoch in enumerate(epochs, start=1): start = epoch.timeline_start_seconds end = epoch.timeline_end_seconds if epoch.ordinal != expected_ordinal: raise SessionIntegrityError("recorded media epoch ordinal is inconsistent") if not math.isfinite(start) or not math.isfinite(end) or start < 0 or end < start: raise SessionIntegrityError("recorded media epoch timeline is invalid") if previous_end is not None and start < previous_end: raise SessionIntegrityError("recorded media codec epochs overlap") previous_end = end def _mp4_fragment_duration_seconds(init_payload: bytes, fragment_payload: bytes) -> float: """Return the bounded decoded duration of one ISO-BMFF video fragment. Camera timeline anchors are complete-fragment host-arrival observations. This helper proves the exact decoded duration declared by one fragment. If any timing field is absent or ambiguous the archive is not advertised as replayable. """ budget = _Mp4ParseBudget() timing = _mp4_video_timing(init_payload, budget) fragment_timing = _mp4_video_fragment_timing( fragment_payload, timing, budget, ) return _checked_fragment_duration_seconds( fragment_timing.duration_units, timing.timescale, ) def _checked_fragment_duration_seconds(duration_units: int, timescale: int) -> float: if duration_units <= 0: raise SessionIntegrityError("recorded media fragment has no positive duration") duration = duration_units / timescale if not math.isfinite(duration) or duration <= 0 or duration > MAX_MP4_FRAGMENT_DURATION_SECONDS: raise SessionIntegrityError("recorded media fragment duration is outside bounds") return duration def _mp4_video_timing(payload: bytes, budget: _Mp4ParseBudget) -> _Mp4VideoTiming: moov_payloads = [ box_payload for box_type, box_payload in _iter_mp4_boxes(payload, budget) if box_type == b"moov" ] if len(moov_payloads) != 1: raise SessionIntegrityError("recorded media init has no unique moov box") moov = moov_payloads[0] moov_boxes = tuple(_iter_mp4_boxes(moov, budget)) defaults: dict[int, tuple[int, int]] = {} for box_type, box_payload in moov_boxes: if box_type != b"mvex": continue for child_type, child_payload in _iter_mp4_boxes(box_payload, budget): if child_type != b"trex": continue track_id, trex_default_duration, trex_default_flags = _parse_trex(child_payload) if track_id in defaults: raise SessionIntegrityError("recorded media init repeats a trex track") defaults[track_id] = (trex_default_duration, trex_default_flags) video_tracks: list[tuple[int, int]] = [] for box_type, trak_payload in moov_boxes: if box_type != b"trak": continue track_id = _trak_track_id(trak_payload, budget) media = _trak_media_timing(trak_payload, budget) if media is not None: video_tracks.append((track_id, media)) if len(video_tracks) != 1: raise SessionIntegrityError("recorded media init has no unique video track") track_id, timescale = video_tracks[0] default_sample = defaults.get(track_id) default_duration = None if default_sample is None else default_sample[0] return _Mp4VideoTiming( track_id=track_id, timescale=timescale, default_sample_duration=( default_duration if default_duration is not None and default_duration > 0 else None ), default_sample_flags=None if default_sample is None else default_sample[1], ) def _trak_track_id(payload: bytes, budget: _Mp4ParseBudget) -> int: track_ids = [ _parse_tkhd_track_id(box_payload) for box_type, box_payload in _iter_mp4_boxes(payload, budget) if box_type == b"tkhd" ] if len(track_ids) != 1 or track_ids[0] <= 0: raise SessionIntegrityError("recorded media track id is invalid") return track_ids[0] def _trak_media_timing(payload: bytes, budget: _Mp4ParseBudget) -> int | None: media_boxes = [ box_payload for box_type, box_payload in _iter_mp4_boxes(payload, budget) if box_type == b"mdia" ] if len(media_boxes) != 1: raise SessionIntegrityError("recorded media track has no unique mdia box") children = tuple(_iter_mp4_boxes(media_boxes[0], budget)) handlers = [ _parse_hdlr_type(box_payload) for box_type, box_payload in children if box_type == b"hdlr" ] if len(handlers) != 1: raise SessionIntegrityError("recorded media track handler is ambiguous") if handlers[0] != b"vide": return None timescales = [ _parse_mdhd_timescale(box_payload) for box_type, box_payload in children if box_type == b"mdhd" ] if len(timescales) != 1 or timescales[0] <= 0: raise SessionIntegrityError("recorded media video timescale is invalid") return timescales[0] def _mp4_video_fragment_timing( payload: bytes, timing: _Mp4VideoTiming, budget: _Mp4ParseBudget, ) -> _Mp4VideoFragmentTiming: moof_payloads = [ box_payload for box_type, box_payload in _iter_mp4_boxes(payload, budget) if box_type == b"moof" ] if len(moof_payloads) != 1: raise SessionIntegrityError("recorded media fragment has no unique moof box") matching_timings: list[_Mp4VideoFragmentTiming] = [] for box_type, traf_payload in _iter_mp4_boxes(moof_payloads[0], budget): if box_type != b"traf": continue boxes = tuple(_iter_mp4_boxes(traf_payload, budget)) tfhd_payloads = [box for kind, box in boxes if kind == b"tfhd"] if len(tfhd_payloads) != 1: raise SessionIntegrityError("recorded media fragment tfhd is ambiguous") track_id, fragment_default_duration, fragment_default_flags = _parse_tfhd(tfhd_payloads[0]) if track_id != timing.track_id: continue tfdt_payloads = [box for kind, box in boxes if kind == b"tfdt"] if len(tfdt_payloads) != 1: raise SessionIntegrityError("recorded media fragment tfdt is ambiguous") base_decode_time = _parse_tfdt(tfdt_payloads[0]) trun_payloads = [box for kind, box in boxes if kind == b"trun"] if not trun_payloads: raise SessionIntegrityError("recorded media video fragment has no trun box") default_duration = fragment_default_duration or timing.default_sample_duration default_flags = ( fragment_default_flags if fragment_default_flags is not None else timing.default_sample_flags ) trun_descriptors = tuple( _parse_trun_descriptor(trun, default_duration, default_flags, budget) for trun in trun_payloads ) fragment_sample_count = sum(item[2] for item in trun_descriptors) if fragment_sample_count != 1: raise SessionIntegrityError( "recorded media video fragment must contain exactly one sample" ) duration = sum(item[0] for item in trun_descriptors) if base_decode_time > MAX_SAFE_INTEGER - duration: raise SessionIntegrityError("recorded media fragment decode time is outside bounds") matching_timings.append( _Mp4VideoFragmentTiming( base_decode_time=base_decode_time, duration_units=duration, random_access=(trun_descriptors[0][1] & 0x00010000) == 0, ) ) if len(matching_timings) != 1: raise SessionIntegrityError("recorded media fragment video track is ambiguous") return matching_timings[0] def _iter_mp4_boxes( payload: bytes, budget: _Mp4ParseBudget, ) -> Iterator[tuple[bytes, bytes]]: offset = 0 payload_length = len(payload) while offset < payload_length: if payload_length - offset < 8: raise SessionIntegrityError("recorded media ISO-BMFF box is truncated") size = int.from_bytes(payload[offset : offset + 4], "big") box_type = payload[offset + 4 : offset + 8] header_length = 8 if size == 1: if payload_length - offset < 16: raise SessionIntegrityError("recorded media ISO-BMFF box is truncated") size = int.from_bytes(payload[offset + 8 : offset + 16], "big") header_length = 16 elif size == 0: size = payload_length - offset if size < header_length or size > payload_length - offset: raise SessionIntegrityError("recorded media ISO-BMFF box size is invalid") budget.consume_box() end = offset + size yield box_type, payload[offset + header_length : end] offset = end def _parse_tkhd_track_id(payload: bytes) -> int: version = _full_box_version(payload) offset = 20 if version == 1 else 12 if version == 0 else -1 return _read_u32(payload, offset, "tkhd track id") def _parse_mdhd_timescale(payload: bytes) -> int: version = _full_box_version(payload) offset = 20 if version == 1 else 12 if version == 0 else -1 return _read_u32(payload, offset, "mdhd timescale") def _parse_hdlr_type(payload: bytes) -> bytes: _full_box_version(payload) if len(payload) < 12: raise SessionIntegrityError("recorded media hdlr box is truncated") return payload[8:12] def _parse_trex(payload: bytes) -> tuple[int, int, int]: _full_box_version(payload) return ( _read_u32(payload, 4, "trex track id"), _read_u32(payload, 12, "trex default sample duration"), _read_u32(payload, 20, "trex default sample flags"), ) def _parse_tfhd(payload: bytes) -> tuple[int, int | None, int | None]: flags = _full_box_flags(payload) track_id = _read_u32(payload, 4, "tfhd track id") cursor = 8 for flag, width in ((0x000001, 8), (0x000002, 4)): if flags & flag: cursor = _advance_box_cursor(payload, cursor, width, "tfhd optional field") default_duration: int | None = None if flags & 0x000008: default_duration = _read_u32(payload, cursor, "tfhd default sample duration") cursor += 4 if flags & 0x000010: cursor = _advance_box_cursor(payload, cursor, 4, "tfhd default sample size") default_flags: int | None = None if flags & 0x000020: default_flags = _read_u32(payload, cursor, "tfhd default sample flags") return ( track_id, default_duration if default_duration and default_duration > 0 else None, default_flags, ) def _parse_tfdt(payload: bytes) -> int: version = _full_box_version(payload) if version == 0: return _read_u32(payload, 4, "tfdt base decode time") if version == 1: if len(payload) < 12: raise SessionIntegrityError("recorded media tfdt base decode time is truncated") return int.from_bytes(payload[4:12], "big") raise SessionIntegrityError("recorded media tfdt version is unsupported") def _parse_trun_descriptor( payload: bytes, default_duration: int | None, default_sample_flags: int | None, budget: _Mp4ParseBudget, ) -> tuple[int, int, int]: flags = _full_box_flags(payload) sample_count = _read_u32(payload, 4, "trun sample count") if sample_count < 1: raise SessionIntegrityError("recorded media trun has no samples") budget.consume_samples(sample_count) cursor = 8 if flags & 0x000001: cursor = _advance_box_cursor(payload, cursor, 4, "trun data offset") first_sample_flags: int | None = None if flags & 0x000004: first_sample_flags = _read_u32(payload, cursor, "trun first sample flags") cursor += 4 if first_sample_flags is not None and flags & 0x000400: raise SessionIntegrityError("recorded media trun sample flags are ambiguous") duration = 0 first_per_sample_flags: int | None = None for sample_index in range(sample_count): if flags & 0x000100: sample_duration = _read_u32(payload, cursor, "trun sample duration") if sample_duration <= 0: raise SessionIntegrityError("recorded media sample duration is invalid") duration += sample_duration cursor += 4 elif default_duration is None or default_duration <= 0: raise SessionIntegrityError("recorded media sample duration is unavailable") else: duration += default_duration if flags & 0x000200: cursor = _advance_box_cursor(payload, cursor, 4, "trun sample size") if flags & 0x000400: sample_flags = _read_u32(payload, cursor, "trun sample flags") if sample_index == 0: first_per_sample_flags = sample_flags cursor += 4 if flags & 0x000800: cursor = _advance_box_cursor( payload, cursor, 4, "trun sample composition time offset", ) effective_flags = ( first_per_sample_flags if first_per_sample_flags is not None else first_sample_flags if first_sample_flags is not None else default_sample_flags ) if effective_flags is None: raise SessionIntegrityError("recorded media sample flags are unavailable") return duration, effective_flags, sample_count def _full_box_version(payload: bytes) -> int: if len(payload) < 4: raise SessionIntegrityError("recorded media full box is truncated") return payload[0] def _full_box_flags(payload: bytes) -> int: _full_box_version(payload) return int.from_bytes(payload[1:4], "big") def _read_u32(payload: bytes, offset: int, description: str) -> int: if offset < 0 or offset + 4 > len(payload): raise SessionIntegrityError(f"recorded media {description} is truncated") return int.from_bytes(payload[offset : offset + 4], "big") def _advance_box_cursor(payload: bytes, cursor: int, width: int, description: str) -> int: if cursor < 0 or width < 0 or cursor + width > len(payload): raise SessionIntegrityError(f"recorded media {description} is truncated") return cursor + width def _mp4_media_type(init_payload: bytes) -> str: marker = init_payload.find(b"avcC") if marker >= 0 and marker + 8 <= len(init_payload): configuration_version = init_payload[marker + 4] if configuration_version == 1: profile = init_payload[marker + 5] compatibility = init_payload[marker + 6] level = init_payload[marker + 7] return f'video/mp4; codecs="avc1.{profile:02X}{compatibility:02X}{level:02X}"' # The archive is still valid evidence, but browser MSE cannot be opened # safely without an explicit codec string. return "video/mp4" def _epoch_by_ordinal( manifest: RecordedMediaManifest, ordinal: int, ) -> RecordedMediaEpoch: if ordinal < 1 or ordinal > len(manifest.epochs): raise SessionIntegrityError("recorded media epoch is outside the manifest") epoch = manifest.epochs[ordinal - 1] if epoch.ordinal != ordinal: raise SessionIntegrityError("recorded media epoch ordinal is inconsistent") return epoch def _validated_file( path: Path, *, parent: Path, media_type: str, expected_bytes: int, expected_sha256: str, filename: str, ) -> RecordedMediaFile: payload = _read_confined_file(path, parent, max(expected_bytes, 1)) if len(payload) != expected_bytes or hashlib.sha256(payload).hexdigest() != expected_sha256: raise SessionIntegrityError("recorded media payload changed after indexing") return RecordedMediaFile( payload=payload, media_type=media_type, byte_length=expected_bytes, sha256=expected_sha256, filename=filename, ) def _confined_file_stat(path: Path, parent: Path) -> os.stat_result: try: resolved_parent = parent.resolve(strict=True) if path.parent.resolve(strict=True) != resolved_parent: raise SessionIntegrityError("recorded media payload escapes its epoch") parent_fd = os.open( resolved_parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0), ) except OSError as exc: raise SessionIntegrityError("recorded media payload is missing") from exc try: flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) descriptor = os.open(path.name, flags, dir_fd=parent_fd) try: stat_result = os.fstat(descriptor) finally: os.close(descriptor) except OSError as exc: raise SessionIntegrityError("recorded media payload is not a regular file") from exc finally: os.close(parent_fd) if not stat.S_ISREG(stat_result.st_mode): raise SessionIntegrityError("recorded media payload is not a regular file") return stat_result def _read_confined_file(path: Path, parent: Path, maximum_bytes: int) -> bytes: try: resolved_parent = parent.resolve(strict=True) if path.parent.resolve(strict=True) != resolved_parent: raise SessionIntegrityError("recorded media payload escapes its epoch") parent_fd = os.open( resolved_parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0), ) except OSError as exc: raise SessionIntegrityError("recorded media payload is missing") from exc try: flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) descriptor = os.open(path.name, flags, dir_fd=parent_fd) try: stat_result = os.fstat(descriptor) if ( not stat.S_ISREG(stat_result.st_mode) or not 0 < stat_result.st_size <= maximum_bytes ): raise SessionIntegrityError("recorded media payload is outside bounds") chunks: list[bytes] = [] remaining = stat_result.st_size while remaining: chunk = os.read(descriptor, min(remaining, 1024 * 1024)) if not chunk: raise SessionIntegrityError("recorded media payload was truncated") chunks.append(chunk) remaining -= len(chunk) if os.read(descriptor, 1): raise SessionIntegrityError("recorded media payload grew during validation") after = os.fstat(descriptor) if ( after.st_dev != stat_result.st_dev or after.st_ino != stat_result.st_ino or after.st_size != stat_result.st_size or after.st_mtime_ns != stat_result.st_mtime_ns ): raise SessionIntegrityError("recorded media payload changed during validation") return b"".join(chunks) finally: os.close(descriptor) except OSError as exc: raise SessionIntegrityError("recorded media payload is unavailable") from exc finally: os.close(parent_fd) def _read_first_confined_line(path: Path, parent: Path, maximum_bytes: int) -> bytes: try: resolved_parent = parent.resolve(strict=True) if path.parent.resolve(strict=True) != resolved_parent: raise SessionIntegrityError("recorded media timeline escapes its session") parent_fd = os.open( resolved_parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0), ) except OSError as exc: raise SessionIntegrityError("recorded media timeline is missing") from exc try: descriptor = os.open( path.name, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0), dir_fd=parent_fd, ) try: if not stat.S_ISREG(os.fstat(descriptor).st_mode): raise SessionIntegrityError("recorded media timeline is not regular") payload = bytearray() while len(payload) <= maximum_bytes: chunk = os.read(descriptor, min(64 * 1024, maximum_bytes + 1 - len(payload))) if not chunk: break newline = chunk.find(b"\n") if newline >= 0: payload.extend(chunk[: newline + 1]) break payload.extend(chunk) if len(payload) > maximum_bytes: raise SessionIntegrityError("recorded media timeline line exceeds its boundary") return bytes(payload) finally: os.close(descriptor) except OSError as exc: raise SessionIntegrityError("recorded media timeline is unavailable") from exc finally: os.close(parent_fd) def _read_media_index( path: Path, parent: Path, *, expected_count: int, ) -> tuple[list[dict[str, Any]], str]: """Stream a sealed JSONL index without imposing a recording-duration cap.""" try: resolved_parent = parent.resolve(strict=True) if path.parent.resolve(strict=True) != resolved_parent: raise SessionIntegrityError("recorded media index escapes its epoch") parent_fd = os.open( resolved_parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0), ) except OSError as exc: raise SessionIntegrityError("recorded media index is missing") from exc descriptor = -1 try: descriptor = os.open( path.name, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0), dir_fd=parent_fd, ) before = os.fstat(descriptor) if not stat.S_ISREG(before.st_mode) or before.st_size <= 0: raise SessionIntegrityError("recorded media index is not regular") entries: list[dict[str, Any]] = [] digest = hashlib.sha256() with os.fdopen(descriptor, "rb") as stream: descriptor = -1 for sequence in range(1, expected_count + 1): raw_line = stream.readline(MAX_MEDIA_INDEX_LINE_BYTES + 1) if not raw_line or len(raw_line) > MAX_MEDIA_INDEX_LINE_BYTES: raise SessionIntegrityError( "recorded media index line is missing or outside bounds" ) if not raw_line.endswith(b"\n"): raise SessionIntegrityError("recorded media index has an incomplete line") digest.update(raw_line) try: entry = json.loads(raw_line) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise SessionIntegrityError( "recorded media index contains invalid JSON" ) from exc if ( not isinstance(entry, dict) or entry.get("schema_version") != CAMERA_INDEX_SCHEMA or entry.get("sequence") != sequence or entry.get("kind") != "media" or entry.get("path") != f"segments/{sequence}.m4s" ): raise SessionIntegrityError("recorded media index entry is inconsistent") entries.append(entry) if stream.read(1): raise SessionIntegrityError( "recorded media index length does not match its summary" ) after = os.fstat(stream.fileno()) if (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns) != ( before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns, ): raise SessionIntegrityError("recorded media index changed during validation") return entries, digest.hexdigest() except OSError as exc: raise SessionIntegrityError("recorded media index is unavailable") from exc finally: if descriptor >= 0: os.close(descriptor) os.close(parent_fd) def _read_json_object(path: Path, maximum_bytes: int) -> dict[str, Any]: try: value = json.loads(_read_confined_file(path, path.parent, maximum_bytes)) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise SessionIntegrityError("recorded media summary is invalid") from exc if not isinstance(value, dict): raise SessionIntegrityError("recorded media summary is not an object") return value def _non_negative_int(value: object) -> TypeGuard[int]: return isinstance(value, int) and not isinstance(value, bool) and value >= 0 def _positive_int(value: object) -> TypeGuard[int]: return isinstance(value, int) and not isinstance(value, bool) and value > 0