From 438c939de5b34084dc7b4cff00ad3f3843731faf Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 14 Jul 2026 03:32:58 +0800 Subject: [PATCH 1/3] feat: validate bounded evidence bundle integrity Co-Authored-By: Codex --- src/us_equity_snapshot_pipelines/evidence.py | 225 +++++++++++++++++++ tests/test_soxl_tqqq_evidence.py | 110 +++++++++ 2 files changed, 335 insertions(+) diff --git a/src/us_equity_snapshot_pipelines/evidence.py b/src/us_equity_snapshot_pipelines/evidence.py index 742d8ac..6acabb4 100644 --- a/src/us_equity_snapshot_pipelines/evidence.py +++ b/src/us_equity_snapshot_pipelines/evidence.py @@ -1,9 +1,15 @@ from __future__ import annotations +import csv +import hashlib +import json import math +import os import re from collections.abc import Mapping from datetime import date, datetime, timedelta +from pathlib import Path +import stat from typing import Any @@ -132,3 +138,222 @@ def validate_metrics_payload(payload: Mapping[str, Any]) -> dict[str, Any]: if not isinstance(metrics, Mapping) or not metrics or not _contains_numeric(metrics): _invalid("numeric metrics evidence") return dict(payload) + + +_FILE_MAX_BYTES = { + "manifest.json": 262_144, "champion_identity.json": 131_072, "prices.csv": 67_108_864, + "data_quality.json": 8_388_608, "daily_returns.csv": 33_554_432, "targets.csv": 33_554_432, + "trades.csv": 33_554_432, "costs.csv": 33_554_432, "metrics.json": 8_388_608, + "walk_forward.json": 16_777_216, "trial_ledger.jsonl": 67_108_864, "robustness.json": 33_554_432, + "risk_sleeve.json": 16_777_216, "strategy_performance.v2.json": 8_388_608, "checksums.sha256": 65_536, +} +_MAX_BUNDLE_BYTES = 361_168_896 +_MAX_LINE_BYTES = 1_048_576 +_HASH_CHUNK_BYTES = 1_048_576 +_JSON_FILES = frozenset({ + "manifest.json", "champion_identity.json", "data_quality.json", "metrics.json", "walk_forward.json", + "robustness.json", "risk_sleeve.json", "strategy_performance.v2.json", +}) +_CHAMPION_FIELDS = frozenset({"schema", "profile", "code_sha", "config_sha256", "plugin_sha256", "execution_timing", "timezone", "calendar", "as_of"}) + + +class _DuplicateJSONKey(ValueError): + pass + + +def _json_pairs(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + result: dict[str, Any] = {} + for key, value in pairs: + if key in result: + raise _DuplicateJSONKey + result[key] = value + return result + + +def _reject_json_constant(value: str) -> None: + raise ValueError(value) + + +def _check_json_depth(value: Any, name: str) -> None: + pending = [(value, 1)] + while pending: + item, depth = pending.pop() + if not isinstance(item, (dict, list)): + continue + if depth > 64: + _invalid(f"{name} JSON nesting") + children = item.values() if isinstance(item, dict) else item + pending.extend((child, depth + 1) for child in children) + + +def _consume(state: dict[str, int], name: str, cap: int, hasher: Any, chunk: bytes, used: int) -> int: + used += len(chunk) + state["total"] += len(chunk) + if used > cap: + _invalid(f"{name} size") + if state["total"] > _MAX_BUNDLE_BYTES: + _invalid("bundle total size") + hasher.update(chunk) + return used + + +def _parse_json_text(text: str, name: str) -> Any: + try: + value = json.loads(text, object_pairs_hook=_json_pairs, parse_constant=_reject_json_constant) + except (UnicodeDecodeError, json.JSONDecodeError, RecursionError, ValueError): + _invalid(f"{name} JSON") + if not isinstance(value, dict): + _invalid(f"{name} JSON object") + _check_json_depth(value, name) + return value + + +def _parse_json_stream(stream: Any, name: str, state: dict[str, int]) -> tuple[dict[str, Any], str]: + hasher = hashlib.sha256() + chunks: list[bytes] = [] + used = 0 + while chunk := stream.read(_HASH_CHUNK_BYTES): + used = _consume(state, name, _FILE_MAX_BYTES[name], hasher, chunk, used) + chunks.append(chunk) + try: + text = b"".join(chunks).decode("utf-8") + except UnicodeDecodeError: + _invalid(f"{name} UTF-8") + return _parse_json_text(text, name), hasher.hexdigest() + + +def _physical_lines(stream: Any, name: str, state: dict[str, int], hasher: Any, encoding: str = "utf-8", require_lf: bool = False): + used = 0 + while raw := stream.readline(_MAX_LINE_BYTES + 1): + used = _consume(state, name, _FILE_MAX_BYTES[name], hasher, raw, used) + if len(raw) > _MAX_LINE_BYTES: + _invalid(f"{name} line size") + if require_lf and not raw.endswith(b"\n"): + _invalid(f"{name} line ending") + body = raw[:-1] if raw.endswith(b"\n") else raw + if not body.strip(): + _invalid(f"{name} blank line") + try: + yield raw.decode(encoding) + except UnicodeDecodeError: + _invalid(f"{name} UTF-8") + + +def _parse_jsonl(stream: Any, name: str, state: dict[str, int]) -> str: + hasher = hashlib.sha256() + count = 0 + for line in _physical_lines(stream, name, state, hasher, require_lf=True): + _parse_json_text(line[:-1], name) + count += 1 + if not count: + _invalid(f"{name} records") + return hasher.hexdigest() + + +def _parse_csv(stream: Any, name: str, state: dict[str, int]) -> str: + hasher = hashlib.sha256() + rows = 0 + header: list[str] | None = None + try: + for row in csv.reader(_physical_lines(stream, name, state, hasher)): + if header is None: + if not row or any(not column for column in row) or len(set(row)) != len(row): + _invalid(f"{name} header") + header = row + else: + if len(row) != len(header): + _invalid(f"{name} row width") + rows += 1 + except csv.Error: + _invalid(f"{name} CSV") + if header is None or rows < 2: + _invalid(f"{name} rows") + return hasher.hexdigest() + + +def _open_regular(root: str, name: str): + path = os.path.join(root, name) + try: + info = os.lstat(path) + if stat.S_ISLNK(info.st_mode) or not stat.S_ISREG(info.st_mode): + _invalid(f"{name} regular file") + fd = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)) + try: + info = os.fstat(fd) + if not stat.S_ISREG(info.st_mode) or info.st_size > _FILE_MAX_BYTES[name]: + _invalid(f"{name} regular file") + return os.fdopen(fd, "rb", closefd=True) + except BaseException: + os.close(fd) + raise + except EvidenceValidationError: + raise + except OSError: + _invalid(f"{name} regular file") + + +def _parse_checksums(stream: Any, state: dict[str, int]) -> tuple[dict[str, str], str]: + hasher = hashlib.sha256() + rows: list[tuple[str, str]] = [] + pattern = re.compile(r"([0-9a-f]{64}) ([^\n]+)\n\Z") + for line in _physical_lines(stream, "checksums.sha256", state, hasher, encoding="ascii", require_lf=True): + match = pattern.fullmatch(line) + if match is None: + _invalid("checksums.sha256 line") + rows.append((match.group(2), match.group(1))) + expected = sorted(set(REQUIRED_ARTIFACT_FILES) - {"checksums.sha256"}) + if [name for name, _ in rows] != expected: + _invalid("checksums.sha256 set") + return dict(rows), hasher.hexdigest() + + +def _parse_artifact(name: str, stream: Any, state: dict[str, int]) -> tuple[Any, str]: + if name in _JSON_FILES: + return _parse_json_stream(stream, name, state) + if name == "trial_ledger.jsonl": + return None, _parse_jsonl(stream, name, state) + if name == "checksums.sha256": + return _parse_checksums(stream, state) + return None, _parse_csv(stream, name, state) + + +def validate_evidence_bundle(bundle_root: str | Path) -> dict[str, Any]: + root = os.fspath(bundle_root) + try: + root_info = os.lstat(root) + except OSError: + _invalid("bundle root") + if stat.S_ISLNK(root_info.st_mode) or not stat.S_ISDIR(root_info.st_mode): + _invalid("bundle root") + state = {"total": 0} + with _open_regular(root, "manifest.json") as stream: + manifest, _ = _parse_json_stream(stream, "manifest.json", state) + manifest = validate_evidence_manifest(manifest) + try: + names = {entry.name for entry in os.scandir(root)} + except OSError: + _invalid("bundle root entries") + if names != {"manifest.json", *REQUIRED_ARTIFACT_FILES}: + _invalid("exact bundle root") + parsed: dict[str, Any] = {} + digests: dict[str, str] = {} + for name in REQUIRED_ARTIFACT_FILES: + with _open_regular(root, name) as stream: + parsed[name], digests[name] = _parse_artifact(name, stream, state) + identity = parsed["champion_identity.json"] + if set(identity) != _CHAMPION_FIELDS or identity["schema"] != "champion_identity.v1": + _invalid("champion identity fields") + for field in _CHAMPION_FIELDS - {"schema"}: + if identity[field] != manifest[field]: + _invalid(f"champion identity {field}") + try: + validate_metrics_payload(parsed["metrics.json"]) + except (EvidenceValidationError, RecursionError): + _invalid("metrics.json") + checksums = parsed["checksums.sha256"] + for name in REQUIRED_ARTIFACT_FILES: + if manifest["artifacts"][name] != digests[name]: + _invalid(f"{name} digest") + if name != "checksums.sha256" and checksums[name] != digests[name]: + _invalid(f"{name} checksum") + return {"bundle_integrity_valid": True, "content_safety_status": "not_evaluated", "manifest": manifest} diff --git a/tests/test_soxl_tqqq_evidence.py b/tests/test_soxl_tqqq_evidence.py index 2a033b7..92b2995 100644 --- a/tests/test_soxl_tqqq_evidence.py +++ b/tests/test_soxl_tqqq_evidence.py @@ -2,6 +2,9 @@ import math import importlib.util +import hashlib +import json +import os from pathlib import Path import pytest @@ -14,6 +17,7 @@ EvidenceValidationError = _evidence.EvidenceValidationError validate_evidence_manifest = _evidence.validate_evidence_manifest validate_metrics_payload = _evidence.validate_metrics_payload +validate_evidence_bundle = getattr(_evidence, "validate_evidence_bundle", None) def _manifest() -> dict[str, object]: @@ -82,3 +86,109 @@ def test_metrics_require_numeric_finite_evidence() -> None: validate_metrics_payload({"metrics": {"risk": {"value": value}}}) with pytest.raises(EvidenceValidationError): validate_metrics_payload({"metrics": {"note": "no numeric evidence", "ok": True}}) + + +def _write_bundle(root: Path) -> None: + root.mkdir(exist_ok=True) + manifest = _manifest() + identity = {field: manifest[field] for field in ("profile", "code_sha", "config_sha256", "plugin_sha256", "execution_timing", "timezone", "calendar", "as_of")} + identity["schema"] = "champion_identity.v1" + content = { + "champion_identity.json": json.dumps(identity), + "prices.csv": "date,value\n2026-07-13,1\n", + "data_quality.json": '{"ok":true}', + "daily_returns.csv": "date,value\n2026-07-13,0.1\n", + "targets.csv": "date,target\n2026-07-13,1\n", + "trades.csv": "date,symbol\n2026-07-13,SOXL\n", + "costs.csv": "date,cost\n2026-07-13,0.01\n", + "metrics.json": '{"metrics":{"sharpe":1.2}}', + "walk_forward.json": '{"windows":[]}', + "trial_ledger.jsonl": '{"trial":1}\n', + "robustness.json": '{"ok":true}', + "risk_sleeve.json": '{"ok":true}', + "strategy_performance.v2.json": '{"returns":[0.1]}', + } + for name, value in content.items(): + (root / name).write_text(value, encoding="utf-8") + checksums = "".join(f"{hashlib.sha256(content[name].encode()).hexdigest()} {name}\n" for name in sorted(content)) + (root / "checksums.sha256").write_text(checksums, encoding="ascii") + manifest["artifacts"] = {name: hashlib.sha256((root / name).read_bytes()).hexdigest() for name in REQUIRED_ARTIFACT_FILES} + (root / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8") + + +def test_valid_bundle_is_structure_only(tmp_path: Path) -> None: + _write_bundle(tmp_path) + assert validate_evidence_bundle is not None + result = validate_evidence_bundle(tmp_path) + assert result["bundle_integrity_valid"] is True + assert result["content_safety_status"] == "not_evaluated" + assert set(result) == {"bundle_integrity_valid", "content_safety_status", "manifest"} + assert "bundle_secret_free" not in result + + +def test_manifest_is_validated_before_other_artifacts(tmp_path: Path) -> None: + _write_bundle(tmp_path) + (tmp_path / "manifest.json").write_text("{}", encoding="utf-8") + (tmp_path / "prices.csv").write_text("bad\n", encoding="utf-8") + with pytest.raises(EvidenceValidationError, match="manifest"): + validate_evidence_bundle(tmp_path) + + +@pytest.mark.parametrize("kind", ["missing", "extra", "directory"]) +def test_bundle_root_has_exact_entries(tmp_path: Path, kind: str) -> None: + _write_bundle(tmp_path) + if kind == "missing": + (tmp_path / "prices.csv").unlink() + elif kind == "extra": + (tmp_path / "unexpected").write_text("x", encoding="utf-8") + else: + (tmp_path / "unexpected").mkdir() + with pytest.raises(EvidenceValidationError): + validate_evidence_bundle(tmp_path) + + +def test_bundle_rejects_symlink_and_duplicate_json(tmp_path: Path) -> None: + _write_bundle(tmp_path) + target = tmp_path / "prices.csv" + target.unlink() + os.symlink("costs.csv", target) + with pytest.raises(EvidenceValidationError): + validate_evidence_bundle(tmp_path) + _write_bundle(tmp_path / "nested") + (tmp_path / "nested" / "data_quality.json").write_text('{"x":1,"x":2}', encoding="utf-8") + with pytest.raises(EvidenceValidationError): + validate_evidence_bundle(tmp_path / "nested") + + +def test_bundle_checksum_champion_and_csv_shape_regressions(tmp_path: Path) -> None: + _write_bundle(tmp_path) + (tmp_path / "checksums.sha256").write_text("0" * 64 + " prices.csv\n", encoding="ascii") + with pytest.raises(EvidenceValidationError): + validate_evidence_bundle(tmp_path) + _write_bundle(tmp_path / "valid") + (tmp_path / "valid" / "trades.csv").write_text("a,a\n1,2\n", encoding="utf-8") + with pytest.raises(EvidenceValidationError): + validate_evidence_bundle(tmp_path / "valid") + + +def test_bundle_root_symlink_is_rejected(tmp_path: Path) -> None: + real = tmp_path / "real" + _write_bundle(real) + os.symlink(real, tmp_path / "link") + with pytest.raises(EvidenceValidationError): + validate_evidence_bundle(tmp_path / "link") + + +@pytest.mark.parametrize("name", _evidence._FILE_MAX_BYTES) +def test_each_file_cap_is_enforced(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, name: str) -> None: + _write_bundle(tmp_path) + monkeypatch.setitem(_evidence._FILE_MAX_BYTES, name, 0) + with pytest.raises(EvidenceValidationError): + validate_evidence_bundle(tmp_path) + + +def test_bundle_total_cap_is_enforced(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + _write_bundle(tmp_path) + monkeypatch.setattr(_evidence, "_MAX_BUNDLE_BYTES", 1) + with pytest.raises(EvidenceValidationError): + validate_evidence_bundle(tmp_path) From 9121209eb5757f5ecb9990f287591ca946dd45d9 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 14 Jul 2026 03:48:40 +0800 Subject: [PATCH 2/3] test: complete R0c integrity regression matrix Co-Authored-By: Codex --- tests/test_soxl_tqqq_evidence.py | 235 +++++++++++++++++++++---------- 1 file changed, 160 insertions(+), 75 deletions(-) diff --git a/tests/test_soxl_tqqq_evidence.py b/tests/test_soxl_tqqq_evidence.py index 92b2995..ad6b223 100644 --- a/tests/test_soxl_tqqq_evidence.py +++ b/tests/test_soxl_tqqq_evidence.py @@ -87,108 +87,193 @@ def test_metrics_require_numeric_finite_evidence() -> None: with pytest.raises(EvidenceValidationError): validate_metrics_payload({"metrics": {"note": "no numeric evidence", "ok": True}}) - def _write_bundle(root: Path) -> None: root.mkdir(exist_ok=True) manifest = _manifest() - identity = {field: manifest[field] for field in ("profile", "code_sha", "config_sha256", "plugin_sha256", "execution_timing", "timezone", "calendar", "as_of")} - identity["schema"] = "champion_identity.v1" - content = { - "champion_identity.json": json.dumps(identity), - "prices.csv": "date,value\n2026-07-13,1\n", - "data_quality.json": '{"ok":true}', - "daily_returns.csv": "date,value\n2026-07-13,0.1\n", - "targets.csv": "date,target\n2026-07-13,1\n", - "trades.csv": "date,symbol\n2026-07-13,SOXL\n", - "costs.csv": "date,cost\n2026-07-13,0.01\n", - "metrics.json": '{"metrics":{"sharpe":1.2}}', - "walk_forward.json": '{"windows":[]}', - "trial_ledger.jsonl": '{"trial":1}\n', - "robustness.json": '{"ok":true}', - "risk_sleeve.json": '{"ok":true}', - "strategy_performance.v2.json": '{"returns":[0.1]}', - } - for name, value in content.items(): - (root / name).write_text(value, encoding="utf-8") + fields = ("profile", "code_sha", "config_sha256", "plugin_sha256", "execution_timing", "timezone", "calendar", "as_of") + identity = {**{field: manifest[field] for field in fields}, "schema": "champion_identity.v1"} + content = {name: '{"ok":true}' for name in ("data_quality.json", "walk_forward.json", "robustness.json", "risk_sleeve.json")} + content.update({name: "a,b\n1,2\n" for name in ("prices.csv", "daily_returns.csv", "targets.csv", "trades.csv", "costs.csv")}) + content.update({"champion_identity.json": json.dumps(identity), "metrics.json": '{"metrics":{"sharpe":1.2}}', "strategy_performance.v2.json": '{"returns":[0.1]}', "trial_ledger.jsonl": '{"trial":1}\n'}) + for name, data in content.items(): + (root / name).write_text(data, encoding="utf-8") checksums = "".join(f"{hashlib.sha256(content[name].encode()).hexdigest()} {name}\n" for name in sorted(content)) (root / "checksums.sha256").write_text(checksums, encoding="ascii") manifest["artifacts"] = {name: hashlib.sha256((root / name).read_bytes()).hexdigest() for name in REQUIRED_ARTIFACT_FILES} (root / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8") -def test_valid_bundle_is_structure_only(tmp_path: Path) -> None: +def _refresh(root: Path) -> None: + names = sorted(set(REQUIRED_ARTIFACT_FILES) - {"checksums.sha256"}) + digests = {name: hashlib.sha256((root / name).read_bytes()).hexdigest() for name in names} + (root / "checksums.sha256").write_text("".join(f"{digests[n]} {n}\n" for n in names), encoding="ascii") + digests["checksums.sha256"] = hashlib.sha256((root / "checksums.sha256").read_bytes()).hexdigest() + manifest = json.loads((root / "manifest.json").read_text()) + manifest["artifacts"] = digests + (root / "manifest.json").write_text(json.dumps(manifest)) + + +def _reject(root: Path, name: str, data: str | bytes, refresh: bool = True) -> None: + path = root / name + if isinstance(data, bytes): + path.write_bytes(data) + else: + path.write_text(data, encoding="utf-8") + if refresh: + _refresh(root) + with pytest.raises(EvidenceValidationError, match=name.replace(".", r"\.")): + validate_evidence_bundle(root) + + +@pytest.fixture +def bundle(tmp_path: Path) -> Path: _write_bundle(tmp_path) - assert validate_evidence_bundle is not None - result = validate_evidence_bundle(tmp_path) - assert result["bundle_integrity_valid"] is True - assert result["content_safety_status"] == "not_evaluated" + return tmp_path + + +def test_valid_bundle_is_structure_only(bundle: Path) -> None: + result = validate_evidence_bundle(bundle) + assert result["bundle_integrity_valid"] is True and result["content_safety_status"] == "not_evaluated" assert set(result) == {"bundle_integrity_valid", "content_safety_status", "manifest"} assert "bundle_secret_free" not in result -def test_manifest_is_validated_before_other_artifacts(tmp_path: Path) -> None: - _write_bundle(tmp_path) - (tmp_path / "manifest.json").write_text("{}", encoding="utf-8") - (tmp_path / "prices.csv").write_text("bad\n", encoding="utf-8") +def test_manifest_and_root_precedence(bundle: Path, tmp_path: Path) -> None: + (bundle / "manifest.json").write_text("{}") + (bundle / "prices.csv").write_text("bad\n") with pytest.raises(EvidenceValidationError, match="manifest"): - validate_evidence_bundle(tmp_path) - - -@pytest.mark.parametrize("kind", ["missing", "extra", "directory"]) -def test_bundle_root_has_exact_entries(tmp_path: Path, kind: str) -> None: - _write_bundle(tmp_path) - if kind == "missing": - (tmp_path / "prices.csv").unlink() - elif kind == "extra": - (tmp_path / "unexpected").write_text("x", encoding="utf-8") - else: - (tmp_path / "unexpected").mkdir() - with pytest.raises(EvidenceValidationError): - validate_evidence_bundle(tmp_path) + validate_evidence_bundle(bundle) + for kind in ("missing", "extra", "directory"): + case = tmp_path / kind + _write_bundle(case) + if kind == "missing": + (case / "prices.csv").unlink() + elif kind == "extra": + (case / "unexpected").write_text("x") + else: + (case / "unexpected").mkdir() + with pytest.raises(EvidenceValidationError, match="bundle root"): + validate_evidence_bundle(case) -def test_bundle_rejects_symlink_and_duplicate_json(tmp_path: Path) -> None: - _write_bundle(tmp_path) - target = tmp_path / "prices.csv" +def test_links_and_fifo(bundle: Path, tmp_path: Path) -> None: + os.symlink(bundle, tmp_path / "link") + with pytest.raises(EvidenceValidationError, match="bundle root"): + validate_evidence_bundle(tmp_path / "link") + artifact = tmp_path / "artifact" + _write_bundle(artifact) + target = artifact / "prices.csv" target.unlink() os.symlink("costs.csv", target) - with pytest.raises(EvidenceValidationError): - validate_evidence_bundle(tmp_path) - _write_bundle(tmp_path / "nested") - (tmp_path / "nested" / "data_quality.json").write_text('{"x":1,"x":2}', encoding="utf-8") - with pytest.raises(EvidenceValidationError): - validate_evidence_bundle(tmp_path / "nested") + with pytest.raises(EvidenceValidationError, match="prices.csv"): + validate_evidence_bundle(artifact) + if hasattr(os, "mkfifo"): + _write_bundle(tmp_path / "fifo") + (tmp_path / "fifo" / "prices.csv").unlink() + os.mkfifo(tmp_path / "fifo" / "prices.csv") + with pytest.raises(EvidenceValidationError, match="prices.csv"): + validate_evidence_bundle(tmp_path / "fifo") -def test_bundle_checksum_champion_and_csv_shape_regressions(tmp_path: Path) -> None: - _write_bundle(tmp_path) - (tmp_path / "checksums.sha256").write_text("0" * 64 + " prices.csv\n", encoding="ascii") - with pytest.raises(EvidenceValidationError): - validate_evidence_bundle(tmp_path) - _write_bundle(tmp_path / "valid") - (tmp_path / "valid" / "trades.csv").write_text("a,a\n1,2\n", encoding="utf-8") - with pytest.raises(EvidenceValidationError): - validate_evidence_bundle(tmp_path / "valid") +@pytest.mark.parametrize("name", tuple(_evidence._FILE_MAX_BYTES)) +def test_file_caps_and_bundle_total(bundle: Path, monkeypatch: pytest.MonkeyPatch, name: str) -> None: + assert set(_evidence._FILE_MAX_BYTES) == {"manifest.json", *REQUIRED_ARTIFACT_FILES} + monkeypatch.setitem(_evidence._FILE_MAX_BYTES, name, (bundle / name).stat().st_size - 1) + with pytest.raises(EvidenceValidationError, match=name.replace(".", r"\.")): + validate_evidence_bundle(bundle) + monkeypatch.setitem(_evidence._FILE_MAX_BYTES, name, 10**9) + monkeypatch.setattr(_evidence, "_MAX_BUNDLE_BYTES", 1) + with pytest.raises(EvidenceValidationError, match="bundle total"): + validate_evidence_bundle(bundle) -def test_bundle_root_symlink_is_rejected(tmp_path: Path) -> None: - real = tmp_path / "real" - _write_bundle(real) - os.symlink(real, tmp_path / "link") - with pytest.raises(EvidenceValidationError): - validate_evidence_bundle(tmp_path / "link") +def test_streaming_contract(bundle: Path, monkeypatch: pytest.MonkeyPatch) -> None: + source = Path(_evidence.__file__).read_text() + assert all(x not in source for x in ("Path.read_text", "Path.read_bytes", ".read()")) + assert "read(_HASH_CHUNK_BYTES)" in source + monkeypatch.setattr(Path, "read_text", lambda *a, **k: (_ for _ in ()).throw(AssertionError())) + monkeypatch.setattr(Path, "read_bytes", lambda *a, **k: (_ for _ in ()).throw(AssertionError())) + validate_evidence_bundle(bundle) + +@pytest.mark.parametrize(("name", "data"), [ + ("data_quality.json", '{"a":1,"\\u0061":2}'), ("data_quality.json", '{"a":{"x":1,"x":2}}'), + ("data_quality.json", "not-json"), ("data_quality.json", "[]"), ("data_quality.json", "NaN"), + ("data_quality.json", b"{\xff}"), ("trial_ledger.jsonl", b""), ("trial_ledger.jsonl", b"\n"), + ("trial_ledger.jsonl", b"[]\n"), ("trial_ledger.jsonl", b'{"a":1}'), + ("trial_ledger.jsonl", b'{"a":1,"\\u0061":2}\n'), ("trades.csv", ",b\n1,2\n"), + ("trades.csv", "a,b\n"), ("trades.csv", "a,b\n\n"), ("trades.csv", "a,a\n1,2\n"), + ("trades.csv", "a,b\n1\n"), ("trades.csv", ""), ("data_quality.json", "depth")]) +def test_bounded_json_jsonl_csv(bundle: Path, name: str, data: str | bytes, monkeypatch: pytest.MonkeyPatch) -> None: + if data == "depth": + data = {} + node = data + for _ in range(65): + node["x"] = {} + node = node["x"] + data = json.dumps(data) + _reject(bundle, name, data) + monkeypatch.setattr(_evidence.json, "loads", lambda *a, **k: (_ for _ in ()).throw(RecursionError)) + with pytest.raises(EvidenceValidationError, match="manifest"): + validate_evidence_bundle(bundle) + + +def test_line_and_file_total_bounds(bundle: Path, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + monkeypatch.setattr(_evidence, "_MAX_LINE_BYTES", 4) + _reject(bundle, "trial_ledger.jsonl", b'{"x":1}\n', False) + monkeypatch.setattr(_evidence, "_MAX_LINE_BYTES", 12) + case = tmp_path / "csv" + _write_bundle(case) + _reject(case, "trades.csv", "a,b\n1234567890,2\n", False) + monkeypatch.setattr(_evidence, "_MAX_LINE_BYTES", 1_048_576) + case = tmp_path / "total" + _write_bundle(case) + monkeypatch.setitem(_evidence._FILE_MAX_BYTES, "trades.csv", 1) + _reject(case, "trades.csv", "a,b\n1,2\n", False) -@pytest.mark.parametrize("name", _evidence._FILE_MAX_BYTES) -def test_each_file_cap_is_enforced(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, name: str) -> None: + +def test_checksum_set_and_digest_edges(bundle: Path, tmp_path: Path) -> None: + valid = (bundle / "checksums.sha256").read_text().splitlines() + cases = [valid[:-1], valid + [valid[0]], valid[:1] + valid[:1] + valid[1:], list(reversed(valid)), + valid[:-1] + ["0" * 64 + " " + valid[0].split(" ")[1]], valid + ["0" * 64 + " checksums.sha256"], + valid + ["0" * 64 + " manifest.json"], [valid[0][:-1]], ["bad"]] + for lines in cases: + _write_bundle(tmp_path) + (tmp_path / "checksums.sha256").write_text("\n".join(lines) + "\n") + with pytest.raises(EvidenceValidationError, match="checksums\\.sha256"): + validate_evidence_bundle(tmp_path) _write_bundle(tmp_path) - monkeypatch.setitem(_evidence._FILE_MAX_BYTES, name, 0) - with pytest.raises(EvidenceValidationError): - validate_evidence_bundle(tmp_path) + _reject(tmp_path, "prices.csv", "a,b\n9,2\n", False) + + +@pytest.mark.parametrize("field", ["profile", "code_sha", "config_sha256", "plugin_sha256", "execution_timing", "timezone", "calendar", "as_of"]) +def test_champion_consistency(bundle: Path, field: str) -> None: + identity = json.loads((bundle / "champion_identity.json").read_text()) + identity[field] = "bad" + (bundle / "champion_identity.json").write_text(json.dumps(identity)) + _refresh(bundle) + with pytest.raises(EvidenceValidationError, match=f"champion identity {field}"): + validate_evidence_bundle(bundle) -def test_bundle_total_cap_is_enforced(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: +def test_champion_metrics_and_redaction(bundle: Path, tmp_path: Path) -> None: + for action in ("unknown", "missing"): + _write_bundle(tmp_path) + identity = json.loads((tmp_path / "champion_identity.json").read_text()) + if action == "unknown": + identity["unknown"] = 1 + else: + identity.pop("calendar") + (tmp_path / "champion_identity.json").write_text(json.dumps(identity)) + _refresh(tmp_path) + with pytest.raises(EvidenceValidationError, match="champion identity fields"): + validate_evidence_bundle(tmp_path) + for data in ('{"metrics":{"note":"x","ok":true}}', '{"metrics":{"x":NaN}}'): + _write_bundle(tmp_path) + _reject(tmp_path, "metrics.json", data) _write_bundle(tmp_path) - monkeypatch.setattr(_evidence, "_MAX_BUNDLE_BYTES", 1) - with pytest.raises(EvidenceValidationError): + needle = "/absolute/user/path?token=SECRET/" + "0" * 64 + (tmp_path / "data_quality.json").write_text('{"x":"' + needle + '"') + with pytest.raises(EvidenceValidationError) as error: validate_evidence_bundle(tmp_path) + assert needle not in str(error.value) and str(tmp_path) not in str(error.value) From e8df36b7673aedcd68f02f883e1773a54a01e80d Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 14 Jul 2026 04:22:19 +0800 Subject: [PATCH 3/3] fix: reject malformed quoted CSV evidence Co-Authored-By: Codex --- src/us_equity_snapshot_pipelines/evidence.py | 2 +- tests/test_soxl_tqqq_evidence.py | 7 +++++++ 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/src/us_equity_snapshot_pipelines/evidence.py b/src/us_equity_snapshot_pipelines/evidence.py index 6acabb4..2c0ae69 100644 --- a/src/us_equity_snapshot_pipelines/evidence.py +++ b/src/us_equity_snapshot_pipelines/evidence.py @@ -255,7 +255,7 @@ def _parse_csv(stream: Any, name: str, state: dict[str, int]) -> str: rows = 0 header: list[str] | None = None try: - for row in csv.reader(_physical_lines(stream, name, state, hasher)): + for row in csv.reader(_physical_lines(stream, name, state, hasher), strict=True): if header is None: if not row or any(not column for column in row) or len(set(row)) != len(row): _invalid(f"{name} header") diff --git a/tests/test_soxl_tqqq_evidence.py b/tests/test_soxl_tqqq_evidence.py index ad6b223..956604a 100644 --- a/tests/test_soxl_tqqq_evidence.py +++ b/tests/test_soxl_tqqq_evidence.py @@ -232,6 +232,13 @@ def test_line_and_file_total_bounds(bundle: Path, monkeypatch: pytest.MonkeyPatc _reject(case, "trades.csv", "a,b\n1,2\n", False) +def test_csv_rejects_unterminated_quote(bundle: Path) -> None: + (bundle / "trades.csv").write_text('a,b\n"unterminated,2\n') + _refresh(bundle) + with pytest.raises(EvidenceValidationError, match="trades\\.csv CSV"): + validate_evidence_bundle(bundle) + + def test_checksum_set_and_digest_edges(bundle: Path, tmp_path: Path) -> None: valid = (bundle / "checksums.sha256").read_text().splitlines() cases = [valid[:-1], valid + [valid[0]], valid[:1] + valid[:1] + valid[1:], list(reversed(valid)),