Harden exact legacy payload validation

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

Author: Pigbibi

src/publish.py

@@ -134,6 +134,10 @@ def _sha256_file(path: Path) -> str:
     return digest.hexdigest()
 
 
+def _reject_non_standard_json_constant(value: str) -> None:
+    raise ValueError(f"Non-standard JSON constant is not allowed: {value}")
+
+
 def _validate_runtime_evidence_identity(
     *,
     identity: Any,
@@ -315,8 +319,11 @@ def build_firestore_payload(
     except UnicodeDecodeError as exc:
         raise ValueError("live_pool_legacy.json exact bytes must be valid UTF-8.") from exc
     try:
-        exact_payload = json.loads(exact_text)
-    except json.JSONDecodeError as exc:
+        exact_payload = json.loads(
+            exact_text,
+            parse_constant=_reject_non_standard_json_constant,
+        )
+    except ValueError as exc:
         raise ValueError("live_pool_legacy.json exact bytes must contain valid JSON.") from exc
     if not isinstance(exact_payload, dict):
         raise ValueError("live_pool_legacy.json exact bytes must contain a JSON object.")
@@ -343,6 +350,12 @@ def build_firestore_payload(
         or exact_payload.get("symbols") != artifacts.live_pool_legacy.get("symbols")
         or exact_payload.get("symbol_map")
         != artifacts.live_pool_legacy.get("symbol_map")
+        or exact_payload.get("pool_size")
+        != artifacts.live_pool_legacy.get("pool_size")
+        or exact_payload.get("pool_size") != artifacts.live_pool.get("pool_size")
+        or exact_payload.get("source_project")
+        != artifacts.live_pool_legacy.get("source_project")
+        or exact_payload.get("source_project") != settings.source_project
         or list(exact_payload.get("symbols", {})) != symbols
         or exact_payload.get("symbol_map") != symbol_map
     ):

tests/test_release_contract.py

@@ -337,10 +337,51 @@ def test_firestore_payload_rejects_top_level_convenience_field_mismatch(self) ->
                         storage_layout,
                     )
 
+    def test_firestore_payload_rejects_pool_size_or_source_project_mismatch(self) -> None:
+        cases = {
+            "pool_size_mismatch": ("pool_size", 4),
+            "pool_size_missing": ("pool_size", None),
+            "source_project_mismatch": ("source_project", "wrong-source"),
+            "source_project_missing": ("source_project", None),
+        }
+        for label, (field, value) in cases.items():
+            with self.subTest(label=label), tempfile.TemporaryDirectory() as tmp_dir:
+                root = Path(tmp_dir)
+                output_dir = self.build_runtime_identity_outputs(root)
+                legacy_path = output_dir / "live_pool_legacy.json"
+                legacy_payload = json.loads(legacy_path.read_text(encoding="utf-8"))
+                if value is None:
+                    legacy_payload.pop(field)
+                else:
+                    legacy_payload[field] = value
+                write_json(legacy_path, legacy_payload)
+
+                manifest_path = output_dir / "artifact_manifest.json"
+                manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+                digest = sha256_file(legacy_path)
+                manifest["artifacts"]["live_pool_legacy"]["sha256"] = digest
+                manifest["runtime_evidence_identity"]["artifacts"][
+                    "live_pool_legacy"
+                ]["sha256"] = digest
+                write_json(manifest_path, manifest)
+
+                artifacts, settings, storage_layout = self.load_publish_context(
+                    output_dir
+                )
+                with self.assertRaisesRegex(ValueError, "convenience fields"):
+                    build_firestore_payload(
+                        settings,
+                        artifacts,
+                        storage_layout,
+                    )
+
     def test_firestore_payload_rejects_invalid_legacy_artifact_bytes(self) -> None:
         cases = {
             "invalid_utf8": (b"\xff", "valid UTF-8"),
             "invalid_json": (b"{", "valid JSON"),
+            "nan": (b'{"value": NaN}', "valid JSON"),
+            "infinity": (b'{"value": Infinity}', "valid JSON"),
+            "negative_infinity": (b'{"value": -Infinity}', "valid JSON"),
             "non_object": (b"[]", "JSON object"),
         }
         for label, (invalid_bytes, expected_error) in cases.items():