tighten raw period bundle bindings

Co-Authored-By: Codex <noreply@openai.com>

Author: Pigbibi

src/political_event_tracking_research/trusted_period_bundle_raw.py

@@ -204,7 +204,7 @@ def reject_constant(_: str) -> None:
     if snapshot["snapshot_version"] != SNAPSHOT_VERSION:
         raise _error("bundle_snapshot_version_invalid")
     _string(snapshot["source_run_id"], _RUN_ID_RE, "bundle_snapshot_run_invalid")
-    if snapshot["source_attempt"] != 1:
+    if type(snapshot["source_attempt"]) is not int or snapshot["source_attempt"] != 1:
         raise _error("bundle_snapshot_attempt_invalid")
     _string(snapshot["workflow_sha"], _SHA1_RE, "bundle_snapshot_workflow_invalid")
     _string(snapshot["producer_sha"], _SHA1_RE, "bundle_snapshot_producer_invalid")
@@ -324,6 +324,9 @@ def pairs(items: list[tuple[str, object]]) -> dict[str, object]:
     _safe_int(parsed["artifact_id"], "bundle_manifest_invalid")
     if type(parsed["source_attempt"]) is not int or parsed["source_attempt"] != 1:
         raise _error("bundle_manifest_invalid")
+    _string(parsed["source_run_id"], _RUN_ID_RE, "bundle_manifest_invalid")
+    if parsed["lock_version"] != LOCK_VERSION or parsed["snapshot_version"] != SNAPSHOT_VERSION:
+        raise _error("bundle_manifest_version_invalid")
     if type(parsed["retention_days"]) is not int or not 1 <= parsed["retention_days"] <= 90:
         raise _error("bundle_manifest_invalid")
     if _canonical_json(parsed, "bundle_manifest_invalid") != value:
@@ -360,6 +363,14 @@ def _manifest_bytes(
     )
 
 
+def _validate_snapshot_binding(
+    snapshot: dict[str, object], lock: PoliticalEventWeeklyPeriodLockV1, reviewed_workflow_sha: str
+) -> None:
+    expected = _expected_snapshot(lock, reviewed_workflow_sha)
+    if snapshot != expected:
+        raise _error("bundle_snapshot_mismatch")
+
+
 def build_period_bundle(
     lock_bytes: object,
     snapshot_bytes: object,
@@ -376,7 +387,8 @@ def build_period_bundle(
         _expected_snapshot(lock, identity["reviewed_workflow_sha"]), "bundle_snapshot_invalid"
     )
     parsed_snapshot = _snapshot(snapshot_bytes)
-    if snapshot_bytes != expected_snapshot or parsed_snapshot != json.loads(expected_snapshot):
+    _validate_snapshot_binding(parsed_snapshot, lock, identity["reviewed_workflow_sha"])
+    if snapshot_bytes != expected_snapshot:
         raise _error("bundle_snapshot_mismatch")
     evidence = _artifact(artifact, lock.source_run_id)
     manifest = _manifest_bytes(lock, identity, evidence, lock_bytes, snapshot_bytes)
@@ -425,6 +437,8 @@ def verify_period_bundle(
         raise _error("bundle_lock_mismatch")
     if values["snapshot_bytes"] != expected_snapshot_bytes or actual_snapshot != expected_snapshot:
         raise _error("bundle_snapshot_mismatch")
+    _validate_snapshot_binding(expected_snapshot, expected_lock, expected_identity["reviewed_workflow_sha"])
+    _validate_snapshot_binding(actual_snapshot, actual_lock, manifest["reviewed_workflow_sha"])
     if actual_evidence != expected_evidence:
         raise _error("bundle_artifact_mismatch")
     if actual_lock.workflow_ref != TRUSTED_WORKFLOW_REF:

tests/test_trusted_period_bundle_raw.py

@@ -135,6 +135,32 @@ def test_manifest_identity_fields_are_validated_before_reconstructed_equality()
         verify_period_bundle(value, lock_bytes(), snapshot_bytes(), IDENTITY_BYTES, artifact())
 
 
+@pytest.mark.parametrize("attempt", [True, 1.0, 2**53, (1, 2)])
+def test_source_attempt_wire_type_is_exact(attempt: object) -> None:
+    value = snapshot_value()
+    value["source_attempt"] = attempt
+    raw = json.dumps(value, sort_keys=True, separators=(",", ":")).encode()
+    with pytest.raises(TrustedPeriodBundleError, match="bundle_snapshot_"):
+        build_period_bundle(lock_bytes(), raw, IDENTITY_BYTES, artifact())
+
+
+def test_hash_correct_semantic_snapshot_forgery_is_rejected() -> None:
+    forged = snapshot_value()
+    forged["source_snapshot_digest"] = "0" * 64
+    forged_bytes = json.dumps(forged, sort_keys=True, separators=(",", ":")).encode()
+    with pytest.raises(TrustedPeriodBundleError, match="bundle_snapshot_mismatch"):
+        verify_period_bundle(bundle(), lock_bytes(), forged_bytes, IDENTITY_BYTES, artifact())
+
+
+def test_manifest_source_attempt_type_is_exact() -> None:
+    value = bundle()
+    manifest = json.loads(value["manifest_bytes"])
+    manifest["source_attempt"] = True
+    value["manifest_bytes"] = json.dumps(manifest, sort_keys=True, separators=(",", ":")).encode()
+    with pytest.raises(TrustedPeriodBundleError, match="bundle_manifest_"):
+        verify_period_bundle(value, lock_bytes(), snapshot_bytes(), IDENTITY_BYTES, artifact())
+
+
 @pytest.mark.parametrize("field", ["artifact", "identity", "lock_expected", "snapshot_expected"])
 def test_context_tamper_fails_closed(field: str) -> None:
     value = bundle()