fix(observatory): disable source artifact compression

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(

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]