diff --git a/src/us_equity_snapshot_pipelines/tqqq_r1_snapshot.py b/src/us_equity_snapshot_pipelines/tqqq_r1_snapshot.py index d39c865..03607c3 100644 --- a/src/us_equity_snapshot_pipelines/tqqq_r1_snapshot.py +++ b/src/us_equity_snapshot_pipelines/tqqq_r1_snapshot.py @@ -9,6 +9,7 @@ import math import os import shutil +import stat import sys import tempfile from dataclasses import dataclass @@ -25,6 +26,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") @@ -268,26 +270,64 @@ def _has_exact_type(value: object, expected: object) -> bool: return value == expected -def verify_tqqq_r1_snapshot( - output_dir: str | Path, +def _snapshot_open_flags(*, directory: bool = False) -> int: + if os.name != "posix": + _invalid("descriptor-safe snapshot verification is unavailable") + 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 verification is unavailable") + return os.O_RDONLY | no_follow | non_blocking | directory_flag + + +def _read_snapshot_member_fd(directory_fd: int, name: str) -> bytes: + """Read one bounded regular member via an already-open directory descriptor.""" + try: + descriptor = os.open(name, _snapshot_open_flags(), dir_fd=directory_fd) + except OSError as exc: + raise SnapshotValidationError("unable to read snapshot members") from exc + try: + metadata = os.fstat(descriptor) + if not stat.S_ISREG(metadata.st_mode): + _invalid("snapshot members must be regular non-symlink files") + if metadata.st_size < 0 or metadata.st_size > MAX_SNAPSHOT_MEMBER_BYTES: + _invalid("snapshot member exceeds size limit") + chunks: list[bytes] = [] + received = 0 + while True: + chunk = os.read(descriptor, 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(descriptor) + if final.st_size != metadata.st_size or received != metadata.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(descriptor) + + +def _verify_tqqq_r1_snapshot_fd( + directory_fd: int, *, expected_manifest_sha256: str, -) -> SnapshotResult: - output = Path(output_dir) - if output.is_symlink(): - _invalid("snapshot root symlink is not allowed") +) -> tuple[str, tuple[tuple[str, bytes], ...]]: + """Verify a snapshot from one stable directory descriptor without path reopening.""" + if type(directory_fd) is not int or directory_fd < 0: + _invalid("invalid snapshot directory descriptor") try: - names = tuple(sorted(path.name for path in output.iterdir())) if output.is_dir() else () + 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}") - 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 + members = {name: _read_snapshot_member_fd(directory_fd, name) for name in OUTPUT_FILENAMES} 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: @@ -330,7 +370,27 @@ 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 member_hashes["manifest.json"], tuple(sorted(members.items())) + + +def verify_tqqq_r1_snapshot( + output_dir: str | Path, + *, + expected_manifest_sha256: str, +) -> SnapshotResult: + """Verify immutable members while preserving the caller-provided output path.""" + output = Path(output_dir) + try: + directory_fd = os.open(os.fspath(output), _snapshot_open_flags(directory=True)) + except OSError as exc: + raise SnapshotValidationError("unable to read snapshot members") from exc + try: + manifest_sha256, _ = _verify_tqqq_r1_snapshot_fd( + directory_fd, expected_manifest_sha256=expected_manifest_sha256 + ) + finally: + os.close(directory_fd) + return SnapshotResult(output_dir=output, manifest_sha256=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..d0d46ad --- /dev/null +++ b/src/us_equity_snapshot_pipelines/tqqq_trusted_snapshot_package.py @@ -0,0 +1,342 @@ +"""Local descriptor-backed verification for immutable TQQQ snapshot evidence.""" + +from __future__ import annotations + +import csv +import errno +import hashlib +import json +import math +import os +import stat +from dataclasses import dataclass, field +from datetime import date +from io import StringIO +from pathlib import Path +from threading import Lock +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 +MAX_EVIDENCE_BYTES = 1024 * 1024 +_MAX_JSON_INTEGER_DIGITS = 309 +_VERIFIED_CONSTRUCTION = object() + + +class TrustedSnapshotPackageError(ValueError): + """Raised when untrusted local package evidence cannot be verified.""" + + +@dataclass(frozen=True, init=False) +class TrustedSnapshotPackage: + """Verified immutable bytes plus one owned directory descriptor.""" + + 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) + _lifecycle_lock: Lock = 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) + object.__setattr__(self, "_lifecycle_lock", Lock()) + + def close(self) -> None: + """Atomically claim and release the owned descriptor exactly once.""" + with self._lifecycle_lock: + descriptor = self._snapshot_fd + if descriptor < 0: + return + object.__setattr__(self, "_snapshot_fd", -1) + try: + os.close(descriptor) + except OSError: + pass + + def read_snapshot_member(self, name: str) -> bytes: + """Return immutable verified bytes; never reopen a mutable member path.""" + with self._lifecycle_lock: + 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: + if os.name != "posix": + _invalid("descriptor-safe trusted-input capability is unavailable") + 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") + access = os.O_WRONLY | os.O_CREAT | os.O_TRUNC if write else os.O_RDONLY + return access | no_follow | non_blocking | directory_flag + + +def _read_regular(path: str | Path, name: str) -> bytes: + """Read exactly one bounded, non-symlink regular file through its descriptor.""" + 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 bounded finite duplicate-free JSON and normalize parser failures.""" + 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, TypeError, OverflowError, RecursionError) as exc: + raise TrustedSnapshotPackageError(f"invalid strict JSON: {name}") from exc + + +def write_strict_json(path: str | Path, payload: object) -> None: + """Write canonical finite JSON without following a destination symlink.""" + try: + encoded = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False).encode("utf-8") + except (TypeError, ValueError, OverflowError, RecursionError) 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 _open_snapshot_directory(snapshot_dir: str | Path) -> int: + try: + return 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 + + +def _is_sha256(value: object) -> bool: + return type(value) is str and len(value) == _HEX_SHA256_LENGTH and all(char in "0123456789abcdef" for char 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, keys: set[str], name: str) -> dict[str, Any]: + if type(value) is not dict or set(value) != 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 or 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 or 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: + """Authenticate anchors before parsing and retain only descriptor-stable bytes.""" + 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 = _open_snapshot_directory(snapshot_dir) + try: + try: + _, snapshot_members = tqqq_r1_snapshot._verify_tqqq_r1_snapshot_fd( + snapshot_fd, expected_manifest_sha256=expected_snapshot_manifest_sha256 + ) + except Exception 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")) + if manifest["calendar_sha256"] != calendar_sha256 or receipt["calendar_sha256"] != calendar_sha256: + _invalid("calendar hash binding mismatch") + 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["session"] != receipt["session"] or manifest["session"] not in calendar["sessions"]: + _invalid("calendar session binding mismatch") + try: + members = dict(snapshot_members) + snapshot_sessions = {row["session"] for row in csv.DictReader(StringIO(members["prices.csv"].decode("utf-8")))} + except (KeyError, UnicodeDecodeError, csv.Error, RecursionError) 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=Path(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=snapshot_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..f376d38 100644 --- a/tests/test_tqqq_r1_snapshot.py +++ b/tests/test_tqqq_r1_snapshot.py @@ -2,8 +2,10 @@ import hashlib import json +import os from datetime import datetime, timezone from pathlib import Path +from unittest.mock import patch import numpy as np import pandas as pd @@ -264,3 +266,41 @@ 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_preserves_caller_path_without_reopening_members(tmp_path: Path) -> None: + result = snapshot.materialize_tqqq_r1_snapshot(_fixture_prices(), tmp_path / "snapshot") + original_open = snapshot.os.open + member_reopens: list[str] = [] + + def observe_open(path: object, flags: int, *args: object, **kwargs: object) -> int: + if isinstance(path, (str, bytes, Path)) and Path(path).parent == result.output_dir: + member_reopens.append(str(path)) + return original_open(path, flags, *args, **kwargs) + + with patch.object(snapshot.os, "open", observe_open): + verified = snapshot.verify_tqqq_r1_snapshot( + result.output_dir, expected_manifest_sha256=result.manifest_sha256 + ) + + assert verified.output_dir == result.output_dir + assert member_reopens == [] + + +def test_snapshot_member_size_limit_rejects_before_streaming_or_parsing( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + oversized = tmp_path / "prices.csv" + oversized.touch() + os.truncate(oversized, snapshot.MAX_SNAPSHOT_MEMBER_BYTES + 1) + directory_fd = os.open(tmp_path, snapshot._snapshot_open_flags(directory=True)) + + def fail_if_streamed(*_args: object, **_kwargs: object) -> bytes: + raise AssertionError("oversized snapshot member must not be streamed") + + monkeypatch.setattr(snapshot.os, "read", fail_if_streamed) + try: + with pytest.raises(snapshot.SnapshotValidationError, match="snapshot member exceeds size limit"): + snapshot._read_snapshot_member_fd(directory_fd, "prices.csv") + finally: + os.close(directory_fd) diff --git a/tests/test_tqqq_trusted_snapshot_package.py b/tests/test_tqqq_trusted_snapshot_package.py new file mode 100644 index 0000000..2b3f1d0 --- /dev/null +++ b/tests/test_tqqq_trusted_snapshot_package.py @@ -0,0 +1,225 @@ +from __future__ import annotations + +import copy +import hashlib +import os +import pickle +import threading +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(*args: object, package_manifest_sha256: str | None = None, receipt_sha256: str | None = None) -> package.TrustedSnapshotPackage: + snapshot_dir, manifest_path, receipt_path, calendar_path, snapshot_manifest_sha256 = args + 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'{"x":NaN}', b'{"x":1e400}', b'{"x":' + b"7" * 5000 + b"}"]) +def test_strict_json_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_authenticated_deep_json_recursion_error_is_normalized(monkeypatch: pytest.MonkeyPatch) -> None: + def recurse(*_args: object, **_kwargs: object) -> object: + raise RecursionError("deep JSON") + + monkeypatch.setattr(package.json, "loads", recurse) + with pytest.raises(package.TrustedSnapshotPackageError, match="invalid strict JSON"): + package.read_strict_json(b'{"calendar":[]}', "calendar evidence") + + +def test_calendar_digest_mismatch_never_calls_calendar_parser(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + args = _write_bound_package(tmp_path) + args[3].write_bytes(b'{"deep":' + b"[" * 1200 + b"0" + b"]" * 1200 + b"}") + parse = package.read_strict_json + parsed_names: list[str] = [] + + def observe(payload: bytes, name: str) -> object: + parsed_names.append(name) + return parse(payload, name) + + monkeypatch.setattr(package, "read_strict_json", observe) + with pytest.raises(package.TrustedSnapshotPackageError, match="calendar hash binding mismatch"): + _load(*args) + assert "calendar evidence" not in parsed_names + + +def test_package_keeps_caller_path_and_never_exposes_linux_descriptor_path(tmp_path: Path) -> None: + args = _write_bound_package(tmp_path) + trusted = _load(*args) + assert trusted.snapshot_dir == args[0] + assert not str(trusted.snapshot_dir).startswith("/proc/self/fd/") + trusted.close() + + +def test_package_uses_verified_bytes_after_member_replacement(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + args = _write_bound_package(tmp_path) + verify = snapshot._verify_tqqq_r1_snapshot_fd + + def replace_after_verify(directory_fd: int, *, expected_manifest_sha256: str) -> tuple[str, tuple[tuple[str, bytes], ...]]: + result = verify(directory_fd, expected_manifest_sha256=expected_manifest_sha256) + (args[0] / "prices.csv").write_text("replaced\n", encoding="utf-8") + return result + + monkeypatch.setattr(package.tqqq_r1_snapshot, "_verify_tqqq_r1_snapshot_fd", replace_after_verify) + trusted = _load(*args) + assert b"45.25" in trusted.read_snapshot_member("prices.csv") + trusted.close() + + +def test_concurrent_close_claims_descriptor_once(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + trusted = _load(*_write_bound_package(tmp_path)) + descriptor = trusted._snapshot_fd + close = package.os.close + calls: list[int] = [] + + def observe(value: int) -> None: + if value == descriptor: + calls.append(value) + close(value) + + monkeypatch.setattr(package.os, "close", observe) + threads = [threading.Thread(target=trusted.close) for _ in range(12)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + assert calls == [descriptor] + + +def test_close_read_race_is_verified_bytes_or_closed_error(tmp_path: Path) -> None: + trusted = _load(*_write_bound_package(tmp_path)) + outcome: list[bytes | Exception] = [] + start = threading.Barrier(2) + + def read() -> None: + start.wait() + try: + outcome.append(trusted.read_snapshot_member("prices.csv")) + except package.TrustedSnapshotPackageError as exc: + outcome.append(exc) + + reader = threading.Thread(target=read) + reader.start() + start.wait() + trusted.close() + reader.join() + assert len(outcome) == 1 + assert isinstance(outcome[0], (bytes, package.TrustedSnapshotPackageError)) + + +def test_package_close_is_idempotent_and_lifecycle_is_not_identity(tmp_path: Path) -> None: + trusted = _load(*_write_bound_package(tmp_path)) + assert "prices.csv" not in repr(trusted) + assert b"45.25" not in repr(trusted).encode() + 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) + before_close = 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_rejects_symlink_fifo_and_oversized_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) + + args = _write_bound_package(tmp_path / "fifo") + args[3].unlink() + os.mkfifo(args[3]) + with pytest.raises(package.TrustedSnapshotPackageError, match="regular non-symlink"): + _load(*args) + + args = _write_bound_package(tmp_path / "huge") + args[3].write_bytes(b"x" * (package.MAX_EVIDENCE_BYTES + 1)) + with pytest.raises(package.TrustedSnapshotPackageError, match="exceeds size limit"): + _load(*args) + + +def test_reader_rejects_device_and_toctou_size_change(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + with pytest.raises(package.TrustedSnapshotPackageError, match="regular non-symlink"): + package._read_regular("/dev/null", "device evidence") + + evidence = tmp_path / "evidence.json" + evidence.write_bytes(b"{}") + fstat = package.os.fstat + calls = 0 + + def changed_size(descriptor: int) -> os.stat_result: + nonlocal calls + calls += 1 + actual = fstat(descriptor) + if calls == 2: + values = list(actual) + values[6] += 1 + return os.stat_result(values) + return actual + + monkeypatch.setattr(package.os, "fstat", changed_size) + with pytest.raises(package.TrustedSnapshotPackageError, match="changed while reading"): + package._read_regular(evidence, "evidence") + + +def test_unsupported_platform_capability_fails_closed(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(package.os, "name", "nt") + with pytest.raises(package.TrustedSnapshotPackageError, match="capability"): + package._safe_open_flags()