From 81cd1634218a2922d60da099022d8f5b4df4da2c Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 29 Jul 2026 14:36:58 +0800 Subject: [PATCH 1/3] feat: add verified TQQQ snapshot package boundary Co-Authored-By: Codex --- .../tqqq_trusted_snapshot_package.py | 227 ++++++++++++++++++ tests/test_tqqq_trusted_snapshot_package.py | 178 ++++++++++++++ 2 files changed, 405 insertions(+) create mode 100644 src/us_equity_snapshot_pipelines/tqqq_trusted_snapshot_package.py create mode 100644 tests/test_tqqq_trusted_snapshot_package.py diff --git a/src/us_equity_snapshot_pipelines/tqqq_trusted_snapshot_package.py b/src/us_equity_snapshot_pipelines/tqqq_trusted_snapshot_package.py new file mode 100644 index 0000000..c0b7e78 --- /dev/null +++ b/src/us_equity_snapshot_pipelines/tqqq_trusted_snapshot_package.py @@ -0,0 +1,227 @@ +"""Verified local boundary for a TQQQ R1 snapshot, receipt, and offline calendar.""" + +from __future__ import annotations + +import csv +import hashlib +import json +from dataclasses import dataclass +from datetime import date +from io import StringIO +from pathlib import Path +from typing import Any + +from . import tqqq_r1_snapshot + +_PACKAGE_VERSION = "tqqq_trusted_snapshot_package.v1" +_RECEIPT_VERSION = "tqqq_snapshot_receipt.v1" +_CALENDAR_VERSION = "tqqq_offline_calendar_evidence.v1" +_HEX_SHA256_LENGTH = 64 +_VERIFIED_CONSTRUCTION = object() + + +class TrustedSnapshotPackageError(ValueError): + """Raised when local package evidence cannot be verified.""" + + +@dataclass(frozen=True, init=False) +class TrustedSnapshotPackage: + """A package returned only after snapshot, receipt, and calendar verification.""" + + snapshot_dir: Path + session: str + snapshot_manifest_sha256: str + receipt_sha256: str + calendar_sha256: str + + def __init__( + self, + *, + _verified: object, + snapshot_dir: Path, + session: str, + snapshot_manifest_sha256: str, + receipt_sha256: str, + calendar_sha256: str, + ) -> None: + if _verified is not _VERIFIED_CONSTRUCTION: + raise TrustedSnapshotPackageError("TrustedSnapshotPackage requires verified loader") + object.__setattr__(self, "snapshot_dir", snapshot_dir) + object.__setattr__(self, "session", session) + object.__setattr__(self, "snapshot_manifest_sha256", snapshot_manifest_sha256) + object.__setattr__(self, "receipt_sha256", receipt_sha256) + object.__setattr__(self, "calendar_sha256", calendar_sha256) + + +def _invalid(message: str) -> None: + raise TrustedSnapshotPackageError(message) + + +def _no_duplicates(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + result: dict[str, Any] = {} + for key, value in pairs: + if key in result: + _invalid("invalid strict JSON") + result[key] = value + return result + + +def _reject_constant(_: str) -> None: + _invalid("invalid strict JSON") + + +def read_strict_json(payload: bytes, name: str) -> object: + """Decode JSON while rejecting duplicate keys and non-finite constants.""" + if type(payload) is not bytes or type(name) is not str: + _invalid("invalid strict JSON") + try: + return json.loads( + payload.decode("utf-8"), + object_pairs_hook=_no_duplicates, + parse_constant=_reject_constant, + ) + except (UnicodeDecodeError, json.JSONDecodeError, TrustedSnapshotPackageError) as exc: + if isinstance(exc, TrustedSnapshotPackageError): + raise + raise TrustedSnapshotPackageError(f"invalid strict JSON: {name}") from exc + + +def write_strict_json(path: str | Path, payload: object) -> None: + """Write canonical JSON and reject NaN/Infinity rather than serializing them.""" + destination = Path(path) + if destination.is_symlink(): + _invalid("strict JSON destination symlink is not allowed") + try: + encoded = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False).encode("utf-8") + except (TypeError, ValueError) as exc: + raise TrustedSnapshotPackageError("non-finite or invalid strict JSON") from exc + try: + destination.write_bytes(encoded + b"\n") + except OSError as exc: + raise TrustedSnapshotPackageError("unable to write strict JSON") from exc + + +def _read_regular(path: str | Path, name: str) -> bytes: + source = Path(path) + if source.is_symlink() or not source.is_file(): + _invalid(f"{name} must be a regular non-symlink file") + try: + return source.read_bytes() + except OSError as exc: + raise TrustedSnapshotPackageError(f"unable to read {name}") from exc + + +def _sha256(path: str | Path, name: str) -> str: + return hashlib.sha256(_read_regular(path, name)).hexdigest() + + +def _is_sha256(value: object) -> bool: + return type(value) is str and len(value) == _HEX_SHA256_LENGTH and all(character in "0123456789abcdef" for character in value) + + +def _session(value: object, name: str) -> str: + if type(value) is not str or len(value) != 10: + _invalid(f"invalid {name} session") + try: + parsed = date.fromisoformat(value) + except ValueError as exc: + raise TrustedSnapshotPackageError(f"invalid {name} session") from exc + if parsed.isoformat() != value: + _invalid(f"invalid {name} session") + if parsed.weekday() >= 5: + _invalid(f"{name} session must be a weekday") + return value + + +def _exact_object(value: object, expected_keys: set[str], name: str) -> dict[str, Any]: + if type(value) is not dict or set(value) != expected_keys: + _invalid(f"invalid {name}") + return value + + +def _package_manifest(value: object) -> dict[str, Any]: + manifest = _exact_object( + value, + {"contract_version", "snapshot_manifest_sha256", "receipt_sha256", "calendar_sha256", "session"}, + "package manifest", + ) + if manifest["contract_version"] != _PACKAGE_VERSION: + _invalid("invalid package manifest") + if not all(_is_sha256(manifest[key]) for key in ("snapshot_manifest_sha256", "receipt_sha256", "calendar_sha256")): + _invalid("invalid package manifest") + _session(manifest["session"], "package") + return manifest + + +def _receipt(value: object) -> dict[str, Any]: + receipt = _exact_object( + value, + {"contract_version", "snapshot_manifest_sha256", "calendar_sha256", "session"}, + "receipt", + ) + if receipt["contract_version"] != _RECEIPT_VERSION: + _invalid("invalid receipt") + if not all(_is_sha256(receipt[key]) for key in ("snapshot_manifest_sha256", "calendar_sha256")): + _invalid("invalid receipt") + _session(receipt["session"], "receipt") + return receipt + + +def _calendar(value: object) -> dict[str, Any]: + calendar = _exact_object(value, {"contract_version", "calendar", "sessions"}, "calendar evidence") + if calendar["contract_version"] != _CALENDAR_VERSION or calendar["calendar"] != "XNYS" or type(calendar["sessions"]) is not list: + _invalid("invalid calendar evidence") + sessions = [_session(session, "calendar") for session in calendar["sessions"]] + if not sessions or len(sessions) != len(set(sessions)) or sessions != sorted(sessions): + _invalid("invalid calendar evidence") + return calendar + + +def _snapshot_sessions(snapshot_dir: Path) -> set[str]: + try: + rows = csv.DictReader(StringIO(_read_regular(snapshot_dir / "prices.csv", "snapshot prices").decode("utf-8"))) + return {row["session"] for row in rows} + except (KeyError, UnicodeDecodeError, csv.Error) as exc: + raise TrustedSnapshotPackageError("invalid verified snapshot prices") from exc + + +def load_verified_trusted_snapshot_package( + snapshot_dir: str | Path, + package_manifest_path: str | Path, + receipt_path: str | Path, + calendar_evidence_path: str | Path, + *, + expected_snapshot_manifest_sha256: str, +) -> TrustedSnapshotPackage: + """Verify all local evidence against the caller-owned snapshot hash trust anchor.""" + if not _is_sha256(expected_snapshot_manifest_sha256): + _invalid("invalid expected snapshot manifest hash") + verified_snapshot = tqqq_r1_snapshot.verify_tqqq_r1_snapshot( + snapshot_dir, + expected_manifest_sha256=expected_snapshot_manifest_sha256, + ) + manifest = _package_manifest(read_strict_json(_read_regular(package_manifest_path, "package manifest"), "package manifest")) + receipt = _receipt(read_strict_json(_read_regular(receipt_path, "receipt"), "receipt")) + calendar = _calendar(read_strict_json(_read_regular(calendar_evidence_path, "calendar evidence"), "calendar evidence")) + receipt_sha256 = _sha256(receipt_path, "receipt") + calendar_sha256 = _sha256(calendar_evidence_path, "calendar evidence") + + if manifest["snapshot_manifest_sha256"] != expected_snapshot_manifest_sha256 or receipt["snapshot_manifest_sha256"] != expected_snapshot_manifest_sha256: + _invalid("snapshot manifest hash binding mismatch") + if manifest["receipt_sha256"] != receipt_sha256: + _invalid("receipt hash binding mismatch") + if manifest["calendar_sha256"] != calendar_sha256 or receipt["calendar_sha256"] != calendar_sha256: + _invalid("calendar hash binding mismatch") + if manifest["session"] != receipt["session"] or manifest["session"] not in calendar["sessions"]: + _invalid("calendar session binding mismatch") + if manifest["session"] not in _snapshot_sessions(verified_snapshot.output_dir): + _invalid("snapshot session binding mismatch") + + return TrustedSnapshotPackage( + _verified=_VERIFIED_CONSTRUCTION, + snapshot_dir=verified_snapshot.output_dir, + session=manifest["session"], + snapshot_manifest_sha256=expected_snapshot_manifest_sha256, + receipt_sha256=receipt_sha256, + calendar_sha256=calendar_sha256, + ) diff --git a/tests/test_tqqq_trusted_snapshot_package.py b/tests/test_tqqq_trusted_snapshot_package.py new file mode 100644 index 0000000..b75cc92 --- /dev/null +++ b/tests/test_tqqq_trusted_snapshot_package.py @@ -0,0 +1,178 @@ +from __future__ import annotations + +import hashlib +from pathlib import Path + +import pandas as pd +import pytest + +from us_equity_snapshot_pipelines import tqqq_r1_snapshot as snapshot +from us_equity_snapshot_pipelines import tqqq_trusted_snapshot_package as package + + +def _prices() -> pd.DataFrame: + return pd.DataFrame( + [ + {"session": "2010-01-04", "symbol": "QQQ", "adjusted_close": 45.25}, + {"session": "2010-01-04", "symbol": "TQQQ", "adjusted_close": 10.5}, + {"session": "2010-01-05", "symbol": "QQQ", "adjusted_close": 46.0}, + {"session": "2010-01-05", "symbol": "TQQQ", "adjusted_close": 11.0}, + ] + ) + + +def _sha256(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def _write_bound_package(tmp_path: Path, *, session: str = "2010-01-05") -> tuple[Path, Path, Path, Path, str]: + snapshot_result = snapshot.materialize_tqqq_r1_snapshot(_prices(), tmp_path / "snapshot") + calendar_path = tmp_path / "calendar.json" + package.write_strict_json( + calendar_path, + { + "contract_version": "tqqq_offline_calendar_evidence.v1", + "calendar": "XNYS", + "sessions": ["2010-01-04", session], + }, + ) + receipt_path = tmp_path / "receipt.json" + package.write_strict_json( + receipt_path, + { + "contract_version": "tqqq_snapshot_receipt.v1", + "snapshot_manifest_sha256": snapshot_result.manifest_sha256, + "calendar_sha256": _sha256(calendar_path), + "session": session, + }, + ) + manifest_path = tmp_path / "package-manifest.json" + package.write_strict_json( + manifest_path, + { + "contract_version": "tqqq_trusted_snapshot_package.v1", + "snapshot_manifest_sha256": snapshot_result.manifest_sha256, + "receipt_sha256": _sha256(receipt_path), + "calendar_sha256": _sha256(calendar_path), + "session": session, + }, + ) + return snapshot_result.output_dir, manifest_path, receipt_path, calendar_path, snapshot_result.manifest_sha256 + + +@pytest.mark.parametrize("value", [float("nan"), float("inf"), float("-inf")]) +def test_strict_json_writer_rejects_non_finite_numbers(tmp_path: Path, value: float) -> None: + with pytest.raises(package.TrustedSnapshotPackageError, match="non-finite"): + package.write_strict_json(tmp_path / "payload.json", {"value": value}) + + +@pytest.mark.parametrize("payload", [b'{"value":NaN}', b'{"value":Infinity}', b'{"value":-Infinity}']) +def test_strict_json_reader_rejects_non_finite_numbers(payload: bytes) -> None: + with pytest.raises(package.TrustedSnapshotPackageError, match="invalid strict JSON"): + package.read_strict_json(payload, "payload") + + +def test_verified_loader_preserves_existing_weekend_rejection(tmp_path: Path) -> None: + snapshot_dir, manifest_path, receipt_path, calendar_path, trusted_manifest_sha256 = _write_bound_package(tmp_path, session="2010-01-03") + + with pytest.raises(package.TrustedSnapshotPackageError, match="weekday"): + package.load_verified_trusted_snapshot_package( + snapshot_dir, + manifest_path, + receipt_path, + calendar_path, + expected_snapshot_manifest_sha256=trusted_manifest_sha256, + ) + + +def test_public_verified_loader_returns_only_verified_package(tmp_path: Path) -> None: + snapshot_dir, manifest_path, receipt_path, calendar_path, trusted_manifest_sha256 = _write_bound_package(tmp_path) + + trusted = package.load_verified_trusted_snapshot_package( + snapshot_dir, + manifest_path, + receipt_path, + calendar_path, + expected_snapshot_manifest_sha256=trusted_manifest_sha256, + ) + + assert isinstance(trusted, package.TrustedSnapshotPackage) + assert trusted.session == "2010-01-05" + assert trusted.snapshot_manifest_sha256 == snapshot.verify_tqqq_r1_snapshot( + snapshot_dir, + expected_manifest_sha256=trusted.snapshot_manifest_sha256, + ).manifest_sha256 + with pytest.raises(TypeError): + package.TrustedSnapshotPackage() # type: ignore[call-arg] + + +def test_caller_cannot_substitute_calendar_provenance_after_receipt_is_accepted(tmp_path: Path) -> None: + snapshot_dir, manifest_path, receipt_path, calendar_path, trusted_manifest_sha256 = _write_bound_package(tmp_path) + package.write_strict_json( + calendar_path, + { + "contract_version": "tqqq_offline_calendar_evidence.v1", + "calendar": "XNYS", + "sessions": ["2010-01-04", "2010-01-05", "2010-01-06"], + }, + ) + + with pytest.raises(package.TrustedSnapshotPackageError, match="calendar hash"): + package.load_verified_trusted_snapshot_package( + snapshot_dir, + manifest_path, + receipt_path, + calendar_path, + expected_snapshot_manifest_sha256=trusted_manifest_sha256, + ) + + +def test_accepted_offline_calendar_evidence_is_bound_to_session_and_package(tmp_path: Path) -> None: + snapshot_dir, manifest_path, receipt_path, calendar_path, trusted_manifest_sha256 = _write_bound_package(tmp_path) + package.write_strict_json( + calendar_path, + { + "contract_version": "tqqq_offline_calendar_evidence.v1", + "calendar": "XNYS", + "sessions": ["2010-01-04"], + }, + ) + receipt = package.read_strict_json(receipt_path.read_bytes(), "receipt") + assert isinstance(receipt, dict) + receipt["calendar_sha256"] = _sha256(calendar_path) + package.write_strict_json(receipt_path, receipt) + manifest = package.read_strict_json(manifest_path.read_bytes(), "package manifest") + assert isinstance(manifest, dict) + manifest["calendar_sha256"] = _sha256(calendar_path) + manifest["receipt_sha256"] = _sha256(receipt_path) + package.write_strict_json(manifest_path, manifest) + + with pytest.raises(package.TrustedSnapshotPackageError, match="calendar session binding"): + package.load_verified_trusted_snapshot_package( + snapshot_dir, + manifest_path, + receipt_path, + calendar_path, + expected_snapshot_manifest_sha256=trusted_manifest_sha256, + ) + + +def test_caller_cannot_substitute_receipt_snapshot_provenance(tmp_path: Path) -> None: + snapshot_dir, manifest_path, receipt_path, calendar_path, trusted_manifest_sha256 = _write_bound_package(tmp_path) + receipt = package.read_strict_json(receipt_path.read_bytes(), "receipt") + assert isinstance(receipt, dict) + receipt["snapshot_manifest_sha256"] = "0" * 64 + package.write_strict_json(receipt_path, receipt) + manifest = package.read_strict_json(manifest_path.read_bytes(), "package manifest") + assert isinstance(manifest, dict) + manifest["receipt_sha256"] = _sha256(receipt_path) + package.write_strict_json(manifest_path, manifest) + + with pytest.raises(package.TrustedSnapshotPackageError, match="snapshot manifest hash binding"): + package.load_verified_trusted_snapshot_package( + snapshot_dir, + manifest_path, + receipt_path, + calendar_path, + expected_snapshot_manifest_sha256=trusted_manifest_sha256, + ) From cb7908b82e83e80e9d6ffc46f3b7678a574c63b0 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 29 Jul 2026 17:20:25 +0800 Subject: [PATCH 2/3] fix: harden trusted TQQQ package verification Co-Authored-By: Codex --- .../tqqq_trusted_snapshot_package.py | 62 ++++++--- tests/test_tqqq_trusted_snapshot_package.py | 131 ++++++++++-------- 2 files changed, 114 insertions(+), 79 deletions(-) diff --git a/src/us_equity_snapshot_pipelines/tqqq_trusted_snapshot_package.py b/src/us_equity_snapshot_pipelines/tqqq_trusted_snapshot_package.py index c0b7e78..fbf7d27 100644 --- a/src/us_equity_snapshot_pipelines/tqqq_trusted_snapshot_package.py +++ b/src/us_equity_snapshot_pipelines/tqqq_trusted_snapshot_package.py @@ -111,8 +111,9 @@ def _read_regular(path: str | Path, name: str) -> bytes: raise TrustedSnapshotPackageError(f"unable to read {name}") from exc -def _sha256(path: str | Path, name: str) -> str: - return hashlib.sha256(_read_regular(path, name)).hexdigest() +def _read_evidence(path: str | Path, name: str) -> tuple[object, str]: + payload = _read_regular(path, name) + return read_strict_json(payload, name), hashlib.sha256(payload).hexdigest() def _is_sha256(value: object) -> bool: @@ -192,21 +193,46 @@ def load_verified_trusted_snapshot_package( calendar_evidence_path: str | Path, *, expected_snapshot_manifest_sha256: str, + expected_package_manifest_sha256: str, + expected_receipt_sha256: str, ) -> TrustedSnapshotPackage: - """Verify all local evidence against the caller-owned snapshot hash trust anchor.""" - if not _is_sha256(expected_snapshot_manifest_sha256): - _invalid("invalid expected snapshot manifest hash") - verified_snapshot = tqqq_r1_snapshot.verify_tqqq_r1_snapshot( - snapshot_dir, - expected_manifest_sha256=expected_snapshot_manifest_sha256, - ) - manifest = _package_manifest(read_strict_json(_read_regular(package_manifest_path, "package manifest"), "package manifest")) - receipt = _receipt(read_strict_json(_read_regular(receipt_path, "receipt"), "receipt")) - calendar = _calendar(read_strict_json(_read_regular(calendar_evidence_path, "calendar evidence"), "calendar evidence")) - receipt_sha256 = _sha256(receipt_path, "receipt") - calendar_sha256 = _sha256(calendar_evidence_path, "calendar evidence") - - if manifest["snapshot_manifest_sha256"] != expected_snapshot_manifest_sha256 or receipt["snapshot_manifest_sha256"] != expected_snapshot_manifest_sha256: + """Verify all local evidence against caller-owned external trust anchors.""" + if not all( + _is_sha256(value) + for value in ( + expected_snapshot_manifest_sha256, + expected_package_manifest_sha256, + expected_receipt_sha256, + ) + ): + _invalid("invalid expected evidence hash") + try: + verified_snapshot = tqqq_r1_snapshot.verify_tqqq_r1_snapshot( + snapshot_dir, + expected_manifest_sha256=expected_snapshot_manifest_sha256, + ) + except tqqq_r1_snapshot.SnapshotValidationError as exc: + raise TrustedSnapshotPackageError("invalid verified snapshot") from exc + try: + resolved_snapshot_dir = verified_snapshot.output_dir.resolve(strict=True) + except OSError as exc: + raise TrustedSnapshotPackageError("unable to resolve verified snapshot") from exc + + manifest_value, package_manifest_sha256 = _read_evidence(package_manifest_path, "package manifest") + receipt_value, receipt_sha256 = _read_evidence(receipt_path, "receipt") + calendar_value, calendar_sha256 = _read_evidence(calendar_evidence_path, "calendar evidence") + if package_manifest_sha256 != expected_package_manifest_sha256: + _invalid("package manifest hash binding mismatch") + if receipt_sha256 != expected_receipt_sha256: + _invalid("receipt hash binding mismatch") + manifest = _package_manifest(manifest_value) + receipt = _receipt(receipt_value) + calendar = _calendar(calendar_value) + + if ( + manifest["snapshot_manifest_sha256"] != expected_snapshot_manifest_sha256 + or receipt["snapshot_manifest_sha256"] != expected_snapshot_manifest_sha256 + ): _invalid("snapshot manifest hash binding mismatch") if manifest["receipt_sha256"] != receipt_sha256: _invalid("receipt hash binding mismatch") @@ -214,12 +240,12 @@ def load_verified_trusted_snapshot_package( _invalid("calendar hash binding mismatch") if manifest["session"] != receipt["session"] or manifest["session"] not in calendar["sessions"]: _invalid("calendar session binding mismatch") - if manifest["session"] not in _snapshot_sessions(verified_snapshot.output_dir): + if manifest["session"] not in _snapshot_sessions(resolved_snapshot_dir): _invalid("snapshot session binding mismatch") return TrustedSnapshotPackage( _verified=_VERIFIED_CONSTRUCTION, - snapshot_dir=verified_snapshot.output_dir, + snapshot_dir=resolved_snapshot_dir, session=manifest["session"], snapshot_manifest_sha256=expected_snapshot_manifest_sha256, receipt_sha256=receipt_sha256, diff --git a/tests/test_tqqq_trusted_snapshot_package.py b/tests/test_tqqq_trusted_snapshot_package.py index b75cc92..ed97fd7 100644 --- a/tests/test_tqqq_trusted_snapshot_package.py +++ b/tests/test_tqqq_trusted_snapshot_package.py @@ -60,6 +60,27 @@ def _write_bound_package(tmp_path: Path, *, session: str = "2010-01-05") -> tupl return snapshot_result.output_dir, manifest_path, receipt_path, calendar_path, snapshot_result.manifest_sha256 +def _load( + snapshot_dir: Path, + manifest_path: Path, + receipt_path: Path, + calendar_path: Path, + snapshot_manifest_sha256: str, + *, + package_manifest_sha256: str | None = None, + receipt_sha256: str | None = None, +) -> package.TrustedSnapshotPackage: + return package.load_verified_trusted_snapshot_package( + snapshot_dir, + manifest_path, + receipt_path, + calendar_path, + expected_snapshot_manifest_sha256=snapshot_manifest_sha256, + expected_package_manifest_sha256=package_manifest_sha256 or _sha256(manifest_path), + expected_receipt_sha256=receipt_sha256 or _sha256(receipt_path), + ) + + @pytest.mark.parametrize("value", [float("nan"), float("inf"), float("-inf")]) def test_strict_json_writer_rejects_non_finite_numbers(tmp_path: Path, value: float) -> None: with pytest.raises(package.TrustedSnapshotPackageError, match="non-finite"): @@ -73,41 +94,33 @@ def test_strict_json_reader_rejects_non_finite_numbers(payload: bytes) -> None: def test_verified_loader_preserves_existing_weekend_rejection(tmp_path: Path) -> None: - snapshot_dir, manifest_path, receipt_path, calendar_path, trusted_manifest_sha256 = _write_bound_package(tmp_path, session="2010-01-03") + args = _write_bound_package(tmp_path, session="2010-01-03") with pytest.raises(package.TrustedSnapshotPackageError, match="weekday"): - package.load_verified_trusted_snapshot_package( - snapshot_dir, - manifest_path, - receipt_path, - calendar_path, - expected_snapshot_manifest_sha256=trusted_manifest_sha256, - ) + _load(*args) def test_public_verified_loader_returns_only_verified_package(tmp_path: Path) -> None: - snapshot_dir, manifest_path, receipt_path, calendar_path, trusted_manifest_sha256 = _write_bound_package(tmp_path) + args = _write_bound_package(tmp_path) - trusted = package.load_verified_trusted_snapshot_package( - snapshot_dir, - manifest_path, - receipt_path, - calendar_path, - expected_snapshot_manifest_sha256=trusted_manifest_sha256, - ) + trusted = _load(*args) assert isinstance(trusted, package.TrustedSnapshotPackage) assert trusted.session == "2010-01-05" + assert trusted.snapshot_dir.is_absolute() assert trusted.snapshot_manifest_sha256 == snapshot.verify_tqqq_r1_snapshot( - snapshot_dir, + args[0], expected_manifest_sha256=trusted.snapshot_manifest_sha256, ).manifest_sha256 with pytest.raises(TypeError): package.TrustedSnapshotPackage() # type: ignore[call-arg] -def test_caller_cannot_substitute_calendar_provenance_after_receipt_is_accepted(tmp_path: Path) -> None: - snapshot_dir, manifest_path, receipt_path, calendar_path, trusted_manifest_sha256 = _write_bound_package(tmp_path) +def test_loader_requires_external_package_manifest_and_receipt_anchors(tmp_path: Path) -> None: + args = _write_bound_package(tmp_path) + snapshot_dir, manifest_path, receipt_path, calendar_path, snapshot_manifest_sha256 = args + package_manifest_sha256 = _sha256(manifest_path) + receipt_sha256 = _sha256(receipt_path) package.write_strict_json( calendar_path, { @@ -116,27 +129,6 @@ def test_caller_cannot_substitute_calendar_provenance_after_receipt_is_accepted( "sessions": ["2010-01-04", "2010-01-05", "2010-01-06"], }, ) - - with pytest.raises(package.TrustedSnapshotPackageError, match="calendar hash"): - package.load_verified_trusted_snapshot_package( - snapshot_dir, - manifest_path, - receipt_path, - calendar_path, - expected_snapshot_manifest_sha256=trusted_manifest_sha256, - ) - - -def test_accepted_offline_calendar_evidence_is_bound_to_session_and_package(tmp_path: Path) -> None: - snapshot_dir, manifest_path, receipt_path, calendar_path, trusted_manifest_sha256 = _write_bound_package(tmp_path) - package.write_strict_json( - calendar_path, - { - "contract_version": "tqqq_offline_calendar_evidence.v1", - "calendar": "XNYS", - "sessions": ["2010-01-04"], - }, - ) receipt = package.read_strict_json(receipt_path.read_bytes(), "receipt") assert isinstance(receipt, dict) receipt["calendar_sha256"] = _sha256(calendar_path) @@ -147,32 +139,49 @@ def test_accepted_offline_calendar_evidence_is_bound_to_session_and_package(tmp_ manifest["receipt_sha256"] = _sha256(receipt_path) package.write_strict_json(manifest_path, manifest) - with pytest.raises(package.TrustedSnapshotPackageError, match="calendar session binding"): - package.load_verified_trusted_snapshot_package( + with pytest.raises(package.TrustedSnapshotPackageError, match="package manifest hash"): + _load( snapshot_dir, manifest_path, receipt_path, calendar_path, - expected_snapshot_manifest_sha256=trusted_manifest_sha256, + snapshot_manifest_sha256, + package_manifest_sha256=package_manifest_sha256, + receipt_sha256=receipt_sha256, ) -def test_caller_cannot_substitute_receipt_snapshot_provenance(tmp_path: Path) -> None: - snapshot_dir, manifest_path, receipt_path, calendar_path, trusted_manifest_sha256 = _write_bound_package(tmp_path) - receipt = package.read_strict_json(receipt_path.read_bytes(), "receipt") - assert isinstance(receipt, dict) - receipt["snapshot_manifest_sha256"] = "0" * 64 - package.write_strict_json(receipt_path, receipt) - manifest = package.read_strict_json(manifest_path.read_bytes(), "package manifest") - assert isinstance(manifest, dict) - manifest["receipt_sha256"] = _sha256(receipt_path) - package.write_strict_json(manifest_path, manifest) +def test_loader_hashes_the_same_evidence_bytes_it_validates(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + args = _write_bound_package(tmp_path) + reads: dict[Path, int] = {} + read_regular = package._read_regular - with pytest.raises(package.TrustedSnapshotPackageError, match="snapshot manifest hash binding"): - package.load_verified_trusted_snapshot_package( - snapshot_dir, - manifest_path, - receipt_path, - calendar_path, - expected_snapshot_manifest_sha256=trusted_manifest_sha256, - ) + def count_evidence_reads(path: str | Path, name: str) -> bytes: + source = Path(path) + if source in set(args[1:4]): + reads[source] = reads.get(source, 0) + 1 + if reads[source] > 1: + raise AssertionError(f"evidence reread: {source.name}") + return read_regular(path, name) + + monkeypatch.setattr(package, "_read_regular", count_evidence_reads) + + _load(*args) + + assert reads == {path: 1 for path in args[1:4]} + + +def test_verified_package_stores_resolved_snapshot_path(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + args = _write_bound_package(tmp_path) + monkeypatch.chdir(tmp_path) + + trusted = _load(Path("snapshot"), *args[1:]) + + assert trusted.snapshot_dir == args[0].resolve(strict=True) + + +def test_snapshot_validation_failures_are_normalized(tmp_path: Path) -> None: + _, manifest_path, receipt_path, calendar_path, _ = _write_bound_package(tmp_path) + + with pytest.raises(package.TrustedSnapshotPackageError, match="invalid verified snapshot"): + _load(tmp_path / "missing-snapshot", manifest_path, receipt_path, calendar_path, "0" * 64) From ea2bddba357ffcfd4ca868efc943483f7f2097c7 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 29 Jul 2026 17:24:55 +0800 Subject: [PATCH 3/3] fix: reject overflowed strict JSON floats Co-Authored-By: Codex --- .../tqqq_trusted_snapshot_package.py | 9 +++++++++ tests/test_tqqq_trusted_snapshot_package.py | 5 +++++ 2 files changed, 14 insertions(+) diff --git a/src/us_equity_snapshot_pipelines/tqqq_trusted_snapshot_package.py b/src/us_equity_snapshot_pipelines/tqqq_trusted_snapshot_package.py index fbf7d27..326327b 100644 --- a/src/us_equity_snapshot_pipelines/tqqq_trusted_snapshot_package.py +++ b/src/us_equity_snapshot_pipelines/tqqq_trusted_snapshot_package.py @@ -5,6 +5,7 @@ import csv import hashlib import json +import math from dataclasses import dataclass from datetime import date from io import StringIO @@ -70,6 +71,13 @@ def _reject_constant(_: str) -> None: _invalid("invalid strict JSON") +def _parse_finite_float(value: str) -> float: + parsed = float(value) + if not math.isfinite(parsed): + _invalid("invalid strict JSON") + return parsed + + def read_strict_json(payload: bytes, name: str) -> object: """Decode JSON while rejecting duplicate keys and non-finite constants.""" if type(payload) is not bytes or type(name) is not str: @@ -79,6 +87,7 @@ def read_strict_json(payload: bytes, name: str) -> object: payload.decode("utf-8"), object_pairs_hook=_no_duplicates, parse_constant=_reject_constant, + parse_float=_parse_finite_float, ) except (UnicodeDecodeError, json.JSONDecodeError, TrustedSnapshotPackageError) as exc: if isinstance(exc, TrustedSnapshotPackageError): diff --git a/tests/test_tqqq_trusted_snapshot_package.py b/tests/test_tqqq_trusted_snapshot_package.py index ed97fd7..54a9087 100644 --- a/tests/test_tqqq_trusted_snapshot_package.py +++ b/tests/test_tqqq_trusted_snapshot_package.py @@ -93,6 +93,11 @@ def test_strict_json_reader_rejects_non_finite_numbers(payload: bytes) -> None: package.read_strict_json(payload, "payload") +def test_strict_json_reader_rejects_overflowed_float() -> None: + with pytest.raises(package.TrustedSnapshotPackageError, match="invalid strict JSON"): + package.read_strict_json(b'{"value":1e400}', "x") + + def test_verified_loader_preserves_existing_weekend_rejection(tmp_path: Path) -> None: args = _write_bound_package(tmp_path, session="2010-01-03")