diff --git a/src/k1link/observatory/portable_artifact_transport.py b/src/k1link/observatory/portable_artifact_transport.py index db3ab41..d265812 100644 --- a/src/k1link/observatory/portable_artifact_transport.py +++ b/src/k1link/observatory/portable_artifact_transport.py @@ -16,11 +16,12 @@ import os import re import secrets import stat +import tarfile from collections.abc import AsyncIterable, Mapping, Sequence from contextlib import suppress from dataclasses import dataclass from pathlib import Path -from typing import Final, Literal +from typing import BinaryIO, Final, Literal, cast from k1link.observatory.portable_result_contract import ( OBSERVATION_ONLY_AUTHORITY, @@ -59,6 +60,16 @@ from k1link.sessions.store import SessionStore PORTABLE_SOURCE_MATERIALIZATION_SCHEMA: Final = ( "missioncore.observatory-portable-source-materialization/v1" ) +PORTABLE_CAMERA_EPOCH_ARCHIVE_SCHEMA: Final = ( + "missioncore.observatory-portable-camera-epoch-archive/v1" +) +PORTABLE_CAMERA_EPOCH_ARCHIVE_BINDING_SCHEMA: Final = ( + "missioncore.observatory-portable-camera-epoch-archive-binding/v1" +) +PORTABLE_CAMERA_EPOCH_ARCHIVE_MEDIA_TYPE: Final = "application/x-tar" +PORTABLE_CAMERA_EPOCH_ARCHIVE_ID_HEADER: Final = ( + "X-Mission-Core-Camera-Epoch-Archive-Id" +) PORTABLE_RESULT_UPLOAD_PLAN_SCHEMA: Final = ( "missioncore.observatory-portable-result-upload-plan/v1" ) @@ -73,6 +84,10 @@ PORTABLE_RESULT_STAGING_DIRECTORY: Final = "observatory-worker-result-staging" MAX_SOURCE_MEMBERS: Final = 100_000 MAX_SOURCE_BYTES: Final = 2 * 1024 * 1024 * 1024 * 1024 +MAX_CAMERA_EPOCH_ARCHIVE_BYTES: Final = ( + MAX_SOURCE_BYTES + MAX_SOURCE_MEMBERS * 1024 + tarfile.RECORDSIZE +) +MAX_CAMERA_EPOCH_ARCHIVE_BINDING_BYTES: Final = 8 * 1024 * 1024 MAX_RESULT_MEMBER_BYTES: Final = 64 * 1024 * 1024 * 1024 MAX_RESULT_PACKAGE_BYTES: Final = 256 * 1024 * 1024 * 1024 MAX_RESULT_MANIFEST_BYTES: Final = 1024 * 1024 @@ -228,6 +243,53 @@ class PortableSourceMaterializationManifest: } +@dataclass(frozen=True, slots=True) +class PortableCameraEpochArchive: + archive_id: str + job_identity_sha256: str + source_bundle_sha256: str + artifact_id: str + camera_epoch: int + member_ids: tuple[str, ...] + byte_length: int + sha256: str + + def __post_init__(self) -> None: + _digest(self.archive_id, "camera epoch archive id") + _digest(self.job_identity_sha256, "camera epoch archive job identity") + _digest(self.source_bundle_sha256, "camera epoch archive source bundle") + _session_id(self.artifact_id) + _positive_int(self.camera_epoch, "camera epoch archive ordinal") + if ( + not 2 <= len(self.member_ids) <= MAX_SOURCE_MEMBERS + or len(set(self.member_ids)) != len(self.member_ids) + ): + raise ValueError("camera epoch archive members are invalid") + for member_id in self.member_ids: + _digest(member_id, "camera epoch archive member id") + if not 1 <= self.byte_length <= MAX_CAMERA_EPOCH_ARCHIVE_BYTES: + raise ValueError("camera epoch archive byte length is invalid") + _digest(self.sha256, "camera epoch archive sha256") + if self.archive_id != portable_camera_epoch_archive_id( + job_identity_sha256=self.job_identity_sha256, + source_bundle_sha256=self.source_bundle_sha256, + artifact_id=self.artifact_id, + camera_epoch=self.camera_epoch, + member_ids=self.member_ids, + byte_length=self.byte_length, + sha256=self.sha256, + ): + raise ValueError("camera epoch archive identity is invalid") + + def as_dict(self) -> dict[str, object]: + return { + "archive_id": self.archive_id, + "media_type": PORTABLE_CAMERA_EPOCH_ARCHIVE_MEDIA_TYPE, + "byte_length": self.byte_length, + "sha256": self.sha256, + } + + @dataclass(frozen=True, slots=True) class PortableResultUploadMember: member_id: str @@ -443,6 +505,39 @@ class PortableObservatoryArtifactTransport: ) return member, destination + def materialize_camera_epoch_archive( + self, + *, + job_id: str, + claim_token: str, + claim_generation: int, + claimant_id: str, + ) -> tuple[PortableCameraEpochArchive, Path]: + """Publish one claim-bound deterministic tar for the admitted camera epoch.""" + + job = self._authorize( + job_id=job_id, + claim_token=claim_token, + claim_generation=claim_generation, + claimant_id=claimant_id, + allowed_states=("claimed", "running"), + ) + manifest = self._resolve_source_manifest(job) + members = _ordered_camera_epoch_members(manifest.members) + archive, path = _publish_camera_epoch_archive( + self._source_cas, + job=job, + members=members, + ) + self._authorize( + job_id=job_id, + claim_token=claim_token, + claim_generation=claim_generation, + claimant_id=claimant_id, + allowed_states=("claimed", "running"), + ) + return archive, path + def stage_result_manifest( self, *, @@ -1033,6 +1128,418 @@ class PortableObservatoryArtifactTransport: return receipt +def portable_camera_epoch_archive_id( + *, + job_identity_sha256: str, + source_bundle_sha256: str, + artifact_id: str, + camera_epoch: int, + member_ids: Sequence[str], + byte_length: int, + sha256: str, +) -> str: + """Return the fixed identity of one packed, content-sealed camera epoch.""" + + _digest(job_identity_sha256, "camera epoch archive job identity") + _digest(source_bundle_sha256, "camera epoch archive source bundle") + _session_id(artifact_id) + _positive_int(camera_epoch, "camera epoch archive ordinal") + canonical_member_ids = tuple(member_ids) + if ( + not 2 <= len(canonical_member_ids) <= MAX_SOURCE_MEMBERS + or len(set(canonical_member_ids)) != len(canonical_member_ids) + ): + raise ValueError("camera epoch archive members are invalid") + for member_id in canonical_member_ids: + _digest(member_id, "camera epoch archive member id") + if not 1 <= byte_length <= MAX_CAMERA_EPOCH_ARCHIVE_BYTES: + raise ValueError("camera epoch archive byte length is invalid") + _digest(sha256, "camera epoch archive sha256") + return hashlib.sha256( + canonical_json( + { + "schema_version": PORTABLE_CAMERA_EPOCH_ARCHIVE_SCHEMA, + "job_identity_sha256": job_identity_sha256, + "source_bundle_sha256": source_bundle_sha256, + "artifact_id": artifact_id, + "camera_epoch": camera_epoch, + "member_ids": list(canonical_member_ids), + "media_type": PORTABLE_CAMERA_EPOCH_ARCHIVE_MEDIA_TYPE, + "byte_length": byte_length, + "sha256": sha256, + } + ) + ).hexdigest() + + +def _ordered_camera_epoch_members( + members: Sequence[PortableSourceMember], +) -> tuple[PortableSourceMember, ...]: + inits = tuple(member for member in members if member.kind == "camera-init") + segments = tuple(member for member in members if member.kind == "camera-segment") + if len(inits) != 1 or not segments: + raise PortableArtifactTransportIntegrityError( + "portable camera epoch member inventory is incomplete" + ) + init = inits[0] + if init.artifact_id is None or init.camera_epoch is None: + raise PortableArtifactTransportIntegrityError( + "portable camera epoch identity is incomplete" + ) + ordered_segments = tuple( + sorted(segments, key=lambda member: member.camera_sequence or 0) + ) + if tuple(member.camera_sequence for member in ordered_segments) != tuple( + range(1, len(ordered_segments) + 1) + ) or any( + member.artifact_id != init.artifact_id + or member.camera_epoch != init.camera_epoch + for member in ordered_segments + ): + raise PortableArtifactTransportIntegrityError( + "portable camera epoch member order changed" + ) + return (init, *ordered_segments) + + +def _camera_archive_relative_name(member: PortableSourceMember) -> str: + if member.kind == "camera-init": + return "init.mp4" + if member.kind != "camera-segment" or member.camera_sequence is None: + raise PortableArtifactTransportIntegrityError( + "portable camera archive member role is invalid" + ) + return f"segments/{member.camera_sequence}.m4s" + + +def _camera_epoch_archive_byte_length( + members: Sequence[PortableSourceMember], +) -> int: + content_bytes = sum( + 512 + ((member.byte_length + 511) // 512) * 512 for member in members + ) + logical_bytes = content_bytes + 1024 + return ((logical_bytes + tarfile.RECORDSIZE - 1) // tarfile.RECORDSIZE) * ( + tarfile.RECORDSIZE + ) + + +def _camera_epoch_archive_binding_identity( + job: ObservatoryRecordedJob, + members: Sequence[PortableSourceMember], +) -> dict[str, object]: + ordered = _ordered_camera_epoch_members(members) + init = ordered[0] + assert init.artifact_id is not None + assert init.camera_epoch is not None + return { + "schema_version": PORTABLE_CAMERA_EPOCH_ARCHIVE_BINDING_SCHEMA, + "job_identity_sha256": job.identity_sha256, + "source_bundle_sha256": job.source_bundle_sha256, + "artifact_id": init.artifact_id, + "camera_epoch": init.camera_epoch, + "member_ids": [member.member_id for member in ordered], + } + + +def _load_camera_epoch_archive_binding( + path: Path, + *, + binding_id: str, + identity: Mapping[str, object], + archive_root: Path, +) -> tuple[PortableCameraEpochArchive, Path]: + try: + payload = _read_exact_regular_file( + path, + maximum_bytes=MAX_CAMERA_EPOCH_ARCHIVE_BINDING_BYTES, + ) + document = object_document( + json.loads(payload.decode("utf-8")), + "camera epoch archive binding", + ) + if canonical_json(document) != payload: + raise PortableArtifactTransportIntegrityError( + "camera epoch archive binding is not canonical" + ) + except PortableArtifactTransportError: + raise + except ( + OSError, + UnicodeDecodeError, + json.JSONDecodeError, + PortableResultPackageIntegrityError, + ValueError, + ) as exc: + raise PortableArtifactTransportIntegrityError( + "camera epoch archive binding is unreadable" + ) from exc + if ( + set(document) != {*identity, "binding_id", "archive", "authority"} + or any(document.get(key) != value for key, value in identity.items()) + or document.get("binding_id") != binding_id + or document.get("authority") != OBSERVATION_ONLY_AUTHORITY + ): + raise PortableArtifactTransportIntegrityError( + "camera epoch archive binding identity changed" + ) + try: + archive_document = object_document( + document.get("archive"), + "camera epoch archive", + ) + except PortableResultPackageIntegrityError as exc: + raise PortableArtifactTransportIntegrityError( + "camera epoch archive seal is invalid" + ) from exc + if ( + set(archive_document) + != {"archive_id", "media_type", "byte_length", "sha256"} + or archive_document.get("media_type") + != PORTABLE_CAMERA_EPOCH_ARCHIVE_MEDIA_TYPE + ): + raise PortableArtifactTransportIntegrityError( + "camera epoch archive seal changed" + ) + try: + member_ids_value = identity["member_ids"] + if not isinstance(member_ids_value, list) or any( + not isinstance(member_id, str) for member_id in member_ids_value + ): + raise ValueError("camera epoch archive member ids are invalid") + result = PortableCameraEpochArchive( + archive_id=string(archive_document.get("archive_id"), "archive id"), + job_identity_sha256=string( + identity["job_identity_sha256"], + "archive job identity", + ), + source_bundle_sha256=string( + identity["source_bundle_sha256"], + "archive source bundle", + ), + artifact_id=string(identity["artifact_id"], "archive artifact id"), + camera_epoch=_positive_int( + identity["camera_epoch"], + "archive camera epoch", + ), + member_ids=tuple(member_ids_value), + byte_length=_positive_int( + archive_document.get("byte_length"), + "archive byte length", + ), + sha256=string(archive_document.get("sha256"), "archive sha256"), + ) + except (PortableResultPackageIntegrityError, ValueError) as exc: + raise PortableArtifactTransportIntegrityError( + "camera epoch archive binding is invalid" + ) from exc + archive_path = archive_root / result.sha256[:2] / result.sha256 + if not _matches_exact_file( + archive_path, + expected_sha256=result.sha256, + expected_byte_length=result.byte_length, + ): + raise PortableArtifactTransportIntegrityError( + "camera epoch archive differs from its immutable binding" + ) + return result, archive_path + + +def _publish_camera_epoch_archive( + root: Path, + *, + job: ObservatoryRecordedJob, + members: Sequence[PortableSourceMember], +) -> tuple[PortableCameraEpochArchive, Path]: + ordered = _ordered_camera_epoch_members(members) + init = ordered[0] + assert init.artifact_id is not None + assert init.camera_epoch is not None + archive_root = _prepare_secure_root(root / "camera-epoch-archives") + binding_identity = _camera_epoch_archive_binding_identity(job, ordered) + binding_id = hashlib.sha256(canonical_json(binding_identity)).hexdigest() + binding_root = _prepare_secure_root(archive_root / "bindings") + binding_path = binding_root / f"{binding_id}.json" + if binding_path.exists(): + return _load_camera_epoch_archive_binding( + binding_path, + binding_id=binding_id, + identity=binding_identity, + archive_root=archive_root, + ) + for member in ordered: + if member._source_path is None: + raise PortableArtifactTransportIntegrityError( + "portable camera archive member has no admitted source" + ) + temporary = archive_root / f".tmp-{secrets.token_hex(16)}" + descriptor = os.open( + temporary, + os.O_WRONLY + | os.O_CREAT + | os.O_EXCL + | getattr(os, "O_NOFOLLOW", 0), + 0o600, + ) + try: + with os.fdopen(descriptor, "wb") as stream: + descriptor = -1 + with tarfile.open( + fileobj=stream, + mode="w", + format=tarfile.USTAR_FORMAT, + ) as archive: + for member in ordered: + info = tarfile.TarInfo(_camera_archive_relative_name(member)) + info.size = member.byte_length + info.mode = 0o600 + info.mtime = 0 + info.uid = 0 + info.gid = 0 + info.uname = "" + info.gname = "" + assert member._source_path is not None + _add_exact_camera_archive_member( + archive, + info, + member._source_path, + expected_sha256=member.sha256, + expected_byte_length=member.byte_length, + ) + stream.flush() + os.fsync(stream.fileno()) + sha256, byte_length = _hash_regular_file( + temporary, + maximum_bytes=MAX_CAMERA_EPOCH_ARCHIVE_BYTES, + ) + if byte_length != _camera_epoch_archive_byte_length(ordered): + raise PortableArtifactTransportIntegrityError( + "portable camera epoch archive size is not deterministic" + ) + result = PortableCameraEpochArchive( + archive_id=portable_camera_epoch_archive_id( + job_identity_sha256=job.identity_sha256, + source_bundle_sha256=job.source_bundle_sha256, + artifact_id=init.artifact_id, + camera_epoch=init.camera_epoch, + member_ids=tuple(member.member_id for member in ordered), + byte_length=byte_length, + sha256=sha256, + ), + job_identity_sha256=job.identity_sha256, + source_bundle_sha256=job.source_bundle_sha256, + artifact_id=init.artifact_id, + camera_epoch=init.camera_epoch, + member_ids=tuple(member.member_id for member in ordered), + byte_length=byte_length, + sha256=sha256, + ) + destination_root = _prepare_secure_root( + archive_root / result.sha256[:2] + ) + destination = destination_root / result.sha256 + _publish_uploaded_file( + temporary, + destination, + expected_sha256=result.sha256, + expected_byte_length=result.byte_length, + ) + _write_immutable_file( + binding_path, + canonical_json( + { + **binding_identity, + "binding_id": binding_id, + "archive": result.as_dict(), + "authority": dict(OBSERVATION_ONLY_AUTHORITY), + } + ), + ) + return result, destination + except PortableArtifactTransportError: + raise + except (OSError, tarfile.TarError, ValueError) as exc: + raise PortableArtifactTransportUnavailableError( + "portable camera epoch archive could not be materialized" + ) from exc + finally: + if descriptor >= 0: + os.close(descriptor) + with suppress(FileNotFoundError): + temporary.unlink() + + +class _ExactArchiveSource: + def __init__(self, stream: BinaryIO, *, expected_byte_length: int) -> None: + self._stream = stream + self._expected_byte_length = expected_byte_length + self._digest = hashlib.sha256() + self.byte_length = 0 + + def read(self, size: int = -1) -> bytes: + payload = self._stream.read(size) + self.byte_length += len(payload) + if self.byte_length > self._expected_byte_length: + raise PortableArtifactTransportIntegrityError( + "portable camera archive source grew while read" + ) + if not payload and self.byte_length != self._expected_byte_length: + raise PortableArtifactTransportIntegrityError( + "portable camera archive source ended early" + ) + self._digest.update(payload) + return payload + + @property + def sha256(self) -> str: + return self._digest.hexdigest() + + +def _add_exact_camera_archive_member( + archive: tarfile.TarFile, + info: tarfile.TarInfo, + source: Path, + *, + expected_sha256: str, + expected_byte_length: int, +) -> None: + descriptor = os.open( + source, + os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0), + ) + with os.fdopen(descriptor, "rb") as source_stream: + before = os.fstat(source_stream.fileno()) + if not stat.S_ISREG(before.st_mode) or before.st_size != expected_byte_length: + raise PortableArtifactTransportIntegrityError( + "portable camera archive source is not the admitted regular file" + ) + verified = _ExactArchiveSource( + source_stream, + expected_byte_length=expected_byte_length, + ) + archive.addfile(info, cast(BinaryIO, verified)) + after = os.fstat(source_stream.fileno()) + stable = ( + before.st_dev, + before.st_ino, + before.st_size, + before.st_mtime_ns, + ) == ( + after.st_dev, + after.st_ino, + after.st_size, + after.st_mtime_ns, + ) + if ( + verified.byte_length != expected_byte_length + or verified.sha256 != expected_sha256 + or not stable + ): + raise PortableArtifactTransportIntegrityError( + "portable camera archive source changed while read" + ) + + def _source_member( job: ObservatoryRecordedJob, *, diff --git a/src/k1link/observatory/worker_http_transport.py b/src/k1link/observatory/worker_http_transport.py index bab5c48..7d35afc 100644 --- a/src/k1link/observatory/worker_http_transport.py +++ b/src/k1link/observatory/worker_http_transport.py @@ -14,21 +14,28 @@ import json import os import re import secrets +import shutil import stat +import tarfile +import tempfile import threading from collections.abc import Mapping from contextlib import suppress from dataclasses import dataclass from pathlib import Path -from typing import Final, Literal, cast +from typing import BinaryIO, Final, Literal, cast from urllib.parse import urlsplit import httpx from k1link.observatory.portable_artifact_transport import ( + MAX_CAMERA_EPOCH_ARCHIVE_BYTES, + PORTABLE_CAMERA_EPOCH_ARCHIVE_ID_HEADER, + PORTABLE_CAMERA_EPOCH_ARCHIVE_MEDIA_TYPE, PORTABLE_RESULT_UPLOAD_PLAN_SCHEMA, PORTABLE_RESULT_UPLOAD_RECEIPT_SCHEMA, PORTABLE_SOURCE_MATERIALIZATION_SCHEMA, + portable_camera_epoch_archive_id, ) from k1link.observatory.portable_result_contract import ( OBSERVATION_ONLY_AUTHORITY, @@ -54,6 +61,7 @@ from k1link.observatory.worker_agent import ( WORKER_HTTP_MAX_JSON_BYTES: Final = 16 * 1024 * 1024 WORKER_HTTP_COPY_CHUNK_BYTES: Final = 1024 * 1024 WORKER_HTTP_DEFAULT_TIMEOUT_SECONDS: Final = 30.0 +WORKER_HTTP_CAMERA_ARCHIVE_READ_TIMEOUT_SECONDS: Final = 30 * 60.0 _SHA256 = re.compile(r"^[a-f0-9]{64}$") _JOB_ID = re.compile(r"^observatory-run-[a-f0-9]{32}$") @@ -136,6 +144,7 @@ class ObservatoryWorkerHttpGateway(ObservatoryWorkerTransport): _CONTOUR_HEADER: contour_id, "Accept": "application/json", } + self._timeout_seconds = timeout_seconds self._work_root = _secure_directory(work_root) self._client = httpx.Client( base_url=self._base_url, @@ -304,6 +313,21 @@ class ObservatoryWorkerHttpGateway(ObservatoryWorkerTransport): "source materialization members select the same local role" ) destinations[destination] = member + camera_members = _ordered_camera_epoch_members(members) + if not all( + _matches_file( + _source_destination(root, member), + member.sha256, + member.byte_length, + ) + for member in camera_members + ): + self._download_camera_epoch_archive( + job=job, + context=context, + members=camera_members, + root=root, + ) for destination, member in destinations.items(): if _matches_file(destination, member.sha256, member.byte_length): continue @@ -463,6 +487,123 @@ class ObservatoryWorkerHttpGateway(ObservatoryWorkerTransport): result_sha256=draft.result_sha256, ) + def _download_camera_epoch_archive( + self, + *, + job: SealedObservatoryRecordedJob, + context: _ClaimContext, + members: tuple[_SourceMember, ...], + root: Path, + ) -> None: + transfer_root = _secure_directory(self._work_root / ".source-transfers") + temporary = transfer_root / f".camera-epoch-{secrets.token_hex(16)}.tar" + descriptor = -1 + try: + with self._client.stream( + "GET", + self._job_path(job.job_id, "source-camera-epoch-archive"), + headers=self._claim_headers(context), + timeout=httpx.Timeout( + connect=self._timeout_seconds, + read=WORKER_HTTP_CAMERA_ARCHIVE_READ_TIMEOUT_SECONDS, + write=self._timeout_seconds, + pool=self._timeout_seconds, + ), + ) as response: + if response.status_code == 404: + return + self._raise_for_status(response) + if ( + response.headers.get("content-type") + != PORTABLE_CAMERA_EPOCH_ARCHIVE_MEDIA_TYPE + ): + raise ObservatoryWorkerHttpError( + "camera epoch archive media type changed" + ) + declared = response.headers.get("content-length") + if ( + declared is None + or not declared.isascii() + or not declared.isdigit() + or len(declared) > len(str(MAX_CAMERA_EPOCH_ARCHIVE_BYTES)) + or str(int(declared)) != declared + ): + raise ObservatoryWorkerHttpError( + "camera epoch archive length is invalid" + ) + expected_bytes = int(declared) + expected_sha256 = response.headers.get(_CONTENT_SHA_HEADER, "") + archive_id = response.headers.get( + PORTABLE_CAMERA_EPOCH_ARCHIVE_ID_HEADER, + "", + ) + if ( + not 1 <= expected_bytes <= MAX_CAMERA_EPOCH_ARCHIVE_BYTES + or _SHA256.fullmatch(expected_sha256) is None + or _SHA256.fullmatch(archive_id) is None + ): + raise ObservatoryWorkerHttpError( + "camera epoch archive seal is invalid" + ) + descriptor = os.open( + temporary, + os.O_WRONLY + | os.O_CREAT + | os.O_EXCL + | getattr(os, "O_NOFOLLOW", 0), + 0o600, + ) + digest = hashlib.sha256() + byte_length = 0 + with os.fdopen(descriptor, "wb") as stream: + descriptor = -1 + for chunk in response.iter_bytes(WORKER_HTTP_COPY_CHUNK_BYTES): + byte_length += len(chunk) + if byte_length > expected_bytes: + raise ObservatoryWorkerHttpError( + "camera epoch archive exceeds its declared length" + ) + digest.update(chunk) + stream.write(chunk) + stream.flush() + os.fsync(stream.fileno()) + if byte_length != expected_bytes or digest.hexdigest() != expected_sha256: + raise ObservatoryWorkerHttpError( + "camera epoch archive content differs from its seal" + ) + init = members[0] + assert init.artifact_id is not None + assert init.camera_epoch is not None + expected_archive_id = portable_camera_epoch_archive_id( + job_identity_sha256=job.identity_sha256, + source_bundle_sha256=job.source_bundle_sha256, + artifact_id=init.artifact_id, + camera_epoch=init.camera_epoch, + member_ids=tuple(member.member_id for member in members), + byte_length=byte_length, + sha256=expected_sha256, + ) + if archive_id != expected_archive_id: + raise ObservatoryWorkerHttpError( + "camera epoch archive identity changed" + ) + _extract_camera_epoch_archive( + temporary, + root=root, + members=members, + ) + except ObservatoryWorkerHttpError: + raise + except httpx.HTTPError as exc: + raise ObservatoryWorkerHttpError( + "camera epoch archive transport is unavailable" + ) from exc + finally: + if descriptor >= 0: + os.close(descriptor) + with suppress(FileNotFoundError): + temporary.unlink() + def _download_member( self, *, @@ -868,6 +1009,192 @@ def _source_member(value: object) -> _SourceMember: ) +def _ordered_camera_epoch_members( + members: tuple[_SourceMember, ...], +) -> tuple[_SourceMember, ...]: + inits = tuple(member for member in members if member.kind == "camera-init") + segments = tuple(member for member in members if member.kind == "camera-segment") + if len(inits) != 1 or not segments: + raise ObservatoryWorkerHttpError( + "source materialization camera epoch is incomplete" + ) + init = inits[0] + if init.artifact_id is None or init.camera_epoch is None: + raise ObservatoryWorkerHttpError( + "source materialization camera identity is incomplete" + ) + ordered_segments = tuple( + sorted(segments, key=lambda member: member.camera_sequence or 0) + ) + if tuple(member.camera_sequence for member in ordered_segments) != tuple( + range(1, len(ordered_segments) + 1) + ) or any( + member.artifact_id != init.artifact_id + or member.camera_epoch != init.camera_epoch + for member in ordered_segments + ): + raise ObservatoryWorkerHttpError( + "source materialization camera member order changed" + ) + return (init, *ordered_segments) + + +def _camera_archive_relative_name(member: _SourceMember) -> str: + if member.kind == "camera-init": + return "init.mp4" + if member.kind != "camera-segment" or member.camera_sequence is None: + raise ObservatoryWorkerHttpError("camera archive member role is invalid") + return f"segments/{member.camera_sequence}.m4s" + + +def _camera_epoch_archive_byte_length(members: tuple[_SourceMember, ...]) -> int: + content_bytes = sum( + 512 + ((member.byte_length + 511) // 512) * 512 for member in members + ) + logical_bytes = content_bytes + 1024 + return ((logical_bytes + tarfile.RECORDSIZE - 1) // tarfile.RECORDSIZE) * ( + tarfile.RECORDSIZE + ) + + +def _extract_camera_epoch_archive( + archive_path: Path, + *, + root: Path, + members: tuple[_SourceMember, ...], +) -> None: + expected_bytes = _camera_epoch_archive_byte_length(members) + try: + archive_metadata = archive_path.lstat() + except OSError as exc: + raise ObservatoryWorkerHttpError( + "camera epoch archive is unavailable" + ) from exc + if ( + stat.S_ISLNK(archive_metadata.st_mode) + or not stat.S_ISREG(archive_metadata.st_mode) + or archive_metadata.st_size != expected_bytes + ): + raise ObservatoryWorkerHttpError( + "camera epoch archive size differs from its member inventory" + ) + transfer_root = _secure_directory(archive_path.parent) + staging = Path( + tempfile.mkdtemp(prefix=".camera-epoch-extract-", dir=transfer_root) + ) + staged: list[Path] = [] + archive_descriptor = -1 + try: + archive_descriptor = os.open( + archive_path, + os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0), + ) + with os.fdopen(archive_descriptor, "rb") as archive_stream: + archive_descriptor = -1 + with tarfile.open(fileobj=archive_stream, mode="r:") as archive: + for member in members: + info = archive.next() + if info is None: + raise ObservatoryWorkerHttpError( + "camera epoch archive member inventory changed" + ) + relative_name = _camera_archive_relative_name(member) + if ( + info.name != relative_name + or info.type != tarfile.REGTYPE + or info.size != member.byte_length + or info.mode != 0o600 + or info.mtime != 0 + or info.uid != 0 + or info.gid != 0 + or info.uname != "" + or info.gname != "" + or info.linkname != "" + or info.pax_headers + ): + raise ObservatoryWorkerHttpError( + "camera epoch archive contains an unsafe member" + ) + source = archive.extractfile(info) + if source is None: + raise ObservatoryWorkerHttpError( + "camera epoch archive member is unreadable" + ) + target = staging.joinpath(*relative_name.split("/")) + _secure_directory(target.parent) + with source: + _copy_camera_archive_member( + cast(BinaryIO, source), + target, + expected_sha256=member.sha256, + expected_byte_length=member.byte_length, + ) + staged.append(target) + if archive.next() is not None: + raise ObservatoryWorkerHttpError( + "camera epoch archive member inventory changed" + ) + for temporary, member in zip(staged, members, strict=True): + _publish_local_file( + temporary, + _source_destination(root, member), + member.sha256, + member.byte_length, + ) + except ObservatoryWorkerHttpError: + raise + except (OSError, tarfile.TarError) as exc: + raise ObservatoryWorkerHttpError( + "camera epoch archive could not be extracted" + ) from exc + finally: + if archive_descriptor >= 0: + os.close(archive_descriptor) + shutil.rmtree(staging, ignore_errors=True) + + +def _copy_camera_archive_member( + source: BinaryIO, + destination: Path, + *, + expected_sha256: str, + expected_byte_length: int, +) -> None: + descriptor = os.open( + destination, + os.O_WRONLY + | os.O_CREAT + | os.O_EXCL + | getattr(os, "O_NOFOLLOW", 0), + 0o600, + ) + digest = hashlib.sha256() + byte_length = 0 + try: + with os.fdopen(descriptor, "wb") as stream: + descriptor = -1 + while chunk := source.read(WORKER_HTTP_COPY_CHUNK_BYTES): + byte_length += len(chunk) + if byte_length > expected_byte_length: + raise ObservatoryWorkerHttpError( + "camera epoch archive member exceeds its declared length" + ) + digest.update(chunk) + stream.write(chunk) + stream.flush() + os.fsync(stream.fileno()) + if byte_length != expected_byte_length or digest.hexdigest() != expected_sha256: + raise ObservatoryWorkerHttpError( + "camera epoch archive member content changed" + ) + finally: + if descriptor >= 0: + os.close(descriptor) + if not _matches_file(destination, expected_sha256, expected_byte_length): + with suppress(OSError): + destination.unlink() + + def _source_destination(root: Path, member: _SourceMember) -> Path: if member.kind == "source-bundle": return root / "source-bundle.json" diff --git a/src/k1link/web/observatory_worker_api.py b/src/k1link/web/observatory_worker_api.py index 8878923..21913a1 100644 --- a/src/k1link/web/observatory_worker_api.py +++ b/src/k1link/web/observatory_worker_api.py @@ -26,6 +26,8 @@ from pydantic import BaseModel, ConfigDict, Field from k1link.observatory.portable_artifact_transport import ( MAX_RESULT_MANIFEST_BYTES, + PORTABLE_CAMERA_EPOCH_ARCHIVE_ID_HEADER, + PORTABLE_CAMERA_EPOCH_ARCHIVE_MEDIA_TYPE, PortableArtifactTransportError, PortableArtifactTransportIntegrityError, PortableArtifactTransportUnavailableError, @@ -381,6 +383,41 @@ def build_observatory_worker_router( .as_dict() ) + @router.get("/recorded-jobs/{job_id}/source-camera-epoch-archive") + def source_camera_epoch_archive( + job_id: Annotated[str, ApiPath(pattern=_JOB_ID_PATTERN)], + claim_token: Annotated[ + str, + Header( + alias=OBSERVATORY_WORKER_CLAIM_TOKEN_HEADER, + pattern=_CLAIM_TOKEN_PATTERN, + ), + ], + claim_generation: Annotated[ + int, + Header(alias=OBSERVATORY_WORKER_CLAIM_GENERATION_HEADER, ge=1), + ], + ) -> FileResponse: + archive, path = _artifact_call( + lambda: artifact_transport.materialize_camera_epoch_archive( + job_id=job_id, + claim_token=claim_token, + claim_generation=claim_generation, + claimant_id=authentication.contour_id, + ) + ) + return FileResponse( + path, + media_type=PORTABLE_CAMERA_EPOCH_ARCHIVE_MEDIA_TYPE, + headers={ + "ETag": f'"{archive.sha256}"', + "Cache-Control": "private, no-store", + "X-Content-Type-Options": "nosniff", + "X-Mission-Core-Content-Sha256": archive.sha256, + PORTABLE_CAMERA_EPOCH_ARCHIVE_ID_HEADER: archive.archive_id, + }, + ) + @router.get("/recorded-jobs/{job_id}/source-members/{member_id}") def source_member( job_id: Annotated[str, ApiPath(pattern=_JOB_ID_PATTERN)], diff --git a/tests/test_observatory_portable_artifact_transport.py b/tests/test_observatory_portable_artifact_transport.py index b495343..2a61935 100644 --- a/tests/test_observatory_portable_artifact_transport.py +++ b/tests/test_observatory_portable_artifact_transport.py @@ -4,6 +4,7 @@ import asyncio import copy import hashlib import json +import tarfile from dataclasses import dataclass from pathlib import Path from typing import cast @@ -484,6 +485,51 @@ def test_source_member_rejects_tampering_and_stale_generation(tmp_path: Path) -> ) +def test_camera_epoch_archive_is_sealed_deterministic_and_cached( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + fixture = _fixture(tmp_path) + + archive, archive_path = fixture.service.materialize_camera_epoch_archive( + job_id=fixture.job.job_id, + claim_token=fixture.claim_token, + claim_generation=fixture.job.claim_generation, + claimant_id="worker-006", + ) + + assert archive_path.name == archive.sha256 + assert archive_path.stat().st_size == archive.byte_length + assert hashlib.sha256(archive_path.read_bytes()).hexdigest() == archive.sha256 + with tarfile.open(archive_path, mode="r:") as packed: + members = packed.getmembers() + assert [member.name for member in members] == ["init.mp4", "segments/1.m4s"] + assert all( + member.type == tarfile.REGTYPE + and member.mode == 0o600 + and member.mtime == 0 + and member.uid == 0 + and member.gid == 0 + and member.uname == "" + and member.gname == "" + for member in members + ) + + def reject_rebuild(*_args: object, **_kwargs: object) -> object: + pytest.fail("immutable camera epoch archive must be reused") + + monkeypatch.setattr(tarfile, "open", reject_rebuild) + repeated, repeated_path = fixture.service.materialize_camera_epoch_archive( + job_id=fixture.job.job_id, + claim_token=fixture.claim_token, + claim_generation=fixture.job.claim_generation, + claimant_id="worker-006", + ) + + assert repeated == archive + assert repeated_path == archive_path + + def test_result_upload_is_atomic_resumable_and_required_before_success( tmp_path: Path, ) -> None: diff --git a/tests/test_observatory_worker_http_transport.py b/tests/test_observatory_worker_http_transport.py index 37bd360..0ed644b 100644 --- a/tests/test_observatory_worker_http_transport.py +++ b/tests/test_observatory_worker_http_transport.py @@ -1,7 +1,9 @@ from __future__ import annotations import hashlib +import io import json +import tarfile from collections.abc import Callable from pathlib import Path @@ -9,9 +11,12 @@ import httpx import pytest from k1link.observatory.portable_artifact_transport import ( + PORTABLE_CAMERA_EPOCH_ARCHIVE_ID_HEADER, + PORTABLE_CAMERA_EPOCH_ARCHIVE_MEDIA_TYPE, PORTABLE_RESULT_UPLOAD_PLAN_SCHEMA, PORTABLE_RESULT_UPLOAD_RECEIPT_SCHEMA, PORTABLE_SOURCE_MATERIALIZATION_SCHEMA, + portable_camera_epoch_archive_id, ) from k1link.observatory.portable_result_contract import ( OBSERVATION_ONLY_AUTHORITY, @@ -29,6 +34,8 @@ from k1link.observatory.worker_agent import ( SealedObservatoryRecordedJob, ) from k1link.observatory.worker_http_transport import ( + WORKER_HTTP_CAMERA_ARCHIVE_READ_TIMEOUT_SECONDS, + WORKER_HTTP_DEFAULT_TIMEOUT_SECONDS, ObservatoryWorkerHttpError, ObservatoryWorkerHttpGateway, ) @@ -138,7 +145,10 @@ def _source_contract( job: SealedObservatoryRecordedJob, *, inject_path: bool = False, + camera_segment_count: int = 1, ) -> tuple[dict[str, object], dict[str, bytes]]: + if camera_segment_count < 1: + raise ValueError("camera segment count must be positive") rows = [ _source_member( job, @@ -175,16 +185,23 @@ def _source_contract( camera_epoch=1, media_type='video/mp4; codecs="avc1.641028"', ), + ] + rows.extend( _source_member( job, kind="camera-segment", - payload=b"sealed-camera-segment", + payload=( + b"sealed-camera-segment" + if sequence == 1 + else f"sealed-camera-segment-{sequence}".encode("ascii") + ), artifact_id="recorded-camera-right", camera_epoch=1, - camera_sequence=1, + camera_sequence=sequence, media_type="video/iso.segment", - ), - ] + ) + for sequence in range(1, camera_segment_count + 1) + ) members = [row for row, _payload in rows] members.sort(key=lambda row: str(row["member_id"])) if inject_path: @@ -210,6 +227,86 @@ def _source_contract( ) +def _camera_archive_response( + job: SealedObservatoryRecordedJob, + manifest: dict[str, object], + payloads: dict[str, bytes], + *, + mutation: str | None = None, +) -> tuple[bytes, dict[str, str]]: + rows = manifest["members"] + assert isinstance(rows, list) + camera_rows = [ + row + for row in rows + if isinstance(row, dict) + and row.get("kind") in {"camera-init", "camera-segment"} + ] + camera_rows.sort( + key=lambda row: ( + 0 if row["kind"] == "camera-init" else 1, + int(row["camera_sequence"] or 0), + ) + ) + packed = io.BytesIO() + with tarfile.open( + fileobj=packed, + mode="w", + format=tarfile.USTAR_FORMAT, + ) as archive: + for index, row in enumerate(camera_rows): + sequence = row["camera_sequence"] + name = ( + "init.mp4" + if row["kind"] == "camera-init" + else f"segments/{sequence}.m4s" + ) + if mutation == "traversal" and index == len(camera_rows) - 1: + name = "../../operator-secret" + member_id = str(row["member_id"]) + payload = payloads[member_id] + info = tarfile.TarInfo(name) + info.mode = 0o600 + info.mtime = 0 + info.uid = 0 + info.gid = 0 + info.uname = "" + info.gname = "" + if mutation == "symlink" and index == len(camera_rows) - 1: + info.type = tarfile.SYMTYPE + info.linkname = "../../operator-secret" + info.size = 0 + archive.addfile(info) + else: + info.size = len(payload) + archive.addfile(info, io.BytesIO(payload)) + archive_payload = packed.getvalue() + sha256 = hashlib.sha256(archive_payload).hexdigest() + init = camera_rows[0] + archive_id = portable_camera_epoch_archive_id( + job_identity_sha256=job.identity_sha256, + source_bundle_sha256=job.source_bundle_sha256, + artifact_id=str(init["artifact_id"]), + camera_epoch=int(init["camera_epoch"]), + member_ids=tuple(str(row["member_id"]) for row in camera_rows), + byte_length=len(archive_payload), + sha256=sha256, + ) + headers = { + "Content-Type": PORTABLE_CAMERA_EPOCH_ARCHIVE_MEDIA_TYPE, + "Content-Length": str(len(archive_payload)), + "X-Mission-Core-Content-Sha256": sha256, + PORTABLE_CAMERA_EPOCH_ARCHIVE_ID_HEADER: archive_id, + } + if mutation == "corruption": + changed = bytearray(archive_payload) + offset = archive_payload.find(b"sealed-camera-init") + assert offset >= 0 + changed[offset] ^= 1 + archive_payload = bytes(changed) + return archive_payload, headers + + def test_http_gateway_materializes_only_exact_claim_bound_members( tmp_path: Path, ) -> None: @@ -232,6 +329,8 @@ def test_http_gateway_materializes_only_exact_claim_bound_members( artifact_requests.append(request) if request.url.path.endswith("/source-materialization"): return httpx.Response(200, json=manifest) + if request.url.path.endswith("/source-camera-epoch-archive"): + return httpx.Response(404) member_id = request.url.path.rsplit("/", 1)[-1] payload = payloads[member_id] return httpx.Response( @@ -266,7 +365,148 @@ def test_http_gateway_materializes_only_exact_claim_bound_members( ) assert persisted == manifest assert "path" not in json.dumps(persisted, sort_keys=True) - assert len(artifact_requests) == 1 + len(payloads) + assert len(artifact_requests) == 2 + len(payloads) + + +def test_http_gateway_packed_camera_epoch_bounds_requests_and_header_timeout( + tmp_path: Path, +) -> None: + job = _job( + bundle_sha256=hashlib.sha256(b"source-bundle").hexdigest(), + capability_sha256=hashlib.sha256(b"source-capability").hexdigest(), + ) + manifest, payloads = _source_contract(job, camera_segment_count=101) + archive_payload, archive_headers = _camera_archive_response( + job, + manifest, + payloads, + ) + artifact_requests: list[httpx.Request] = [] + camera_member_requests: list[str] = [] + archive_read_timeouts: list[float] = [] + rows = manifest["members"] + assert isinstance(rows, list) + camera_member_ids = { + str(row["member_id"]) + for row in rows + if isinstance(row, dict) + and row.get("kind") in {"camera-init", "camera-segment"} + } + + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path.endswith("/claims"): + return _claim_response() + artifact_requests.append(request) + if request.url.path.endswith("/source-materialization"): + return httpx.Response(200, json=manifest) + if request.url.path.endswith("/source-camera-epoch-archive"): + timeout = request.extensions.get("timeout") + assert isinstance(timeout, dict) + read_timeout = timeout.get("read") + assert isinstance(read_timeout, float) + assert read_timeout > WORKER_HTTP_DEFAULT_TIMEOUT_SECONDS + archive_read_timeouts.append(read_timeout) + return httpx.Response( + 200, + content=archive_payload, + headers=archive_headers, + ) + member_id = request.url.path.rsplit("/", 1)[-1] + if member_id in camera_member_ids: + camera_member_requests.append(member_id) + payload = payloads[member_id] + return httpx.Response( + 200, + content=payload, + headers={ + "X-Mission-Core-Content-Sha256": hashlib.sha256( + payload + ).hexdigest() + }, + ) + + with ObservatoryWorkerHttpGateway( + base_url="http://127.0.0.1:18080", + bearer_token=BEARER_TOKEN, + work_root=tmp_path / "worker", + transport=httpx.MockTransport(handler), + ) as gateway: + _cache_claim(gateway) + stage = gateway.materialize(job) + + assert archive_read_timeouts == [ + WORKER_HTTP_CAMERA_ARCHIVE_READ_TIMEOUT_SECONDS + ] + assert camera_member_requests == [] + assert len(artifact_requests) == 6 + assert ( + stage.root / "camera/epoch-1/segments/101.m4s" + ).read_bytes() == b"sealed-camera-segment-101" + expected_files = { + "source-bundle.json", + "source-capability.json", + "mqtt.raw.k1mqtt", + "mqtt.metadata.jsonl", + "materialization-manifest.json", + "camera/epoch-1/init.mp4", + *(f"camera/epoch-1/segments/{sequence}.m4s" for sequence in range(1, 102)), + } + assert { + path.relative_to(stage.root).as_posix() + for path in stage.root.rglob("*") + if path.is_file() + } == expected_files + + +@pytest.mark.parametrize( + ("mutation", "message"), + [ + ("corruption", "content differs"), + ("traversal", "unsafe member"), + ("symlink", "unsafe member"), + ], +) +def test_http_gateway_rejects_corrupt_or_unsafe_camera_archive( + tmp_path: Path, + mutation: str, + message: str, +) -> None: + job = _job( + bundle_sha256=hashlib.sha256(b"source-bundle").hexdigest(), + capability_sha256=hashlib.sha256(b"source-capability").hexdigest(), + ) + manifest, payloads = _source_contract(job) + archive_payload, archive_headers = _camera_archive_response( + job, + manifest, + payloads, + mutation=mutation, + ) + + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path.endswith("/claims"): + return _claim_response() + if request.url.path.endswith("/source-materialization"): + return httpx.Response(200, json=manifest) + if request.url.path.endswith("/source-camera-epoch-archive"): + return httpx.Response( + 200, + content=archive_payload, + headers=archive_headers, + ) + pytest.fail("unsafe camera archive must fail before member fallback") + + with ObservatoryWorkerHttpGateway( + base_url="http://127.0.0.1:18080", + bearer_token=BEARER_TOKEN, + work_root=tmp_path / "worker", + transport=httpx.MockTransport(handler), + ) as gateway: + _cache_claim(gateway) + with pytest.raises(ObservatoryWorkerHttpError, match=message): + gateway.materialize(job) + + assert not list(tmp_path.rglob("operator-secret")) def test_http_gateway_rejects_server_selected_source_path(tmp_path: Path) -> None: