diff --git a/src/us_equity_snapshot_pipelines/tqqq_r1_snapshot.py b/src/us_equity_snapshot_pipelines/tqqq_r1_snapshot.py index d39c865..dd5997b 100644 --- a/src/us_equity_snapshot_pipelines/tqqq_r1_snapshot.py +++ b/src/us_equity_snapshot_pipelines/tqqq_r1_snapshot.py @@ -4,14 +4,16 @@ import ctypes import errno +import fcntl import hashlib import json import math import os import shutil +import stat import sys import tempfile -from dataclasses import dataclass +from dataclasses import dataclass, field from io import BytesIO from pathlib import Path from typing import Any @@ -36,6 +38,7 @@ class SnapshotValidationError(ValueError): class SnapshotResult: output_dir: Path manifest_sha256: str + verified_members: tuple[tuple[str, bytes], ...] = field(default=(), repr=False, compare=False) def _sha256(path: Path) -> str: @@ -268,27 +271,12 @@ def _has_exact_type(value: object, expected: object) -> bool: return value == expected -def verify_tqqq_r1_snapshot( - output_dir: str | Path, +def _verify_snapshot_members( + members: dict[str, bytes], *, + output_dir: Path, expected_manifest_sha256: str, ) -> SnapshotResult: - output = Path(output_dir) - if output.is_symlink(): - _invalid("snapshot root symlink is not allowed") - try: - names = tuple(sorted(path.name for path in output.iterdir())) if output.is_dir() else () - except OSError as exc: - raise SnapshotValidationError("unable to read snapshot members") from exc - if names != tuple(sorted(OUTPUT_FILENAMES)): - _invalid(f"unexpected output files: {names}") - if any(not (output / name).is_file() or (output / name).is_symlink() for name in OUTPUT_FILENAMES): - _invalid("snapshot members must be regular non-symlink files") - try: - members = {name: (output / name).read_bytes() for name in OUTPUT_FILENAMES} - except OSError as exc: - raise SnapshotValidationError("unable to read snapshot members") from exc - member_hashes = {name: hashlib.sha256(content).hexdigest() for name, content in members.items()} if type(expected_manifest_sha256) is not str or member_hashes["manifest.json"] != expected_manifest_sha256: _invalid("trusted manifest hash mismatch") @@ -330,7 +318,94 @@ def verify_tqqq_r1_snapshot( expected_validation = {"valid": True, "row_count": len(prices), "symbols": list(SYMBOLS)} if not _has_exact_type(validation, expected_validation): _invalid("invalid validation") - return SnapshotResult(output_dir=output, manifest_sha256=member_hashes["manifest.json"]) + return SnapshotResult( + output_dir=output_dir, + manifest_sha256=member_hashes["manifest.json"], + verified_members=tuple(sorted(members.items())), + ) + + +def read_tqqq_r1_snapshot_member_fd(directory_fd: int, name: str) -> bytes: + if name not in OUTPUT_FILENAMES: + _invalid("invalid snapshot member") + try: + member_fd = os.open(name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=directory_fd) + except OSError as exc: + raise SnapshotValidationError("unable to read snapshot members") from exc + try: + member_stat = os.fstat(member_fd) + if not stat.S_ISREG(member_stat.st_mode): + _invalid("snapshot members must be regular non-symlink files") + chunks: list[bytes] = [] + while True: + chunk = os.read(member_fd, 1024 * 1024) + if not chunk: + return b"".join(chunks) + chunks.append(chunk) + except OSError as exc: + raise SnapshotValidationError("unable to read snapshot members") from exc + finally: + os.close(member_fd) + + +def verify_tqqq_r1_snapshot_fd( + directory_fd: int, + *, + expected_manifest_sha256: str, +) -> SnapshotResult: + """Verify snapshot members relative to an already-open stable directory.""" + if type(directory_fd) is not int or directory_fd < 0: + _invalid("invalid snapshot directory descriptor") + try: + names = tuple(sorted(os.listdir(directory_fd))) + except OSError as exc: + raise SnapshotValidationError("unable to read snapshot members") from exc + if names != tuple(sorted(OUTPUT_FILENAMES)): + _invalid(f"unexpected output files: {names}") + members = {name: read_tqqq_r1_snapshot_member_fd(directory_fd, name) for name in OUTPUT_FILENAMES} + try: + if Path("/proc/self/fd").is_dir(): + output_dir = Path(f"/proc/self/fd/{directory_fd}").resolve(strict=True) + elif sys.platform == "darwin": + output_dir = Path( + fcntl.fcntl(directory_fd, fcntl.F_GETPATH, b"\0" * 1024).split(b"\0", 1)[0].decode() + ) + else: + _invalid("stable snapshot directory capability is unavailable") + except (OSError, UnicodeDecodeError) as exc: + raise SnapshotValidationError("unable to resolve snapshot directory descriptor") from exc + return _verify_snapshot_members( + members, + output_dir=output_dir, + expected_manifest_sha256=expected_manifest_sha256, + ) + + +def verify_tqqq_r1_snapshot( + output_dir: str | Path, + *, + expected_manifest_sha256: str, +) -> SnapshotResult: + output = Path(output_dir) + if output.is_symlink(): + _invalid("snapshot root symlink is not allowed") + try: + names = tuple(sorted(path.name for path in output.iterdir())) if output.is_dir() else () + except OSError as exc: + raise SnapshotValidationError("unable to read snapshot members") from exc + if names != tuple(sorted(OUTPUT_FILENAMES)): + _invalid(f"unexpected output files: {names}") + if any(not (output / name).is_file() or (output / name).is_symlink() for name in OUTPUT_FILENAMES): + _invalid("snapshot members must be regular non-symlink files") + try: + members = {name: (output / name).read_bytes() for name in OUTPUT_FILENAMES} + except OSError as exc: + raise SnapshotValidationError("unable to read snapshot members") from exc + return _verify_snapshot_members( + members, + output_dir=output, + expected_manifest_sha256=expected_manifest_sha256, + ) def materialize_tqqq_r1_snapshot( 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..7b3d20d --- /dev/null +++ b/src/us_equity_snapshot_pipelines/tqqq_trusted_snapshot_package.py @@ -0,0 +1,338 @@ +"""Verified local boundary for a TQQQ R1 snapshot, receipt, and offline calendar.""" + +from __future__ import annotations + +import csv +import hashlib +import json +import math +import os +import sys +try: + import fcntl +except ImportError: # pragma: no cover - unsupported platforms fail closed below. + fcntl = None # type: ignore[assignment] +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 + _snapshot_fd: int + _snapshot_members: tuple[tuple[str, bytes], ...] + + def __init__( + self, + *, + _verified: object, + snapshot_dir: Path, + session: str, + snapshot_manifest_sha256: str, + receipt_sha256: str, + calendar_sha256: str, + snapshot_fd: int, + snapshot_members: tuple[tuple[str, bytes], ...], + ) -> 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) + object.__setattr__(self, "_snapshot_fd", snapshot_fd) + object.__setattr__(self, "_snapshot_members", snapshot_members) + + def close(self) -> None: + """Release the stable snapshot directory descriptor.""" + fd = self._snapshot_fd + if fd >= 0: + try: + os.close(fd) + except OSError: + pass + object.__setattr__(self, "_snapshot_fd", -1) + + def read_snapshot_member(self, name: str) -> bytes: + """Read a verified snapshot member through the stable directory descriptor.""" + if self._snapshot_fd < 0: + raise TrustedSnapshotPackageError("trusted snapshot package is closed") + members = dict(self._snapshot_members) + if name not in members: + raise TrustedSnapshotPackageError("invalid verified snapshot member") + return members[name] + + def __enter__(self) -> "TrustedSnapshotPackage": + return self + + def __exit__(self, _type: object, _value: object, _traceback: object) -> None: + self.close() + + def __del__(self) -> None: + try: + self.close() + except (AttributeError, TypeError): + pass + + +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 _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: + _invalid("invalid strict JSON") + try: + return json.loads( + payload.decode("utf-8"), + object_pairs_hook=_no_duplicates, + parse_constant=_reject_constant, + parse_float=_parse_finite_float, + ) + except TrustedSnapshotPackageError: + raise + except (UnicodeDecodeError, json.JSONDecodeError, ValueError) as exc: + 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) + 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: + fd = os.open(destination, os.O_WRONLY | os.O_CREAT | os.O_TRUNC | os.O_NOFOLLOW, 0o666) + with os.fdopen(fd, "wb") as output: + output.write(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 _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 _open_stable_snapshot_dir(snapshot_dir: str | Path) -> tuple[int, Path]: + """Open a directory descriptor and return a display path plus stable fd.""" + directory_flag = getattr(os, "O_DIRECTORY", 0) + no_follow_flag = getattr(os, "O_NOFOLLOW", 0) + if not directory_flag or not no_follow_flag: + _invalid("stable snapshot directory capability is unavailable") + try: + fd = os.open(os.fspath(snapshot_dir), os.O_RDONLY | directory_flag | no_follow_flag) + except FileNotFoundError as exc: + raise TrustedSnapshotPackageError("invalid verified snapshot") from exc + except OSError as exc: + raise TrustedSnapshotPackageError("unable to open stable snapshot directory") from exc + try: + if Path("/proc/self/fd").is_dir(): + display_path = Path(f"/proc/self/fd/{fd}") + elif sys.platform == "darwin" and fcntl is not None: + display_path = Path(fcntl.fcntl(fd, fcntl.F_GETPATH, b"\0" * 1024).split(b"\0", 1)[0].decode()) + else: + raise OSError("stable directory display path unavailable") + except (OSError, UnicodeDecodeError) as exc: + os.close(fd) + raise TrustedSnapshotPackageError("stable snapshot directory capability is unavailable") from exc + return fd, display_path + + +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 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, + expected_package_manifest_sha256: str, + expected_receipt_sha256: str, +) -> TrustedSnapshotPackage: + """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") + snapshot_fd, stable_snapshot_dir = _open_stable_snapshot_dir(snapshot_dir) + try: + try: + verified_snapshot = tqqq_r1_snapshot.verify_tqqq_r1_snapshot_fd( + snapshot_fd, + expected_manifest_sha256=expected_snapshot_manifest_sha256, + ) + except tqqq_r1_snapshot.SnapshotValidationError as exc: + raise TrustedSnapshotPackageError("invalid verified snapshot") from exc + + package_manifest_payload = _read_regular(package_manifest_path, "package manifest") + receipt_payload = _read_regular(receipt_path, "receipt") + calendar_payload = _read_regular(calendar_evidence_path, "calendar evidence") + package_manifest_sha256 = hashlib.sha256(package_manifest_payload).hexdigest() + receipt_sha256 = hashlib.sha256(receipt_payload).hexdigest() + calendar_sha256 = hashlib.sha256(calendar_payload).hexdigest() + 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(read_strict_json(package_manifest_payload, "package manifest")) + if manifest["calendar_sha256"] != calendar_sha256: + _invalid("calendar hash binding mismatch") + receipt = _receipt(read_strict_json(receipt_payload, "receipt")) + calendar = _calendar(read_strict_json(calendar_payload, "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") + try: + snapshot_members = dict(verified_snapshot.verified_members) + snapshot_sessions = { + row["session"] + for row in csv.DictReader(StringIO(snapshot_members["prices.csv"].decode("utf-8"))) + } + except (KeyError, UnicodeDecodeError, csv.Error) as exc: + raise TrustedSnapshotPackageError("invalid verified snapshot prices") from exc + if manifest["session"] not in snapshot_sessions: + _invalid("snapshot session binding mismatch") + + return TrustedSnapshotPackage( + _verified=_VERIFIED_CONSTRUCTION, + snapshot_dir=stable_snapshot_dir, + session=manifest["session"], + snapshot_manifest_sha256=expected_snapshot_manifest_sha256, + receipt_sha256=receipt_sha256, + calendar_sha256=calendar_sha256, + snapshot_fd=snapshot_fd, + snapshot_members=verified_snapshot.verified_members, + ) + except Exception: + os.close(snapshot_fd) + raise diff --git a/tests/test_tqqq_r1_snapshot.py b/tests/test_tqqq_r1_snapshot.py index 8ae631d..4b6f2c3 100644 --- a/tests/test_tqqq_r1_snapshot.py +++ b/tests/test_tqqq_r1_snapshot.py @@ -2,6 +2,7 @@ import hashlib import json +import os from datetime import datetime, timezone from pathlib import Path @@ -42,6 +43,20 @@ def _refresh_trusted_metadata(output_dir: Path) -> str: return hashlib.sha256(manifest_path.read_bytes()).hexdigest() +def test_descriptor_verifier_returns_a_usable_platform_path(tmp_path: Path) -> None: + result = snapshot.materialize_tqqq_r1_snapshot(_fixture_prices(), tmp_path / "snapshot") + directory_fd = os.open(result.output_dir, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) + try: + verified = snapshot.verify_tqqq_r1_snapshot_fd( + directory_fd, + expected_manifest_sha256=result.manifest_sha256, + ) + finally: + os.close(directory_fd) + + assert verified.output_dir.is_dir() + + def test_materialize_writes_deterministic_immutable_artifacts(tmp_path: Path) -> None: result = snapshot.materialize_tqqq_r1_snapshot(_fixture_prices(), tmp_path / "snapshot") diff --git a/tests/test_tqqq_trusted_snapshot_package.py b/tests/test_tqqq_trusted_snapshot_package.py new file mode 100644 index 0000000..ad2fb40 --- /dev/null +++ b/tests/test_tqqq_trusted_snapshot_package.py @@ -0,0 +1,286 @@ +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 + + +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"): + 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_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") + + with pytest.raises(package.TrustedSnapshotPackageError, match="weekday"): + _load(*args) + + +def test_public_verified_loader_returns_only_verified_package(tmp_path: Path) -> None: + args = _write_bound_package(tmp_path) + + 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( + args[0], + expected_manifest_sha256=trusted.snapshot_manifest_sha256, + ).manifest_sha256 + with pytest.raises(TypeError): + package.TrustedSnapshotPackage() # type: ignore[call-arg] + + +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, + { + "contract_version": "tqqq_offline_calendar_evidence.v1", + "calendar": "XNYS", + "sessions": ["2010-01-04", "2010-01-05", "2010-01-06"], + }, + ) + 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="package manifest hash"): + _load( + snapshot_dir, + manifest_path, + receipt_path, + calendar_path, + snapshot_manifest_sha256, + package_manifest_sha256=package_manifest_sha256, + receipt_sha256=receipt_sha256, + ) + + +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 + + 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.is_absolute() + assert b"45.25" in trusted.read_snapshot_member("prices.csv") + + +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) + + +def test_strict_json_reader_normalizes_oversized_integer(tmp_path: Path) -> None: + payload = b'{"value":' + (b"7" * 5000) + b"}" + + with pytest.raises(package.TrustedSnapshotPackageError, match="invalid strict JSON"): + package.read_strict_json(payload, "oversized") + + +def test_loader_authenticates_evidence_before_decoding(tmp_path: Path) -> None: + args = _write_bound_package(tmp_path) + snapshot_dir, manifest_path, receipt_path, calendar_path, snapshot_manifest_sha256 = args + receipt_sha256 = _sha256(receipt_path) + receipt_path.write_bytes(b'{"value":' + (b"8" * 5000) + b"}") + + with pytest.raises(package.TrustedSnapshotPackageError, match="receipt hash binding mismatch"): + _load( + snapshot_dir, + manifest_path, + receipt_path, + calendar_path, + snapshot_manifest_sha256, + receipt_sha256=receipt_sha256, + ) + + +def test_loader_keeps_verified_snapshot_directory_stable_after_replacement( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + args = _write_bound_package(tmp_path) + snapshot_dir, manifest_path, receipt_path, calendar_path, snapshot_manifest_sha256 = args + verify = package.tqqq_r1_snapshot.verify_tqqq_r1_snapshot_fd + + def verify_then_replace(directory_fd: int, *, expected_manifest_sha256: str): + result = verify(directory_fd, expected_manifest_sha256=expected_manifest_sha256) + original = Path(snapshot_dir) + original.rename(tmp_path / "verified-original") + replacement = original + replacement.mkdir() + (replacement / "prices.csv").write_text( + "session,symbol,adjusted_close\n2010-01-05,QQQ,999\n", + encoding="utf-8", + ) + return result + + monkeypatch.setattr(package.tqqq_r1_snapshot, "verify_tqqq_r1_snapshot_fd", verify_then_replace) + + trusted = _load(args[0], *args[1:]) + + assert trusted.snapshot_dir.is_absolute() + assert b"999" not in trusted.read_snapshot_member("prices.csv") + + +def test_package_preserves_verified_member_bytes_after_member_replacement(tmp_path: Path) -> None: + args = _write_bound_package(tmp_path) + trusted = _load(*args) + (args[0] / "prices.csv").write_text( + "session,symbol,adjusted_close\n2010-01-05,QQQ,999\n", + encoding="utf-8", + ) + + assert b"999" not in trusted.read_snapshot_member("prices.csv") + + +def test_snapshot_members_are_opened_nonblocking(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + args = _write_bound_package(tmp_path) + open_file = snapshot.os.open + + def require_nonblocking(path: str | bytes | Path, flags: int, *values: object, **options: object) -> int: + if path in snapshot.OUTPUT_FILENAMES: + assert flags & snapshot.os.O_NONBLOCK + return open_file(path, flags, *values, **options) + + monkeypatch.setattr(snapshot.os, "open", require_nonblocking) + + _load(*args) + + +def test_strict_json_writer_does_not_follow_raced_symlink( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + target = tmp_path / "target.json" + target.write_text("unchanged\n", encoding="utf-8") + destination = tmp_path / "destination.json" + destination.symlink_to(target) + monkeypatch.setattr(Path, "is_symlink", lambda _self: False) + + with pytest.raises(package.TrustedSnapshotPackageError, match="unable to write strict JSON"): + package.write_strict_json(destination, {"value": 1}) + + assert target.read_text(encoding="utf-8") == "unchanged\n"