diff --git a/src/us_equity_snapshot_pipelines/tqqq_r1_snapshot.py b/src/us_equity_snapshot_pipelines/tqqq_r1_snapshot.py index d39c865..d93a50a 100644 --- a/src/us_equity_snapshot_pipelines/tqqq_r1_snapshot.py +++ b/src/us_equity_snapshot_pipelines/tqqq_r1_snapshot.py @@ -9,15 +9,21 @@ 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 import pandas as pd +try: + import fcntl +except ImportError: # pragma: no cover - unsupported platforms fail closed when verification is requested. + fcntl = None # type: ignore[assignment] + CONTRACT_VERSION = "tqqq_r1_qqq_tqqq_immutable_snapshot.v2" SYMBOLS = ("QQQ", "TQQQ") REQUESTED_LOWER_BOUND = "2010-01-01" @@ -25,6 +31,7 @@ PLUGIN = "ABSENT_DISABLED" MODE = "core_only" OUTPUT_FILENAMES = ("prices.csv", "manifest.json", "validation.json", "sha256sums.json") +MAX_SNAPSHOT_MEMBER_BYTES = 16 * 1024 * 1024 _MAX_FLOAT_INTEGER_DECIMAL = format(sys.float_info.max, ".0f") @@ -36,6 +43,9 @@ class SnapshotValidationError(ValueError): class SnapshotResult: output_dir: Path manifest_sha256: str + verified_members: tuple[tuple[str, bytes], ...] = field(default=(), repr=False, compare=False) + + __hash__ = None def _sha256(path: Path) -> str: @@ -268,27 +278,66 @@ def _has_exact_type(value: object, expected: object) -> bool: return value == expected -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") +def _descriptor_flags(*, directory: bool = False) -> int: + no_follow = getattr(os, "O_NOFOLLOW", 0) + non_blocking = getattr(os, "O_NONBLOCK", 0) + directory_flag = getattr(os, "O_DIRECTORY", 0) if directory else 0 + if not no_follow or not non_blocking or (directory and not directory_flag): + _invalid("descriptor-safe snapshot capability is unavailable") + return os.O_RDONLY | no_follow | non_blocking | directory_flag + + +def _snapshot_display_path(directory_fd: int) -> Path: + try: + if sys.platform.startswith("linux") and Path("/proc/self/fd").is_dir(): + return Path(f"/proc/self/fd/{directory_fd}") + if sys.platform == "darwin" and fcntl is not None and hasattr(fcntl, "F_GETPATH"): + return Path(fcntl.fcntl(directory_fd, fcntl.F_GETPATH, b"\0" * 1024).split(b"\0", 1)[0].decode()) + except (OSError, UnicodeDecodeError) as exc: + raise SnapshotValidationError("unable to resolve snapshot directory descriptor") from exc + _invalid("stable snapshot directory capability is unavailable") + + +def read_tqqq_r1_snapshot_member_fd(directory_fd: int, name: str) -> bytes: + """Read one regular snapshot member once through a trusted directory descriptor.""" + if type(directory_fd) is not int or directory_fd < 0 or name not in OUTPUT_FILENAMES: + _invalid("invalid snapshot member") try: - names = tuple(sorted(path.name for path in output.iterdir())) if output.is_dir() else () + member_fd = os.open(name, _descriptor_flags(), dir_fd=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}") - 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} + initial = os.fstat(member_fd) + if not stat.S_ISREG(initial.st_mode): + _invalid("snapshot members must be regular non-symlink files") + if initial.st_size < 0 or initial.st_size > MAX_SNAPSHOT_MEMBER_BYTES: + _invalid("snapshot member exceeds size limit") + chunks: list[bytes] = [] + received = 0 + while True: + chunk = os.read(member_fd, min(1024 * 1024, MAX_SNAPSHOT_MEMBER_BYTES - received + 1)) + if not chunk: + break + received += len(chunk) + if received > MAX_SNAPSHOT_MEMBER_BYTES: + _invalid("snapshot member exceeds size limit") + chunks.append(chunk) + final = os.fstat(member_fd) + if final.st_size != initial.st_size or received != initial.st_size: + _invalid("snapshot member changed while reading") + return b"".join(chunks) except OSError as exc: raise SnapshotValidationError("unable to read snapshot members") from exc + finally: + os.close(member_fd) + +def _verify_snapshot_members( + members: dict[str, bytes], + *, + output_dir: Path, + expected_manifest_sha256: str, +) -> SnapshotResult: 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 +379,48 @@ 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 verify_tqqq_r1_snapshot_fd( + directory_fd: int, + *, + expected_manifest_sha256: str, +) -> SnapshotResult: + """Verify one snapshot using only the supplied stable directory descriptor.""" + 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} + return _verify_snapshot_members( + members, + output_dir=_snapshot_display_path(directory_fd), + expected_manifest_sha256=expected_manifest_sha256, + ) + + +def verify_tqqq_r1_snapshot( + output_dir: str | Path, + *, + expected_manifest_sha256: str, +) -> SnapshotResult: + try: + directory_fd = os.open(os.fspath(output_dir), _descriptor_flags(directory=True)) + except OSError as exc: + raise SnapshotValidationError("unable to read snapshot members") from exc + try: + return verify_tqqq_r1_snapshot_fd(directory_fd, expected_manifest_sha256=expected_manifest_sha256) + finally: + os.close(directory_fd) 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..5d44ba5 --- /dev/null +++ b/src/us_equity_snapshot_pipelines/tqqq_trusted_snapshot_package.py @@ -0,0 +1,367 @@ +"""Descriptor-backed local verification boundary for TQQQ R1 snapshot evidence.""" + +from __future__ import annotations + +import csv +import errno +import hashlib +import json +import math +import os +import stat +import sys +from dataclasses import dataclass, field +from datetime import date +from io import StringIO +from pathlib import Path +from typing import Any + +try: + import fcntl +except ImportError: # pragma: no cover - unsupported platforms fail closed when loading. + fcntl = None # type: ignore[assignment] + +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 +MAX_EVIDENCE_BYTES = 1024 * 1024 +_MAX_JSON_INTEGER_DIGITS = 309 +_VERIFIED_CONSTRUCTION = object() + + +class TrustedSnapshotPackageError(ValueError): + """Raised when local package evidence cannot be verified.""" + + +@dataclass(frozen=True, init=False) +class TrustedSnapshotPackage: + """Verified evidence and one owned directory descriptor; never reopens member paths.""" + + snapshot_dir: Path + session: str + snapshot_manifest_sha256: str + receipt_sha256: str + calendar_sha256: str + _snapshot_fd: int = field(repr=False, compare=False) + _snapshot_members: tuple[tuple[str, bytes], ...] = field(repr=False, compare=False) + + __hash__ = None + + 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 sole owned descriptor; repeated close calls are harmless.""" + descriptor = self._snapshot_fd + if descriptor >= 0: + try: + os.close(descriptor) + except OSError: + pass + object.__setattr__(self, "_snapshot_fd", -1) + + def read_snapshot_member(self, name: str) -> bytes: + """Return the already verified bytes without reopening a mutable filesystem path.""" + if self._snapshot_fd < 0: + raise TrustedSnapshotPackageError("trusted snapshot package is closed") + members = dict(self._snapshot_members) + if type(name) is not str or 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 __copy__(self) -> TrustedSnapshotPackage: + raise TrustedSnapshotPackageError("trusted snapshot package cannot be copied") + + def __deepcopy__(self, _memo: dict[int, object]) -> TrustedSnapshotPackage: + raise TrustedSnapshotPackageError("trusted snapshot package cannot be copied") + + def __reduce_ex__(self, _protocol: int) -> object: + raise TrustedSnapshotPackageError("trusted snapshot package cannot be pickled") + + def __del__(self) -> None: + try: + self.close() + except (AttributeError, TypeError): + pass + + +def _invalid(message: str) -> None: + raise TrustedSnapshotPackageError(message) + + +def _safe_open_flags(*, directory: bool = False, write: bool = False) -> int: + no_follow = getattr(os, "O_NOFOLLOW", 0) + non_blocking = getattr(os, "O_NONBLOCK", 0) + directory_flag = getattr(os, "O_DIRECTORY", 0) if directory else 0 + if not no_follow or not non_blocking or (directory and not directory_flag): + _invalid("descriptor-safe trusted-input capability is unavailable") + flags = (os.O_WRONLY | os.O_CREAT | os.O_TRUNC) if write else os.O_RDONLY + return flags | no_follow | non_blocking | directory_flag + + +def _read_regular(path: str | Path, name: str) -> bytes: + """Open one evidence file once, admit its descriptor, then return those exact bytes.""" + try: + descriptor = os.open(os.fspath(path), _safe_open_flags()) + except OSError as exc: + if exc.errno == errno.ELOOP: + _invalid(f"{name} must be a regular non-symlink file") + raise TrustedSnapshotPackageError(f"unable to read {name}") from exc + try: + initial = os.fstat(descriptor) + if not stat.S_ISREG(initial.st_mode): + _invalid(f"{name} must be a regular non-symlink file") + if initial.st_size < 0 or initial.st_size > MAX_EVIDENCE_BYTES: + _invalid(f"{name} exceeds size limit") + chunks: list[bytes] = [] + received = 0 + while True: + chunk = os.read(descriptor, min(1024 * 1024, MAX_EVIDENCE_BYTES - received + 1)) + if not chunk: + break + received += len(chunk) + if received > MAX_EVIDENCE_BYTES: + _invalid(f"{name} exceeds size limit") + chunks.append(chunk) + final = os.fstat(descriptor) + if final.st_size != initial.st_size or received != initial.st_size: + _invalid(f"{name} changed while reading") + return b"".join(chunks) + except OSError as exc: + raise TrustedSnapshotPackageError(f"unable to read {name}") from exc + finally: + os.close(descriptor) + + +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 _parse_bounded_int(value: str) -> int: + if len(value.lstrip("-")) > _MAX_JSON_INTEGER_DIGITS: + _invalid("invalid strict JSON") + return int(value) + + +def read_strict_json(payload: bytes, name: str) -> object: + """Decode finite, duplicate-free JSON with normalized untrusted-input errors.""" + 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, + parse_int=_parse_bounded_int, + ) + except TrustedSnapshotPackageError: + raise + except (UnicodeDecodeError, ValueError, json.JSONDecodeError) as exc: + raise TrustedSnapshotPackageError(f"invalid strict JSON: {name}") from exc + + +def write_strict_json(path: str | Path, payload: object) -> None: + """Write finite canonical JSON without following a raced destination symlink.""" + 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: + descriptor = os.open(os.fspath(path), _safe_open_flags(write=True), 0o600) + with os.fdopen(descriptor, "wb") as output: + output.write(encoded + b"\n") + except OSError as exc: + raise TrustedSnapshotPackageError("unable to write strict JSON") from exc + + +def _stable_snapshot_dir(snapshot_dir: str | Path) -> tuple[int, Path]: + try: + descriptor = os.open(os.fspath(snapshot_dir), _safe_open_flags(directory=True)) + except OSError as exc: + raise TrustedSnapshotPackageError("unable to open stable snapshot directory") from exc + try: + if sys.platform.startswith("linux") and Path("/proc/self/fd").is_dir(): + display_path = Path(f"/proc/self/fd/{descriptor}") + elif sys.platform == "darwin" and fcntl is not None and hasattr(fcntl, "F_GETPATH"): + display_path = Path(fcntl.fcntl(descriptor, fcntl.F_GETPATH, b"\0" * 1024).split(b"\0", 1)[0].decode()) + else: + _invalid("stable snapshot directory capability is unavailable") + except (OSError, UnicodeDecodeError, TrustedSnapshotPackageError) as exc: + os.close(descriptor) + raise TrustedSnapshotPackageError("stable snapshot directory capability is unavailable") from exc + return descriptor, 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 or parsed.weekday() >= 5: + _invalid(f"invalid {name} session") + 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(item, "calendar") for item 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 anchored local evidence and return a non-copyable stable trusted package.""" + 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 = _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")) + receipt = _receipt(read_strict_json(receipt_payload, "receipt")) + calendar = _calendar(read_strict_json(calendar_payload, "calendar evidence")) + if manifest["calendar_sha256"] != calendar_sha256 or receipt["calendar_sha256"] != calendar_sha256: + _invalid("calendar hash binding mismatch") + 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["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..0777eb6 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 @@ -264,3 +265,56 @@ def test_materialize_rejects_python_int_beyond_digit_limit(tmp_path: Path) -> No with pytest.raises(snapshot.SnapshotValidationError): snapshot.materialize_tqqq_r1_snapshot(prices, tmp_path / "snapshot") + + +def test_verify_uses_descriptor_backed_member_bytes_after_member_replacement(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) + (result.output_dir / "prices.csv").write_text("replaced\n", encoding="utf-8") + + assert b"45.25" in dict(verified.verified_members)["prices.csv"] + assert "45.25" not in repr(verified) + with pytest.raises(TypeError): + hash(verified) + + +def test_descriptor_verifier_rejects_fifo_member_without_blocking(tmp_path: Path) -> None: + result = snapshot.materialize_tqqq_r1_snapshot(_fixture_prices(), tmp_path / "snapshot") + prices_path = result.output_dir / "prices.csv" + prices_path.unlink() + os.mkfifo(prices_path) + directory_fd = os.open(result.output_dir, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) + try: + with pytest.raises(snapshot.SnapshotValidationError, match="regular non-symlink"): + snapshot.verify_tqqq_r1_snapshot_fd(directory_fd, expected_manifest_sha256=result.manifest_sha256) + finally: + os.close(directory_fd) + + +def test_descriptor_member_reader_requires_nonblocking_and_size_admission( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + result = snapshot.materialize_tqqq_r1_snapshot(_fixture_prices(), tmp_path / "snapshot") + open_file = snapshot.os.open + observed_flags: list[int] = [] + + def observe_member_open(path: str | bytes | Path, flags: int, *args: object, **kwargs: object) -> int: + if path in snapshot.OUTPUT_FILENAMES: + observed_flags.append(flags) + return open_file(path, flags, *args, **kwargs) + + monkeypatch.setattr(snapshot.os, "open", observe_member_open) + directory_fd = os.open(result.output_dir, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) + try: + snapshot.verify_tqqq_r1_snapshot_fd(directory_fd, expected_manifest_sha256=result.manifest_sha256) + finally: + os.close(directory_fd) + + assert observed_flags and all(flags & os.O_NOFOLLOW and flags & os.O_NONBLOCK for flags in observed_flags) diff --git a/tests/test_tqqq_trusted_snapshot_package.py b/tests/test_tqqq_trusted_snapshot_package.py new file mode 100644 index 0000000..9a12265 --- /dev/null +++ b/tests/test_tqqq_trusted_snapshot_package.py @@ -0,0 +1,172 @@ +from __future__ import annotations + +import copy +import hashlib +import os +import pickle +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("payload", [b'{"value":NaN}', b'{"value":1e400}', b'{"value":' + b"7" * 5000 + b"}"]) +def test_strict_json_reader_normalizes_nonfinite_and_huge_numbers(payload: bytes) -> None: + with pytest.raises(package.TrustedSnapshotPackageError, match="invalid strict JSON"): + package.read_strict_json(payload, "payload") + + +def test_loader_authenticates_evidence_before_decoding(tmp_path: Path) -> None: + args = _write_bound_package(tmp_path) + receipt_sha256 = _sha256(args[2]) + args[2].write_bytes(b'{"value":' + b"8" * 5000 + b"}") + + with pytest.raises(package.TrustedSnapshotPackageError, match="receipt hash binding mismatch"): + _load(*args, receipt_sha256=receipt_sha256) + + +def test_loader_keeps_verified_bytes_after_member_and_directory_replacement(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + args = _write_bound_package(tmp_path) + verified = package.tqqq_r1_snapshot.verify_tqqq_r1_snapshot_fd + + def verify_then_replace(directory_fd: int, *, expected_manifest_sha256: str): + result = verified(directory_fd, expected_manifest_sha256=expected_manifest_sha256) + args[0].rename(tmp_path / "verified-original") + args[0].mkdir() + (args[0] / "prices.csv").write_text("replaced\n", encoding="utf-8") + return result + + monkeypatch.setattr(package.tqqq_r1_snapshot, "verify_tqqq_r1_snapshot_fd", verify_then_replace) + trusted = _load(*args) + + assert b"45.25" in trusted.read_snapshot_member("prices.csv") + assert "45.25" not in repr(trusted) + + +def test_loader_rejects_fifo_member_without_blocking(tmp_path: Path) -> None: + args = _write_bound_package(tmp_path) + prices_path = args[0] / "prices.csv" + prices_path.unlink() + os.mkfifo(prices_path) + + with pytest.raises(package.TrustedSnapshotPackageError, match="invalid verified snapshot"): + _load(*args) + + +def test_loader_rejects_symlinked_evidence(tmp_path: Path) -> None: + args = _write_bound_package(tmp_path) + target = tmp_path / "target.json" + target.write_bytes(args[1].read_bytes()) + args[1].unlink() + args[1].symlink_to(target) + + with pytest.raises(package.TrustedSnapshotPackageError, match="regular non-symlink"): + _load(*args) + + +def test_verified_package_owns_descriptor_and_rejects_copy_pickle_and_hash(tmp_path: Path) -> None: + trusted = _load(*_write_bound_package(tmp_path)) + before_close = trusted + + with pytest.raises(TypeError): + hash(trusted) + with pytest.raises(package.TrustedSnapshotPackageError): + copy.copy(trusted) + with pytest.raises(package.TrustedSnapshotPackageError): + copy.deepcopy(trusted) + with pytest.raises(package.TrustedSnapshotPackageError): + pickle.dumps(trusted) + + trusted.close() + trusted.close() + assert trusted == before_close + with pytest.raises(package.TrustedSnapshotPackageError, match="closed"): + trusted.read_snapshot_member("prices.csv") + + +def test_loader_uses_single_nonblocking_regular_evidence_open(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + args = _write_bound_package(tmp_path) + open_file = package.os.open + evidence_names = {os.fspath(path) for path in args[1:4]} + observed: list[int] = [] + + def observe(path: str | bytes | Path, flags: int, *values: object, **options: object) -> int: + if os.fspath(path) in evidence_names: + observed.append(flags) + return open_file(path, flags, *values, **options) + + monkeypatch.setattr(package.os, "open", observe) + _load(*args) + + assert len(observed) == 3 + assert all(flags & os.O_NOFOLLOW and flags & os.O_NONBLOCK for flags in observed)