1530 lines
58 KiB
Python
1530 lines
58 KiB
Python
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/v2"
|
|
RECORDED_MEDIA_PREPARATION_SCHEMA = "missioncore.recorded-media-preparation/v1"
|
|
MAX_MEDIA_INDEX_BYTES = 32 * 1024 * 1024
|
|
MAX_INIT_BYTES = 8 * 1024 * 1024
|
|
MAX_MEDIA_SEGMENTS = 500_000
|
|
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
|
|
MAX_PREPARED_MEDIA_SIDECAR_BYTES = 256 * 1024 * 1024
|
|
|
|
_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
|
|
|
|
|
|
@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
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class _Mp4VideoFragmentTiming:
|
|
base_decode_time: int
|
|
duration_units: int
|
|
|
|
@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:
|
|
payload = _read_confined_file(path, root, MAX_PREPARED_MEDIA_SIDECAR_BYTES)
|
|
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 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,
|
|
}
|
|
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 1 <= len(segment_documents) <= MAX_MEDIA_SEGMENTS
|
|
):
|
|
raise SessionIntegrityError("recorded media prepared segments are invalid")
|
|
segments_root = epoch_path / "segments"
|
|
segments: list[RecordedMediaSegment] = []
|
|
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")
|
|
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
|
|
):
|
|
raise SessionIntegrityError("recorded media prepared segment is invalid")
|
|
segments.append(
|
|
RecordedMediaSegment(
|
|
sequence=sequence,
|
|
path=segments_root / f"{sequence}.m4s",
|
|
byte_length=int(byte_length),
|
|
sha256=sha256,
|
|
)
|
|
)
|
|
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 0 < len(encoded) <= MAX_PREPARED_MEDIA_SIDECAR_BYTES:
|
|
raise SessionIntegrityError("recorded media preparation is outside bounds")
|
|
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,
|
|
}
|
|
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_INDEX_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_MEDIA_SEGMENTS:
|
|
raise SessionIntegrityError("recorded media segment count is invalid")
|
|
|
|
try:
|
|
raw_lines = (
|
|
_read_confined_file(
|
|
epoch / "index.jsonl",
|
|
epoch,
|
|
MAX_MEDIA_INDEX_BYTES,
|
|
)
|
|
.decode("utf-8")
|
|
.splitlines()
|
|
)
|
|
except UnicodeDecodeError as exc:
|
|
raise SessionIntegrityError("recorded media index is unavailable") from exc
|
|
if len(raw_lines) != int(segment_count):
|
|
raise SessionIntegrityError("recorded media index length does not match its summary")
|
|
entries: list[dict[str, Any]] = []
|
|
for sequence, line in enumerate(raw_lines, start=1):
|
|
try:
|
|
entry = json.loads(line)
|
|
except json.JSONDecodeError as exc:
|
|
raise SessionIntegrityError("recorded media index contains invalid JSON") from exc
|
|
if not isinstance(entry, dict):
|
|
raise SessionIntegrityError("recorded media index entry is not an object")
|
|
if (
|
|
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)
|
|
|
|
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")
|
|
segments: list[RecordedMediaSegment] = []
|
|
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")
|
|
segments.append(
|
|
RecordedMediaSegment(
|
|
sequence=sequence,
|
|
path=path,
|
|
byte_length=int(byte_length),
|
|
sha256=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] = []
|
|
for segment in segments:
|
|
payload = _read_confined_file(
|
|
segment.path,
|
|
segments_root,
|
|
segment.byte_length,
|
|
)
|
|
if hashlib.sha256(payload).hexdigest() != segment.sha256:
|
|
raise SessionIntegrityError("recorded media segment digest changed")
|
|
fragment_timings.append(
|
|
_mp4_video_fragment_timing(
|
|
payload,
|
|
timing,
|
|
_Mp4ParseBudget(),
|
|
)
|
|
)
|
|
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")
|
|
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, 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 = _parse_trex(child_payload)
|
|
if track_id in defaults:
|
|
raise SessionIntegrityError("recorded media init repeats a trex track")
|
|
defaults[track_id] = trex_default_duration
|
|
|
|
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_duration = defaults.get(track_id)
|
|
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
|
|
),
|
|
)
|
|
|
|
|
|
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 = _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
|
|
duration = sum(
|
|
_parse_trun_duration_units(trun, default_duration, budget) for trun in trun_payloads
|
|
)
|
|
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,
|
|
)
|
|
)
|
|
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]:
|
|
_full_box_version(payload)
|
|
return (
|
|
_read_u32(payload, 4, "trex track id"),
|
|
_read_u32(payload, 12, "trex default sample duration"),
|
|
)
|
|
|
|
|
|
def _parse_tfhd(payload: bytes) -> tuple[int, 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
|
|
for flag in (0x000010, 0x000020):
|
|
if flags & flag:
|
|
cursor = _advance_box_cursor(payload, cursor, 4, "tfhd optional field")
|
|
return track_id, default_duration if default_duration and default_duration > 0 else None
|
|
|
|
|
|
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_duration_units(
|
|
payload: bytes,
|
|
default_duration: int | None,
|
|
budget: _Mp4ParseBudget,
|
|
) -> 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")
|
|
if flags & 0x000004:
|
|
cursor = _advance_box_cursor(payload, cursor, 4, "trun first sample flags")
|
|
per_sample_width = sum(4 for flag in (0x000100, 0x000200, 0x000400, 0x000800) if flags & flag)
|
|
if per_sample_width and sample_count > (len(payload) - cursor) // per_sample_width:
|
|
raise SessionIntegrityError("recorded media trun samples are truncated")
|
|
if flags & 0x000100:
|
|
duration = 0
|
|
for _ in range(sample_count):
|
|
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 += per_sample_width
|
|
return duration
|
|
if default_duration is None or default_duration <= 0:
|
|
raise SessionIntegrityError("recorded media sample duration is unavailable")
|
|
if per_sample_width:
|
|
_advance_box_cursor(
|
|
payload,
|
|
cursor,
|
|
sample_count * per_sample_width,
|
|
"trun sample table",
|
|
)
|
|
return sample_count * default_duration
|
|
|
|
|
|
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_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
|