fix: accept loader S3 checksum metadata

market_pipeline_lib/storage.py

@@ -28,6 +28,17 @@
 S3_RECEIPT_PROVIDERS = frozenset({"S3", "S3_COMPATIBLE"})
 
 
+def _s3_metadata_sha256(metadata: object) -> str:
+    """Read either canonical S3 SHA key without accepting contradictory claims."""
+    if not isinstance(metadata, dict):
+        return ""
+    adapter_hash = str(metadata.get("sha256") or "")
+    loader_hash = str(metadata.get("content-sha256") or "")
+    if adapter_hash and loader_hash and adapter_hash != loader_hash:
+        return ""
+    return adapter_hash or loader_hash
+
+
 def sha256_hex_and_base64(path: Path, chunk_size: int = 1024 * 1024) -> tuple[str, str]:
     """Return (hex, base64) SHA-256 digests of one file in a single read.
 
@@ -367,8 +378,7 @@ def _receipt_from_head(
         *different* object already owns the immutable key; anything else means
         our own write cannot be trusted.
         """
-        metadata = head.get("Metadata", {})
-        actual_hash = metadata.get("sha256", "") if isinstance(metadata, dict) else ""
+        actual_hash = _s3_metadata_sha256(head.get("Metadata"))
         actual_size = int(head.get("ContentLength", 0))
         if actual_hash != expected_hash or actual_size != expected_size:
             if conflict:
@@ -530,7 +540,7 @@ def verify(
             if self._is_missing(exc):
                 return VerificationResult(False, "", 0, "object missing")
             raise
-        actual = head.get("Metadata", {}).get("sha256", "")
+        actual = _s3_metadata_sha256(head.get("Metadata"))
         return VerificationResult(
             actual == expected_sha256,
             actual,
@@ -552,8 +562,7 @@ def verify_version(
             if self._is_missing(exc):
                 return VerificationResult(False, "", 0, "object version missing")
             raise
-        metadata = head.get("Metadata", {})
-        actual_hash = metadata.get("sha256", "") if isinstance(metadata, dict) else ""
+        actual_hash = _s3_metadata_sha256(head.get("Metadata"))
         actual_size = int(head.get("ContentLength", 0))
         if str(head.get("VersionId", "")) != provider_version_id:
             return VerificationResult(False, actual_hash, actual_size, "version mismatch")

tests/test_storage_adapter.py

@@ -358,6 +358,44 @@ def test_verify_reports_a_sha256_metadata_mismatch(self) -> None:
 
 
 class S3ObjectStoreContractTests(unittest.TestCase):
+    def test_verify_version_accepts_the_loader_content_sha256_metadata_key(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            source = write_small_parquet(Path(temporary) / "source.parquet")
+            client = FakeS3Client()
+            store = S3ObjectStore("bucket", client=client)
+            receipt = store.put(source, "market/source.parquet")
+            stored = client.versions[("bucket", receipt.object_key, receipt.provider_version_id)]
+            stored["Metadata"] = {"content-sha256": receipt.content_hash}
+
+            verified = store.verify_version(
+                receipt.object_key,
+                receipt.provider_version_id,
+                receipt.content_hash,
+                receipt.byte_size,
+            )
+
+            self.assertTrue(verified.ok)
+            self.assertEqual(verified.content_hash, receipt.content_hash)
+
+    def test_verify_version_rejects_conflicting_sha256_metadata_keys(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            source = write_small_parquet(Path(temporary) / "source.parquet")
+            client = FakeS3Client()
+            store = S3ObjectStore("bucket", client=client)
+            receipt = store.put(source, "market/source.parquet")
+            stored = client.versions[("bucket", receipt.object_key, receipt.provider_version_id)]
+            stored["Metadata"]["content-sha256"] = "f" * 64
+
+            verified = store.verify_version(
+                receipt.object_key,
+                receipt.provider_version_id,
+                receipt.content_hash,
+                receipt.byte_size,
+            )
+
+            self.assertFalse(verified.ok)
+            self.assertEqual(verified.message, "sha256 metadata mismatch")
+
     def test_verify_version_uses_the_exact_s3_version_and_all_attestations(self) -> None:
         with tempfile.TemporaryDirectory() as temporary:
             source = write_small_parquet(Path(temporary) / "source.parquet")