diff --git a/src/k1link/observatory/worker_http_transport.py b/src/k1link/observatory/worker_http_transport.py index 7d35afc..c72ebc7 100644 --- a/src/k1link/observatory/worker_http_transport.py +++ b/src/k1link/observatory/worker_http_transport.py @@ -502,7 +502,10 @@ class ObservatoryWorkerHttpGateway(ObservatoryWorkerTransport): with self._client.stream( "GET", self._job_path(job.job_id, "source-camera-epoch-archive"), - headers=self._claim_headers(context), + headers={ + **self._claim_headers(context), + "Accept-Encoding": "identity", + }, timeout=httpx.Timeout( connect=self._timeout_seconds, read=WORKER_HTTP_CAMERA_ARCHIVE_READ_TIMEOUT_SECONDS, @@ -513,6 +516,14 @@ class ObservatoryWorkerHttpGateway(ObservatoryWorkerTransport): if response.status_code == 404: return self._raise_for_status(response) + content_encoding = response.headers.get("content-encoding") + if ( + content_encoding is not None + and content_encoding.lower() != "identity" + ): + raise ObservatoryWorkerHttpError( + "camera epoch archive content encoding changed" + ) if ( response.headers.get("content-type") != PORTABLE_CAMERA_EPOCH_ARCHIVE_MEDIA_TYPE @@ -620,9 +631,20 @@ class ObservatoryWorkerHttpGateway(ObservatoryWorkerTransport): with self._client.stream( "GET", self._job_path(job_id, f"source-members/{member.member_id}"), - headers=self._claim_headers(context), + headers={ + **self._claim_headers(context), + "Accept-Encoding": "identity", + }, ) as response: self._raise_for_status(response) + content_encoding = response.headers.get("content-encoding") + if ( + content_encoding is not None + and content_encoding.lower() != "identity" + ): + raise ObservatoryWorkerHttpError( + "source member content encoding changed" + ) declared = response.headers.get("content-length") if declared is not None and declared != str(member.byte_length): raise ObservatoryWorkerHttpError( diff --git a/tests/test_observatory_worker_http_transport.py b/tests/test_observatory_worker_http_transport.py index 0ed644b..ad74ea5 100644 --- a/tests/test_observatory_worker_http_transport.py +++ b/tests/test_observatory_worker_http_transport.py @@ -1,5 +1,6 @@ from __future__ import annotations +import gzip import hashlib import io import json @@ -146,6 +147,8 @@ def _source_contract( *, inject_path: bool = False, camera_segment_count: int = 1, + source_bundle_payload: bytes = b"source-bundle", + source_capability_payload: bytes = b"source-capability", ) -> tuple[dict[str, object], dict[str, bytes]]: if camera_segment_count < 1: raise ValueError("camera segment count must be positive") @@ -153,13 +156,13 @@ def _source_contract( _source_member( job, kind="source-bundle", - payload=b"source-bundle", + payload=source_bundle_payload, media_type="application/json", ), _source_member( job, kind="source-capability", - payload=b"source-capability", + payload=source_capability_payload, media_type="application/json", ), _source_member( @@ -329,6 +332,7 @@ 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) + assert request.headers["accept-encoding"] == "identity" if request.url.path.endswith("/source-camera-epoch-archive"): return httpx.Response(404) member_id = request.url.path.rsplit("/", 1)[-1] @@ -368,6 +372,132 @@ def test_http_gateway_materializes_only_exact_claim_bound_members( assert len(artifact_requests) == 2 + len(payloads) +def test_artifact_gets_request_identity_encoding_before_length_validation( + tmp_path: Path, +) -> None: + bundle_payload = b'{"bundle":"' + b"b" * 4096 + b'"}' + capability_payload = b'{"capability":"' + b"c" * 4096 + b'"}' + assert len(gzip.compress(bundle_payload)) != len(bundle_payload) + assert len(gzip.compress(capability_payload)) != len(capability_payload) + job = _job( + bundle_sha256=hashlib.sha256(bundle_payload).hexdigest(), + capability_sha256=hashlib.sha256(capability_payload).hexdigest(), + ) + manifest, payloads = _source_contract( + job, + source_bundle_payload=bundle_payload, + source_capability_payload=capability_payload, + ) + artifact_encodings: list[str] = [] + + 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) + accept_encoding = request.headers["accept-encoding"] + artifact_encodings.append(accept_encoding) + 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] + if "gzip" in accept_encoding: + compressed = gzip.compress(payload) + return httpx.Response( + 200, + content=compressed, + headers={ + "Content-Encoding": "gzip", + "Content-Length": str(len(compressed)), + "X-Mission-Core-Content-Sha256": hashlib.sha256( + payload + ).hexdigest(), + }, + ) + 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 artifact_encodings == ["identity"] * (1 + len(payloads)) + assert (stage.root / "source-bundle.json").read_bytes() == bundle_payload + assert ( + stage.root / "source-capability.json" + ).read_bytes() == capability_payload + + +@pytest.mark.parametrize("target", ["camera-archive", "source-member"]) +def test_artifact_get_rejects_non_identity_content_encoding( + tmp_path: Path, + target: 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, + ) + + 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) + assert request.headers["accept-encoding"] == "identity" + if request.url.path.endswith("/source-camera-epoch-archive"): + if target == "camera-archive": + return httpx.Response( + 200, + stream=httpx.ByteStream(archive_payload), + headers={**archive_headers, "Content-Encoding": "gzip"}, + ) + return httpx.Response(404) + member_id = request.url.path.rsplit("/", 1)[-1] + payload = payloads[member_id] + return httpx.Response( + 200, + stream=httpx.ByteStream(payload), + headers={ + "Content-Encoding": "gzip", + "Content-Length": str(len(payload)), + "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) + with pytest.raises( + ObservatoryWorkerHttpError, + match="content encoding changed", + ): + gateway.materialize(job) + + def test_http_gateway_packed_camera_epoch_bounds_requests_and_header_timeout( tmp_path: Path, ) -> None: @@ -400,6 +530,7 @@ def test_http_gateway_packed_camera_epoch_bounds_requests_and_header_timeout( if request.url.path.endswith("/source-materialization"): return httpx.Response(200, json=manifest) if request.url.path.endswith("/source-camera-epoch-archive"): + assert request.headers["accept-encoding"] == "identity" timeout = request.extensions.get("timeout") assert isinstance(timeout, dict) read_timeout = timeout.get("read") @@ -412,6 +543,7 @@ def test_http_gateway_packed_camera_epoch_bounds_requests_and_header_timeout( headers=archive_headers, ) member_id = request.url.path.rsplit("/", 1)[-1] + assert request.headers["accept-encoding"] == "identity" if member_id in camera_member_ids: camera_member_requests.append(member_id) payload = payloads[member_id]