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..8dabd68 --- /dev/null +++ b/src/us_equity_snapshot_pipelines/tqqq_trusted_snapshot_package.py @@ -0,0 +1,654 @@ +"""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 BytesIO, StringIO +from pathlib import Path +from threading import Lock +from typing import Any + +import pandas as pd + +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_SNAPSHOT_MEMBER_BYTES = 16 * 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 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 failures.""" + if type(payload) is not bytes or type(name) is not str: + _invalid("invalid strict JSON") + if len(payload) > MAX_EVIDENCE_BYTES: + _invalid(f"{name} exceeds size limit") + 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 _lexically_normalized_snapshot_dir( + snapshot_dir: str | Path, +) -> str: + try: + raw_path = os.fspath(snapshot_dir) + except TypeError as exc: + raise TrustedSnapshotPackageError( + "invalid snapshot directory", + ) from exc + if type(raw_path) is not str or not raw_path: + _invalid("invalid snapshot directory") + return os.path.normpath(raw_path) + + +def _open_snapshot_directory(snapshot_dir: str | Path) -> int: + normalized = _lexically_normalized_snapshot_dir(snapshot_dir) + try: + return os.open( + normalized, + _safe_open_flags(directory=True), + ) + except OSError as exc: + raise TrustedSnapshotPackageError( + "unable to open stable snapshot directory", + ) from exc + + +def _read_snapshot_member(directory_fd: int, name: str) -> bytes: + try: + descriptor = os.open( + name, + _safe_open_flags(), + dir_fd=directory_fd, + ) + except OSError as exc: + raise TrustedSnapshotPackageError( + "unable to read snapshot members", + ) from exc + try: + initial = os.fstat(descriptor) + 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( + 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 != initial.st_size or received != initial.st_size: + _invalid("snapshot member changed while reading") + return b"".join(chunks) + except OSError as exc: + raise TrustedSnapshotPackageError( + "unable to read snapshot members", + ) from exc + finally: + os.close(descriptor) + + +def _verify_snapshot_fd( + directory_fd: int, + *, + expected_manifest_sha256: str, +) -> tuple[str, tuple[tuple[str, bytes], ...]]: + try: + names = tuple(sorted(os.listdir(directory_fd))) + except OSError as exc: + raise TrustedSnapshotPackageError( + "unable to read snapshot members", + ) from exc + if names != tuple(sorted(tqqq_r1_snapshot.OUTPUT_FILENAMES)): + _invalid(f"unexpected output files: {names}") + members = {name: _read_snapshot_member(directory_fd, name) for name in tqqq_r1_snapshot.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: + _invalid("trusted manifest hash mismatch") + + try: + sums = tqqq_r1_snapshot._parse_json_object( + members["sha256sums.json"], + "sha256sums", + ) + manifest = tqqq_r1_snapshot._parse_json_object( + members["manifest.json"], + "manifest", + ) + validation = tqqq_r1_snapshot._parse_json_object( + members["validation.json"], + "validation", + ) + if set(sums) != { + "prices.csv", + "manifest.json", + "validation.json", + }: + _invalid("invalid sha256sums") + for name, expected in sums.items(): + if type(expected) is not str or member_hashes[name] != expected: + _invalid(f"hash mismatch: {name}") + prices = tqqq_r1_snapshot._normalized_prices( + pd.read_csv( + BytesIO(members["prices.csv"]), + dtype={tqqq_r1_snapshot.PRICE_FIELD: "string"}, + ), + require_exact_columns=True, + require_canonical_symbols=True, + require_canonical_sessions=True, + ) + except TrustedSnapshotPackageError: + raise + except Exception as exc: + raise TrustedSnapshotPackageError( + "invalid verified snapshot", + ) from exc + + expected_manifest = { + "contract_version": tqqq_r1_snapshot.CONTRACT_VERSION, + "symbols": list(tqqq_r1_snapshot.SYMBOLS), + "requested_lower_bound": (tqqq_r1_snapshot.REQUESTED_LOWER_BOUND), + "price_field": tqqq_r1_snapshot.PRICE_FIELD, + "plugin": tqqq_r1_snapshot.PLUGIN, + "mode": tqqq_r1_snapshot.MODE, + "size": 0, + "row_count": len(prices), + "prices_sha256": member_hashes["prices.csv"], + } + if not tqqq_r1_snapshot._has_exact_type( + manifest, + expected_manifest, + ): + _invalid("invalid manifest") + expected_validation = { + "valid": True, + "row_count": len(prices), + "symbols": list(tqqq_r1_snapshot.SYMBOLS), + } + if not tqqq_r1_snapshot._has_exact_type( + validation, + expected_validation, + ): + _invalid("invalid validation") + return ( + member_hashes["manifest.json"], + tuple(sorted(members.items())), + ) + + +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 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 = _verify_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_trusted_snapshot_package.py b/tests/test_tqqq_trusted_snapshot_package.py new file mode 100644 index 0000000..a4aa9b5 --- /dev/null +++ b/tests/test_tqqq_trusted_snapshot_package.py @@ -0,0 +1,451 @@ +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"}", + b'{"x":1,"x":2}', + b"\xff", + ], +) +def test_strict_json_normalizes_adversarial_inputs(payload: bytes) -> None: + with pytest.raises( + package.TrustedSnapshotPackageError, + match="invalid strict JSON", + ): + package.read_strict_json(payload, "payload") + + +def test_strict_json_rejects_oversized_bytes_before_parser( + monkeypatch: pytest.MonkeyPatch, +) -> None: + parser_called = False + + def observe(*_args: object, **_kwargs: object) -> object: + nonlocal parser_called + parser_called = True + return {} + + monkeypatch.setattr(package.json, "loads", observe) + payload = b" " * (package.MAX_EVIDENCE_BYTES + 1) + + with pytest.raises( + package.TrustedSnapshotPackageError, + match="exceeds size limit", + ): + package.read_strict_json(payload, "oversized payload") + + assert parser_called is False + + +def test_strict_json_accepts_exact_size_boundary() -> None: + payload = b" " * (package.MAX_EVIDENCE_BYTES - 2) + b"{}" + + assert len(payload) == package.MAX_EVIDENCE_BYTES + assert package.read_strict_json(payload, "boundary 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_verified_bytes( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + args = _write_bound_package(tmp_path) + verify = package._verify_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, + "_verify_snapshot_fd", + replace_after_verify, + ) + trusted = _load(*args) + + assert trusted.snapshot_dir == args[0] + assert not str(trusted.snapshot_dir).startswith("/proc/self/fd/") + assert b"45.25" in trusted.read_snapshot_member("prices.csv") + trusted.close() + + +def test_loader_accepts_relative_snapshot_root_and_preserves_metadata( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + args = _write_bound_package(tmp_path) + monkeypatch.chdir(tmp_path) + relative_snapshot_dir = Path("snapshot") + + trusted = _load(relative_snapshot_dir, *args[1:]) + + assert trusted.snapshot_dir == relative_snapshot_dir + trusted.close() + + +@pytest.mark.parametrize( + "root_form", + [ + lambda path: path, + lambda path: os.fspath(path) + os.sep, + lambda path: os.fspath(path) + os.sep + ".", + ], + ids=["path", "trailing-slash", "trailing-dot"], +) +def test_loader_rejects_all_final_symlink_snapshot_root_forms( + tmp_path: Path, + root_form: object, +) -> None: + args = _write_bound_package(tmp_path) + link = tmp_path / "snapshot-link" + link.symlink_to(args[0], target_is_directory=True) + + with pytest.raises( + package.TrustedSnapshotPackageError, + match="stable snapshot directory", + ): + _load(root_form(link), *args[1:]) + + +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_not_transferable( + 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) + + trusted.close() + trusted.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_snapshot_member_size_limit_rejects_before_read( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + oversized = tmp_path / "prices.csv" + oversized.touch() + os.truncate( + oversized, + package._MAX_SNAPSHOT_MEMBER_BYTES + 1, + ) + directory_fd = package._open_snapshot_directory(tmp_path) + + def fail_if_read(*_args: object, **_kwargs: object) -> bytes: + raise AssertionError( + "oversized snapshot member must not be read", + ) + + monkeypatch.setattr(package.os, "read", fail_if_read) + try: + with pytest.raises( + package.TrustedSnapshotPackageError, + match="snapshot member exceeds size limit", + ): + package._read_snapshot_member( + directory_fd, + "prices.csv", + ) + finally: + os.close(directory_fd) + + +def test_loader_rejects_snapshot_member_symlink(tmp_path: Path) -> None: + args = _write_bound_package(tmp_path) + target = tmp_path / "prices-target.csv" + prices_path = args[0] / "prices.csv" + target.write_bytes(prices_path.read_bytes()) + prices_path.unlink() + prices_path.symlink_to(target) + + with pytest.raises( + package.TrustedSnapshotPackageError, + match="invalid verified snapshot", + ): + _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()