fix(sidecar): bind audit receipt to per-file SHA256 (CR-11)

Audit receipt now includes file_records (path/size/sha256) per bucket.
Upload manifest cross-checks receipt file_records against actual
inventory bytes before publication, rejecting stale receipts that
certify a different byte set.

Regression: replacing an audited parquet with arbitrary bytes while
preserving file count now fails manifest creation.

Author: apstenku123

scripts/audit_sidecar_parquet.py

@@ -1589,10 +1589,20 @@ def _audit_file(
     if stats.bad_rows:
         stats.bad_files = 1
 
+    import hashlib as _hashlib
+
+    file_size = path.stat().st_size
+    file_digest = _hashlib.sha256()
+    with path.open("rb") as _fh:
+        while _chunk := _fh.read(8 * 1024 * 1024):
+            file_digest.update(_chunk)
+
     return {
         "kind": kind,
         "bucket": bucket,
         "path": path_str,
+        "size": file_size,
+        "sha256": file_digest.hexdigest(),
         "stats": stats.as_dict(),
     }
 
@@ -1614,19 +1624,37 @@ def _rollup(results: list[dict[str, Any]]) -> dict[str, Any]:
     by_kind: dict[str, AuditStats] = {}
     by_bucket: dict[str, AuditStats] = {}
     by_kind_bucket: dict[str, AuditStats] = {}
+    file_records_by_bucket: dict[str, list[dict[str, Any]]] = {}
     bad_files: list[str] = []
 
     for result in results:
         kind = result["kind"]
         bucket = result["bucket"]
+        key = f"{kind}/{bucket}"
         stats = _stats_from_dict(result["stats"])
         total.add(stats)
         by_kind.setdefault(kind, AuditStats()).add(stats)
         by_bucket.setdefault(bucket, AuditStats()).add(stats)
-        by_kind_bucket.setdefault(f"{kind}/{bucket}", AuditStats()).add(stats)
+        by_kind_bucket.setdefault(key, AuditStats()).add(stats)
+        file_records_by_bucket.setdefault(key, []).append(
+            {
+                "path": result["path"],
+                "size": result.get("size", 0),
+                "sha256": result.get("sha256", ""),
+            }
+        )
         if result["stats"]["bad_files"] or result["stats"]["bad_rows"]:
             bad_files.append(result["path"])
 
+    by_kind_bucket_out: dict[str, Any] = {}
+    for k, v in sorted(by_kind_bucket.items()):
+        entry = v.as_dict()
+        entry["file_records"] = sorted(
+            file_records_by_bucket.get(k, []),
+            key=lambda r: r["path"],
+        )
+        by_kind_bucket_out[k] = entry
+
     return {
         "total": total.as_dict(),
         "by_kind": {k: v.as_dict() for k, v in sorted(by_kind.items())},
@@ -1637,7 +1665,7 @@ def _rollup(results: list[dict[str, Any]]) -> dict[str, Any]:
                 key=lambda kv: int(kv[0]) if kv[0].isdigit() else kv[0],
             )
         },
-        "by_kind_bucket": {k: v.as_dict() for k, v in sorted(by_kind_bucket.items())},
+        "by_kind_bucket": by_kind_bucket_out,
         "bad_files": bad_files[:10000],
     }
 

scripts/sidecar_manifest_contract.py

@@ -201,6 +201,7 @@ def validate_audit_receipt(
     receipt: object,
     *,
     selected_keys: Sequence[str],
+    inventory: Mapping[str, Sequence[Mapping[str, object]]] | None = None,
 ) -> dict[str, Any]:
     if not isinstance(receipt, dict):
         raise ValueError("sidecar audit receipt must be an object")
@@ -225,9 +226,47 @@ def validate_audit_receipt(
             raise ValueError(f"sidecar audit bucket is not green: {key}")
         if int(row.get("files", 0)) <= 0 or int(row.get("valid_tokens", 0)) <= 0:
             raise ValueError(f"sidecar audit bucket has no verified data: {key}")
+    if inventory is not None:
+        _validate_receipt_file_records(by_kind_bucket, selected_keys, inventory)
     return receipt
 
 
+def _validate_receipt_file_records(
+    by_kind_bucket: dict,
+    selected_keys: Sequence[str],
+    inventory: Mapping[str, Sequence[Mapping[str, object]]],
+) -> None:
+    """Cross-check per-file path/size/SHA256 records in the receipt against
+    the actual upload inventory, binding the receipt to exact bytes."""
+    for key in selected_keys:
+        row = by_kind_bucket[key]
+        file_records = row.get("file_records")
+        inv_records = inventory.get(key)
+        if inv_records is None:
+            continue
+        if file_records is None:
+            raise ValueError(
+                f"sidecar audit receipt bucket {key} lacks file_records; "
+                f"cannot bind receipt to inventory bytes"
+            )
+        receipt_set = {
+            (str(r["path"]), int(r["size"]), str(r["sha256"]))
+            for r in file_records
+        }
+        inventory_set = {
+            (str(r["path"]), int(r["size"]), str(r["sha256"]))
+            for r in inv_records
+        }
+        if receipt_set != inventory_set:
+            stale = sorted(receipt_set - inventory_set)
+            new = sorted(inventory_set - receipt_set)
+            raise ValueError(
+                f"sidecar audit receipt file_records do not match inventory "
+                f"for bucket {key}: "
+                f"stale={stale[:3]} new={new[:3]}"
+            )
+
+
 def expected_bucket_remotes() -> set[str]:
     return {
         f"parquet/{kind}/{bucket}"

scripts/upload_verified_sidecar_to_nebius_s3.py

@@ -422,11 +422,8 @@ def _write_manifest(
     if len(audit_receipts) != 1:
         raise SystemExit("exactly one final audit receipt is required")
     audit_path = audit_receipts[0]
-    audit_payload = validate_audit_receipt(
-        json.loads(audit_path.read_text(encoding="utf-8")),
-        selected_keys=selected_keys,
-    )
     inventory: list[dict[str, object]] = []
+    inventory_by_bucket: dict[str, list[dict[str, object]]] = {}
     for remote, local in selections:
         record = inventory_directory(local, remote=remote)
         inventory.append(
@@ -438,10 +435,25 @@ def _write_manifest(
         )
         key = remote.removeprefix("parquet/") if remote.startswith("parquet/") else None
         if key is not None:
+            inventory_by_bucket[key] = [
+                {"path": f["path"], "size": f["size"], "sha256": f["sha256"]}
+                for f in record["files"]
+            ]
+    audit_payload = validate_audit_receipt(
+        json.loads(audit_path.read_text(encoding="utf-8")),
+        selected_keys=selected_keys,
+        inventory=inventory_by_bucket,
+    )
+    for remote, local in selections:
+        key = remote.removeprefix("parquet/") if remote.startswith("parquet/") else None
+        if key is not None:
+            inv_record = next(
+                r for r in inventory if r["remote"] == remote
+            )
             expected_files = int(audit_payload["by_kind_bucket"][key]["files"])
-            if int(record["file_count"]) != expected_files:
+            if int(inv_record["file_count"]) != expected_files:
                 raise SystemExit(
-                    f"inventory file count for {remote}={record['file_count']} "
+                    f"inventory file count for {remote}={inv_record['file_count']} "
                     f"differs from audit receipt={expected_files}"
                 )
     audit_selection = next(

tests/test_verified_sidecar_manifest_selection.py

@@ -24,6 +24,8 @@
     inventory_directory,
     inventory_sha256,
     selection_policy,
+    sha256_file,
+    validate_audit_receipt,
     validate_manifest,
 )
 
@@ -130,26 +132,33 @@ def _build_receipt(
     valid_tokens: dict[str, int],
     bad: dict[str, tuple[int, int]] | None = None,
     omit: set[str] | None = None,
+    file_records: dict[str, list[dict[str, object]]] | None = None,
 ) -> Path:
     """Write a real audit receipt JSON with a ``by_kind_bucket`` map.
 
     ``bad`` maps a bucket key to ``(bad_files, bad_rows)``; ``omit`` drops keys
     entirely (stale/narrow receipt). The aggregate ``total`` is kept green on
     purpose to prove the gate is per-selected-bucket, not the coarse aggregate.
+    ``file_records`` maps a bucket key to per-file identity records
+    (path/size/sha256) that bind the receipt to exact bytes.
     """
     bad = bad or {}
     omit = omit or set()
-    by_kind_bucket: dict[str, dict[str, int]] = {}
+    file_records = file_records or {}
+    by_kind_bucket: dict[str, dict[str, object]] = {}
     for key, tokens in valid_tokens.items():
         if key in omit:
             continue
         bf, br = bad.get(key, (0, 0))
-        by_kind_bucket[key] = {
+        entry: dict[str, object] = {
             "valid_tokens": tokens,
             "files": 1,
             "bad_files": bf,
             "bad_rows": br,
         }
+        if key in file_records:
+            entry["file_records"] = file_records[key]
+        by_kind_bucket[key] = entry
     payload = {
         "schema": AUDIT_SCHEMA,
         "status": "verified",
@@ -258,18 +267,31 @@ def _download_fixture(
     return finalize_manifest(payload), root
 
 
-def _populate_upload_sources(selections: tuple[tuple[str, Path], ...]) -> None:
+def _populate_upload_sources(
+    selections: tuple[tuple[str, Path], ...],
+) -> dict[str, list[dict[str, object]]]:
+    """Create source directories and return per-bucket file identity records."""
+    file_records: dict[str, list[dict[str, object]]] = {}
     for index, (remote, local) in enumerate(selections, 1):
         local.mkdir(parents=True, exist_ok=True)
         if remote.startswith("parquet/"):
             key = remote.removeprefix("parquet/")
             bucket = int(remote.rsplit("/", 1)[1])
+            shard_path = local / f"part-{index:03d}.parquet"
             _write_green_shard(
-                local / f"part-{index:03d}.parquet",
+                shard_path,
                 bucket=bucket,
                 valid_tokens=_BUCKET_VALID_TOKENS[key],
                 source_id=index,
             )
+            file_records.setdefault(key, []).append(
+                {
+                    "path": shard_path.name,
+                    "size": shard_path.stat().st_size,
+                    "sha256": sha256_file(shard_path),
+                }
+            )
+    return file_records
 
 
 def test_token_total_is_profile_aware_subset(tmp_path: Path) -> None:
@@ -494,10 +516,14 @@ def test_upload_main_dry_run_writes_manifest_and_prints_targets(
     monkeypatch.chdir(tmp_path)
     # Every selected source dir must exist or _existing_sources RAISEs; this
     # also creates the audit dir that holds the receipt.
-    _populate_upload_sources(upload.ALL_VALID_SELECTIONS)
+    fr = _populate_upload_sources(upload.ALL_VALID_SELECTIONS)
     # Real temp receipt at the relative path the script reads (_audit_receipts),
     # covering every selected parquet bucket and green.
-    _build_receipt(upload._audit_receipts()[0], valid_tokens=_BUCKET_VALID_TOKENS)
+    _build_receipt(
+        upload._audit_receipts()[0],
+        valid_tokens=_BUCKET_VALID_TOKENS,
+        file_records=fr,
+    )
 
     manifest_path = tmp_path / "manifest.json"
     rc = upload.main(
@@ -549,11 +575,12 @@ def test_upload_main_dry_run_does_not_gate_default_on_standalone_pr_bucket(
     tmp_path: Path, monkeypatch
 ) -> None:
     monkeypatch.chdir(tmp_path)
-    _populate_upload_sources(upload.ALL_VALID_SELECTIONS)
+    fr = _populate_upload_sources(upload.ALL_VALID_SELECTIONS)
     _build_receipt(
         upload._audit_receipts()[0],
         valid_tokens=_BUCKET_VALID_TOKENS,
         bad={"pr/8192": (1, 0)},
+        file_records=fr,
     )
 
     rc = upload.main(
@@ -574,13 +601,14 @@ def test_upload_main_dry_run_raises_on_non_green_explicit_standalone_pr_bucket(
     tmp_path: Path, monkeypatch
 ) -> None:
     monkeypatch.chdir(tmp_path)
-    _populate_upload_sources(
+    fr = _populate_upload_sources(
         upload.CODE_COMMIT_SELECTIONS + upload.STANDALONE_PR_SELECTIONS
     )
     _build_receipt(
         upload._audit_receipts()[0],
         valid_tokens=_BUCKET_VALID_TOKENS,
         bad={"pr/8192": (1, 0)},
+        file_records=fr,
     )
 
     with pytest.raises(SystemExit) as exc:
@@ -777,3 +805,92 @@ def test_download_main_dry_run_copies_manifest_uri(
     # syncs code/commit buckets, but not standalone PR diagnostics.
     assert "s3://mybucket/myprefix/parquet/code/1024/" in out
     assert "parquet/pr/" not in out
+
+
+# --- CR-11: audit receipt must bind to per-file bytes -------------------------
+
+
+def test_audit_receipt_rejects_stale_file_records_after_byte_swap(
+    tmp_path: Path,
+) -> None:
+    """A green receipt whose file_records reference the original bytes must be
+    rejected when the underlying parquet is replaced with different content
+    (same file count).  A fresh receipt matching the new bytes must pass.
+    """
+    shard_dir = tmp_path / "code" / "1024"
+    shard_dir.mkdir(parents=True)
+    shard = shard_dir / "part-001.parquet"
+    _write_green_shard(shard, bucket=1024, valid_tokens=42, source_id=1)
+
+    original_sha = sha256_file(shard)
+    original_size = shard.stat().st_size
+    original_records = [
+        {"path": "part-001.parquet", "size": original_size, "sha256": original_sha}
+    ]
+
+    receipt_path = tmp_path / "sidecar_parquet_audit.json"
+    _build_receipt(
+        receipt_path,
+        valid_tokens={"code/1024": 42},
+        file_records={"code/1024": original_records},
+    )
+    receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
+
+    # Inventory matching the original bytes passes.
+    inventory_ok = {"code/1024": original_records}
+    validate_audit_receipt(receipt, selected_keys=["code/1024"], inventory=inventory_ok)
+
+    # Replace the shard with different bytes (same file count).
+    shard.write_bytes(b"\x00" * original_size)
+    swapped_sha = sha256_file(shard)
+    assert swapped_sha != original_sha
+    swapped_records = [
+        {"path": "part-001.parquet", "size": original_size, "sha256": swapped_sha}
+    ]
+    inventory_swapped = {"code/1024": swapped_records}
+
+    # The stale receipt must be rejected against the swapped inventory.
+    with pytest.raises(ValueError, match="file_records do not match inventory"):
+        validate_audit_receipt(
+            receipt, selected_keys=["code/1024"], inventory=inventory_swapped
+        )
+
+    # A fresh receipt binding to the new bytes must pass.
+    fresh_receipt_path = tmp_path / "fresh_audit.json"
+    _build_receipt(
+        fresh_receipt_path,
+        valid_tokens={"code/1024": 42},
+        file_records={"code/1024": swapped_records},
+    )
+    fresh_receipt = json.loads(fresh_receipt_path.read_text(encoding="utf-8"))
+    validate_audit_receipt(
+        fresh_receipt, selected_keys=["code/1024"], inventory=inventory_swapped
+    )
+
+
+def test_audit_receipt_rejects_missing_file_records_when_inventory_supplied(
+    tmp_path: Path,
+) -> None:
+    """When inventory is supplied but the receipt bucket lacks file_records,
+    validation must fail rather than silently skipping the byte binding.
+    """
+    receipt_path = tmp_path / "sidecar_parquet_audit.json"
+    _build_receipt(receipt_path, valid_tokens={"code/1024": 42})
+    receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
+
+    shard_dir = tmp_path / "code" / "1024"
+    shard_dir.mkdir(parents=True)
+    shard = shard_dir / "part-001.parquet"
+    _write_green_shard(shard, bucket=1024, valid_tokens=42, source_id=1)
+    inv = {
+        "code/1024": [
+            {
+                "path": "part-001.parquet",
+                "size": shard.stat().st_size,
+                "sha256": sha256_file(shard),
+            }
+        ]
+    }
+
+    with pytest.raises(ValueError, match="lacks file_records"):
+        validate_audit_receipt(receipt, selected_keys=["code/1024"], inventory=inv)