diff --git a/schemas/soxl_tqqq_clean_cutover_snapshot.v1.schema.json b/schemas/soxl_tqqq_clean_cutover_snapshot.v1.schema.json new file mode 100644 index 0000000..498ceb0 --- /dev/null +++ b/schemas/soxl_tqqq_clean_cutover_snapshot.v1.schema.json @@ -0,0 +1 @@ +{"$schema":"https://json-schema.org/draft/2020-12/schema","additionalProperties":false,"properties":{"calendar_timezone":{"const":"America/New_York","type":"string"},"generation":{"const":"clean_cutover_v1","type":"string"},"offline_fixture":{"const":true,"type":"boolean"},"pair_id":{"const":"QQQ_TQQQ","type":"string"},"plugin":{"const":"ABSENT_DISABLED","type":"string"},"rows":{"items":{"additionalProperties":false,"properties":{"adjusted_close":{"maxLength":32,"pattern":"^(?:0\\.[0-9]*[1-9]|[1-9][0-9]*(?:\\.[0-9]*[1-9])?)$","type":"string"},"session":{"format":"date","type":"string"},"symbol":{"enum":["QQQ","TQQQ"],"type":"string"}},"required":["adjusted_close","session","symbol"],"type":"object"},"minItems":2,"type":"array"},"schema_version":{"const":"soxl_tqqq_clean_cutover_snapshot.v1","type":"string"},"sessions":{"items":{"format":"date","type":"string"},"minItems":1,"type":"array","uniqueItems":true},"size":{"const":0,"type":"integer"},"symbols":{"items":false,"minItems":2,"prefixItems":[{"const":"QQQ"},{"const":"TQQQ"}],"type":"array"}},"required":["calendar_timezone","generation","offline_fixture","pair_id","plugin","rows","schema_version","sessions","size","symbols"],"type":"object"} diff --git a/src/us_equity_snapshot_pipelines/soxl_tqqq_clean_cutover_snapshot.py b/src/us_equity_snapshot_pipelines/soxl_tqqq_clean_cutover_snapshot.py new file mode 100644 index 0000000..987caef --- /dev/null +++ b/src/us_equity_snapshot_pipelines/soxl_tqqq_clean_cutover_snapshot.py @@ -0,0 +1,391 @@ +"""Offline-only trusted QQQ/TQQQ clean-cutover snapshot package boundary.""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import date +from decimal import Decimal, InvalidOperation +import hashlib +import json +import os +from pathlib import Path +import re +import stat + + +SCHEMA_VERSION = "soxl_tqqq_clean_cutover_snapshot.v1" +GENERATION = "clean_cutover_v1" +PLUGIN = "ABSENT_DISABLED" +PAIR_ID = "QQQ_TQQQ" +SYMBOLS = ("QQQ", "TQQQ") +CALENDAR_TIMEZONE = "America/New_York" +TOP_LEVEL_FIELDS = frozenset( + { + "calendar_timezone", + "generation", + "offline_fixture", + "pair_id", + "plugin", + "rows", + "schema_version", + "sessions", + "size", + "symbols", + } +) +ROW_FIELDS = frozenset({"adjusted_close", "session", "symbol"}) +CANONICAL_DECIMAL_PATTERN = r"(?:0\.[0-9]*[1-9]|[1-9][0-9]*(?:\.[0-9]*[1-9])?)" +_DECIMAL_RE = re.compile(rf"^{CANONICAL_DECIMAL_PATTERN}$") +MAX_DECIMAL_LENGTH = 32 +MAX_JSON_INT_DIGITS = 64 +MAX_READ_BYTES = 16 * 1_024 * 1_024 +_READ_CHUNK_BYTES = 65_536 +_SHA256_RE = re.compile(r"^[0-9a-f]{64}$") +_PACKAGE_TOKEN = object() + + +class SnapshotValidationError(ValueError): + """The immutable offline snapshot did not satisfy its trust contract.""" + + +def _invalid(message: str) -> None: + raise SnapshotValidationError(message) + + +def canonical_json_bytes(value: object) -> bytes: + """Serialize JSON with the exact encoding shared by runtime and schema tests.""" + try: + encoded = json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True).encode("utf-8") + except (TypeError, ValueError) as exc: + raise SnapshotValidationError("cannot canonicalize JSON") from exc + if len(encoded) > MAX_READ_BYTES: + _invalid("canonical JSON exceeds maximum read size") + return encoded + + +def _validate_sha256(value: object, field: str) -> None: + if type(value) is not str or _SHA256_RE.fullmatch(value) is None: + _invalid(f"invalid {field}") + + +def _is_canonical_session(value: object) -> bool: + if type(value) is not str: + return False + try: + return date.fromisoformat(value).isoformat() == value + except ValueError: + return False + + +def _validate_expected_sessions(value: object) -> None: + if type(value) is not tuple or not value: + _invalid("expected_sessions must be a nonempty tuple") + if any(not _is_canonical_session(session) for session in value): + _invalid("expected_sessions contains an invalid session") + if tuple(sorted(value)) != value or len(set(value)) != len(value): + _invalid("expected_sessions must be sorted and unique") + + +def _validate_external_bindings(bindings: "ExternalBindings") -> None: + _validate_sha256(bindings.source_sha256, "source_sha256") + _validate_sha256(bindings.calendar_sha256, "calendar_sha256") + _validate_sha256(bindings.manifest_sha256, "manifest_sha256") + _validate_sha256(bindings.content_sha256, "content_sha256") + _validate_expected_sessions(bindings.expected_sessions) + + +@dataclass(frozen=True, slots=True) +class ExternalBindings: + """Typed control-plane receipt binding for one complete offline fixture file.""" + + source_sha256: str + calendar_sha256: str + manifest_sha256: str + content_sha256: str + expected_sessions: tuple[str, ...] + + def __post_init__(self) -> None: + _validate_external_bindings(self) + + +@dataclass(frozen=True, slots=True, init=False) +class TrustedSnapshotPackage: + """Immutable package that only this module's strict factory can create.""" + + canonical_bytes: bytes + external_bindings: ExternalBindings + snapshot_id: str + + def __init__(self, *, _token: object, canonical_bytes: bytes, external_bindings: ExternalBindings) -> None: + if _token is not _PACKAGE_TOKEN: + raise TypeError("TrustedSnapshotPackage must be created by build_trusted_snapshot_package") + object.__setattr__(self, "canonical_bytes", canonical_bytes) + object.__setattr__(self, "external_bindings", external_bindings) + object.__setattr__(self, "snapshot_id", f"sha256-{external_bindings.manifest_sha256}") + + +def public_json_schema() -> dict[str, object]: + """Generate the checked-in public schema from the runtime constants.""" + decimal_schema = {"maxLength": MAX_DECIMAL_LENGTH, "pattern": rf"^{CANONICAL_DECIMAL_PATTERN}$", "type": "string"} + row_schema = { + "additionalProperties": False, + "properties": { + "adjusted_close": decimal_schema, + "session": {"format": "date", "type": "string"}, + "symbol": {"enum": list(SYMBOLS), "type": "string"}, + }, + "required": sorted(ROW_FIELDS), + "type": "object", + } + return { + "$schema": "https://json-schema.org/draft/2020-12/schema", + "additionalProperties": False, + "properties": { + "calendar_timezone": {"const": CALENDAR_TIMEZONE, "type": "string"}, + "generation": {"const": GENERATION, "type": "string"}, + "offline_fixture": {"const": True, "type": "boolean"}, + "pair_id": {"const": PAIR_ID, "type": "string"}, + "plugin": {"const": PLUGIN, "type": "string"}, + "rows": {"items": row_schema, "minItems": 2, "type": "array"}, + "schema_version": {"const": SCHEMA_VERSION, "type": "string"}, + "sessions": { + "items": {"format": "date", "type": "string"}, + "minItems": 1, + "type": "array", + "uniqueItems": True, + }, + "size": {"const": 0, "type": "integer"}, + "symbols": { + "items": False, + "minItems": 2, + "prefixItems": [{"const": SYMBOLS[0]}, {"const": SYMBOLS[1]}], + "type": "array", + }, + }, + "required": sorted(TOP_LEVEL_FIELDS), + "type": "object", + } + + +def _required_open_flags(*, directory: bool = False, nonblocking: bool = False) -> int: + try: + flags = os.O_RDONLY | os.O_NOFOLLOW + if directory: + flags |= os.O_DIRECTORY + if nonblocking: + flags |= os.O_NONBLOCK + except AttributeError as exc: + raise SnapshotValidationError("descriptor-safe open capability is unavailable") from exc + if type(flags) is not int: + _invalid("descriptor-safe open capability is unavailable") + return flags + + +def _relative_components(relative_path: object) -> tuple[str, ...]: + if not isinstance(relative_path, (str, Path)): + _invalid("relative_path must be a path string") + candidate = Path(relative_path) + components = candidate.parts + if candidate.is_absolute() or not components or any(component in {"", ".", ".."} for component in components): + _invalid("relative_path must stay below trusted_root") + return components + + +def _open_trusted_root(trusted_root: object) -> int: + if not isinstance(trusted_root, (str, Path)): + _invalid("trusted_root must be a path string") + root_path = Path(trusted_root) + if root_path.is_absolute(): + anchor = root_path.anchor + components = root_path.parts[1:] + else: + anchor = "." + components = root_path.parts + if any(component == ".." for component in components): + _invalid("trusted_root must not contain parent traversal") + descriptor = os.open(anchor, _required_open_flags(directory=True)) + try: + for component in components: + child_descriptor = os.open(component, _required_open_flags(directory=True), dir_fd=descriptor) + if not stat.S_ISDIR(os.fstat(child_descriptor).st_mode): + os.close(child_descriptor) + _invalid("trusted_root is not a directory") + os.close(descriptor) + descriptor = child_descriptor + return descriptor + except BaseException: + try: + os.close(descriptor) + except OSError: + pass + raise + + +def _read_trusted_file(*, trusted_root: object, relative_path: object) -> bytes: + components = _relative_components(relative_path) + root_fd = -1 + parent_fd = -1 + file_fd = -1 + try: + root_fd = _open_trusted_root(trusted_root) + parent_fd = root_fd + root_fd = -1 + for component in components[:-1]: + child_fd = os.open(component, _required_open_flags(directory=True), dir_fd=parent_fd) + if not stat.S_ISDIR(os.fstat(child_fd).st_mode): + os.close(child_fd) + _invalid("candidate ancestor is not a directory") + os.close(parent_fd) + parent_fd = child_fd + file_fd = os.open(components[-1], _required_open_flags(nonblocking=True), dir_fd=parent_fd) + file_stat = os.fstat(file_fd) + if not stat.S_ISREG(file_stat.st_mode): + _invalid("candidate is not a regular file") + if file_stat.st_size > MAX_READ_BYTES: + _invalid("candidate exceeds maximum read size") + data = bytearray() + while True: + chunk = os.read(file_fd, _READ_CHUNK_BYTES) + if not chunk: + break + data.extend(chunk) + if len(data) > MAX_READ_BYTES: + _invalid("candidate exceeds maximum read size") + if len(data) != file_stat.st_size: + _invalid("candidate changed during readback") + return bytes(data) + except SnapshotValidationError: + raise + except (OSError, TypeError, ValueError) as exc: + raise SnapshotValidationError("strict readback failed") from exc + finally: + for descriptor in (file_fd, parent_fd, root_fd): + if descriptor >= 0: + try: + os.close(descriptor) + except OSError: + pass + + +def _reject_duplicate_keys(pairs: list[tuple[object, object]]) -> dict[str, object]: + result: dict[str, object] = {} + for key, value in pairs: + if type(key) is not str or key in result: + _invalid("duplicate or invalid JSON key") + result[key] = value + return result + + +def _parse_int(value: str) -> int: + if len(value.lstrip("-")) > MAX_JSON_INT_DIGITS: + _invalid("JSON integer is too large") + try: + return int(value) + except ValueError as exc: + raise SnapshotValidationError("invalid JSON integer") from exc + + +def _reject_json_float(_: str) -> object: + _invalid("JSON floats are not allowed") + + +def _parse_canonical_json(raw: bytes) -> dict[str, object]: + try: + parsed = json.loads( + raw.decode("utf-8"), + object_pairs_hook=_reject_duplicate_keys, + parse_constant=_reject_json_float, + parse_float=_reject_json_float, + parse_int=_parse_int, + ) + except SnapshotValidationError: + raise + except (UnicodeDecodeError, json.JSONDecodeError, ValueError, TypeError) as exc: + raise SnapshotValidationError("invalid UTF-8 JSON") from exc + if type(parsed) is not dict: + _invalid("snapshot payload must be a JSON object") + if canonical_json_bytes(parsed) != raw: + _invalid("snapshot payload is not canonical JSON") + return parsed + + +def _validate_decimal(value: object) -> None: + if type(value) is not str or len(value) > MAX_DECIMAL_LENGTH or _DECIMAL_RE.fullmatch(value) is None: + _invalid("invalid canonical adjusted_close") + try: + decimal_value = Decimal(value) + except InvalidOperation as exc: + raise SnapshotValidationError("invalid canonical adjusted_close") from exc + if not decimal_value.is_finite() or decimal_value <= 0: + _invalid("adjusted_close is outside the permitted finite range") + + +def _validate_payload(payload: dict[str, object], bindings: ExternalBindings) -> None: + if set(payload) != TOP_LEVEL_FIELDS: + _invalid("snapshot payload fields are invalid") + constants = { + "calendar_timezone": CALENDAR_TIMEZONE, + "generation": GENERATION, + "offline_fixture": True, + "pair_id": PAIR_ID, + "plugin": PLUGIN, + "schema_version": SCHEMA_VERSION, + "size": 0, + "symbols": list(SYMBOLS), + } + for field, expected in constants.items(): + if payload.get(field) != expected or type(payload.get(field)) is not type(expected): + _invalid(f"snapshot {field} is invalid") + sessions = payload.get("sessions") + if type(sessions) is not list or any(not _is_canonical_session(session) for session in sessions): + _invalid("snapshot sessions are invalid") + if tuple(sessions) != bindings.expected_sessions: + _invalid("snapshot sessions do not match external calendar coverage") + rows = payload.get("rows") + if type(rows) is not list or len(rows) != 2 * len(bindings.expected_sessions): + _invalid("snapshot rows are incomplete") + expected_keys = [(session, symbol) for session in bindings.expected_sessions for symbol in SYMBOLS] + observed_keys: list[tuple[str, str]] = [] + for row in rows: + if type(row) is not dict or set(row) != ROW_FIELDS: + _invalid("snapshot row fields are invalid") + session = row.get("session") + symbol = row.get("symbol") + if not _is_canonical_session(session) or type(symbol) is not str or symbol not in SYMBOLS: + _invalid("snapshot row identity is invalid") + _validate_decimal(row.get("adjusted_close")) + observed_keys.append((session, symbol)) + if observed_keys != expected_keys: + _invalid("snapshot rows must be sorted, unique, and pair-complete") + + +def build_trusted_snapshot_package( + *, trusted_root: Path | str, relative_path: Path | str, bindings: ExternalBindings +) -> TrustedSnapshotPackage: + """Read, externally bind, and validate one offline fixture into the only public package type.""" + if type(bindings) is not ExternalBindings: + _invalid("bindings must be an ExternalBindings value") + _validate_external_bindings(bindings) + raw = _read_trusted_file(trusted_root=trusted_root, relative_path=relative_path) + if hashlib.sha256(raw).hexdigest() != bindings.content_sha256: + _invalid("external content digest does not match before JSON parsing") + payload = _parse_canonical_json(raw) + _validate_payload(payload, bindings) + return TrustedSnapshotPackage(_token=_PACKAGE_TOKEN, canonical_bytes=raw, external_bindings=bindings) + + +def validate_trusted_snapshot_package(package: object) -> TrustedSnapshotPackage: + """Reject every consumer input except a factory-created immutable package.""" + if type(package) is not TrustedSnapshotPackage: + _invalid("consumer input must be a TrustedSnapshotPackage") + if type(package.canonical_bytes) is not bytes or type(package.external_bindings) is not ExternalBindings: + _invalid("trusted package has invalid fields") + _validate_external_bindings(package.external_bindings) + if hashlib.sha256(package.canonical_bytes).hexdigest() != package.external_bindings.content_sha256: + _invalid("trusted package content digest is invalid") + expected_id = f"sha256-{package.external_bindings.manifest_sha256}" + if package.snapshot_id != expected_id: + _invalid("trusted package snapshot_id is invalid") + payload = _parse_canonical_json(package.canonical_bytes) + _validate_payload(payload, package.external_bindings) + return package diff --git a/tests/test_soxl_tqqq_clean_cutover_snapshot.py b/tests/test_soxl_tqqq_clean_cutover_snapshot.py new file mode 100644 index 0000000..2ac51dd --- /dev/null +++ b/tests/test_soxl_tqqq_clean_cutover_snapshot.py @@ -0,0 +1,300 @@ +from __future__ import annotations + +import hashlib +import json +import os +from pathlib import Path + +import pytest + +from us_equity_snapshot_pipelines import soxl_tqqq_clean_cutover_snapshot as snapshot + + +HASH_A = "a" * 64 +HASH_B = "b" * 64 +HASH_C = "c" * 64 +HASH_D = "d" * 64 +SESSIONS = ("2024-01-02", "2024-01-03") + + +def _bindings() -> snapshot.ExternalBindings: + return snapshot.ExternalBindings( + source_sha256=HASH_A, + calendar_sha256=HASH_B, + manifest_sha256=HASH_C, + content_sha256=HASH_D, + expected_sessions=SESSIONS, + ) + + +def _payload(*, sessions: tuple[str, ...] = SESSIONS) -> dict[str, object]: + rows = [ + {"session": session, "symbol": symbol, "adjusted_close": price} + for session, price_qqq, price_tqqq in (("2024-01-02", "1.25", "2.5"), ("2024-01-03", "1.5", "3.25")) + if session in sessions + for symbol, price in (("QQQ", price_qqq), ("TQQQ", price_tqqq)) + ] + return { + "calendar_timezone": "America/New_York", + "generation": "clean_cutover_v1", + "offline_fixture": True, + "pair_id": "QQQ_TQQQ", + "plugin": "ABSENT_DISABLED", + "rows": rows, + "schema_version": "soxl_tqqq_clean_cutover_snapshot.v1", + "sessions": list(sessions), + "size": 0, + "symbols": ["QQQ", "TQQQ"], + } + + +def _canonical(payload: object) -> bytes: + return json.dumps(payload, ensure_ascii=False, separators=(",", ":"), sort_keys=True).encode("utf-8") + + +def _write_package(tmp_path: Path, payload: object | None = None) -> tuple[Path, snapshot.ExternalBindings]: + path = tmp_path / "fixture.json" + raw = _canonical(_payload() if payload is None else payload) + path.write_bytes(raw) + bindings = snapshot.ExternalBindings( + source_sha256=HASH_A, + calendar_sha256=HASH_B, + manifest_sha256=HASH_C, + content_sha256=hashlib.sha256(raw).hexdigest(), + expected_sessions=SESSIONS, + ) + return path, bindings + + +def _build(tmp_path: Path, payload: object | None = None) -> snapshot.TrustedSnapshotPackage: + path, bindings = _write_package(tmp_path, payload) + return snapshot.build_trusted_snapshot_package( + trusted_root=tmp_path, + relative_path=path.name, + bindings=bindings, + ) + + +def test_schema_artifact_is_generated_from_runtime_contract() -> None: + schema_path = Path(__file__).parents[1] / "schemas" / "soxl_tqqq_clean_cutover_snapshot.v1.schema.json" + assert schema_path.read_bytes() == snapshot.canonical_json_bytes(snapshot.public_json_schema()) + b"\n" + + +def test_factory_returns_frozen_digest_bound_offline_package(tmp_path: Path) -> None: + package = _build(tmp_path) + assert package.snapshot_id == f"sha256-{HASH_C}" + assert package.external_bindings.source_sha256 == HASH_A + assert package.canonical_bytes == _canonical(_payload()) + with pytest.raises((AttributeError, TypeError)): + package.snapshot_id = "changed" # type: ignore[misc] + assert snapshot.validate_trusted_snapshot_package(package) == package + + +@pytest.mark.parametrize("raw", [b"{}", b"[]", b'{"a":1}', [{"session": "2024-01-02"}]] ) +def test_public_boundary_rejects_raw_inputs(raw: bytes) -> None: + with pytest.raises(snapshot.SnapshotValidationError): + snapshot.validate_trusted_snapshot_package(raw) + + +def test_package_cannot_be_directly_constructed_or_given_free_form_digest() -> None: + with pytest.raises(TypeError): + snapshot.TrustedSnapshotPackage() # type: ignore[call-arg] + with pytest.raises(TypeError): + snapshot.TrustedSnapshotPackage( # type: ignore[call-arg] + _token=object(), canonical_bytes=b"", external_bindings=_bindings() + ) + with pytest.raises(TypeError): + snapshot.build_trusted_snapshot_package( # type: ignore[call-arg] + trusted_root=Path("."), relative_path="fixture.json", content_sha256=HASH_A + ) + + +@pytest.mark.parametrize( + "mutate", + [ + lambda value: value.update({"generation": "legacy"}), + lambda value: value.update({"plugin": "PRESENT"}), + lambda value: value.update({"offline_fixture": False}), + lambda value: value.update({"size": 1}), + lambda value: value.update({"symbols": ["TQQQ", "QQQ"]}), + lambda value: value.update({"sessions": ["2024-01-03", "2024-01-02"]}), + lambda value: value.update({"sessions": ["2024-01-02"]}), + ], +) +def test_exact_constants_and_calendar_coverage_are_required(tmp_path: Path, mutate) -> None: + payload = _payload() + mutate(payload) + with pytest.raises(snapshot.SnapshotValidationError): + _build(tmp_path, payload) + + +def test_decimal_contract_accepts_canonical_sub_unit_and_shared_length_boundary(tmp_path: Path) -> None: + payload = _payload() + payload["rows"][0]["adjusted_close"] = "0.5" # type: ignore[index] + assert _build(tmp_path, payload) + payload = _payload() + payload["rows"][0]["adjusted_close"] = "9" * snapshot.MAX_DECIMAL_LENGTH # type: ignore[index] + assert _build(tmp_path, payload) + schema = snapshot.public_json_schema() + decimal_schema = schema["properties"]["rows"]["items"]["properties"]["adjusted_close"] # type: ignore[index] + assert decimal_schema["maxLength"] == snapshot.MAX_DECIMAL_LENGTH # type: ignore[index] + assert decimal_schema["pattern"] == rf"^{snapshot.CANONICAL_DECIMAL_PATTERN}$" # type: ignore[index] + + +@pytest.mark.parametrize("decimal", [".5", "0", "0.0", "1.0", "01", "1e2", "NaN", "Infinity", "9" * 33]) +def test_decimal_contract_rejects_noncanonical_or_unbounded_values(tmp_path: Path, decimal: str) -> None: + payload = _payload() + payload["rows"][0]["adjusted_close"] = decimal # type: ignore[index] + with pytest.raises(snapshot.SnapshotValidationError): + _build(tmp_path, payload) + + +def test_rows_must_be_pair_complete_sorted_and_unique(tmp_path: Path) -> None: + payload = _payload() + payload["rows"].reverse() # type: ignore[index] + with pytest.raises(snapshot.SnapshotValidationError): + _build(tmp_path, payload) + payload = _payload() + payload["rows"].append(payload["rows"][0]) # type: ignore[index] + with pytest.raises(snapshot.SnapshotValidationError): + _build(tmp_path, payload) + payload = _payload() + payload["rows"].pop() # type: ignore[index] + with pytest.raises(snapshot.SnapshotValidationError): + _build(tmp_path, payload) + + +def test_digest_mismatch_and_noncanonical_or_duplicate_json_fail_before_trust(tmp_path: Path) -> None: + path, bindings = _write_package(tmp_path) + bad_bindings = snapshot.ExternalBindings( + source_sha256=HASH_A, + calendar_sha256=HASH_B, + manifest_sha256=HASH_C, + content_sha256=HASH_D, + expected_sessions=SESSIONS, + ) + with pytest.raises(snapshot.SnapshotValidationError): + snapshot.build_trusted_snapshot_package(trusted_root=tmp_path, relative_path=path.name, bindings=bad_bindings) + duplicate = b'{"rows":{"session":"2024-01-02","session":"2024-01-02"}}' + path.write_bytes(duplicate) + duplicate_bindings = snapshot.ExternalBindings( + source_sha256=HASH_A, + calendar_sha256=HASH_B, + manifest_sha256=HASH_C, + content_sha256=hashlib.sha256(duplicate).hexdigest(), + expected_sessions=SESSIONS, + ) + with pytest.raises(snapshot.SnapshotValidationError): + snapshot.build_trusted_snapshot_package( + trusted_root=tmp_path, relative_path=path.name, bindings=duplicate_bindings + ) + + +@pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlink support required") +def test_strict_readback_rejects_root_leaf_and_ancestor_symlinks_and_escapes(tmp_path: Path) -> None: + path, bindings = _write_package(tmp_path) + link_root = tmp_path / "root-link" + link_root.symlink_to(tmp_path, target_is_directory=True) + for root, candidate in ((link_root, path.name), (tmp_path, "missing/../fixture.json"), (tmp_path, "/tmp/fixture.json")): + with pytest.raises(snapshot.SnapshotValidationError): + snapshot.build_trusted_snapshot_package(trusted_root=root, relative_path=candidate, bindings=bindings) + leaf = tmp_path / "leaf-link.json" + leaf.symlink_to(path) + with pytest.raises(snapshot.SnapshotValidationError): + snapshot.build_trusted_snapshot_package(trusted_root=tmp_path, relative_path=leaf.name, bindings=bindings) + nested = tmp_path / "nested" + nested.symlink_to(tmp_path, target_is_directory=True) + with pytest.raises(snapshot.SnapshotValidationError): + snapshot.build_trusted_snapshot_package(trusted_root=tmp_path, relative_path="nested/fixture.json", bindings=bindings) + real_parent = tmp_path / "real-parent" + trusted_root = real_parent / "trusted-root" + trusted_root.mkdir(parents=True) + source_path, root_bindings = _write_package(trusted_root) + parent_link = tmp_path / "parent-link" + parent_link.symlink_to(real_parent, target_is_directory=True) + with pytest.raises(snapshot.SnapshotValidationError): + snapshot.build_trusted_snapshot_package( + trusted_root=parent_link / trusted_root.name, + relative_path=source_path.name, + bindings=root_bindings, + ) + + +def test_strict_readback_rejects_missing_root_root_file_nonregular_and_oversize(monkeypatch, tmp_path: Path) -> None: + path, bindings = _write_package(tmp_path) + for root in (tmp_path / "missing", path): + with pytest.raises(snapshot.SnapshotValidationError): + snapshot.build_trusted_snapshot_package(trusted_root=root, relative_path=path.name, bindings=bindings) + fifo = tmp_path / "input.fifo" + os.mkfifo(fifo) + with pytest.raises(snapshot.SnapshotValidationError): + snapshot.build_trusted_snapshot_package(trusted_root=tmp_path, relative_path=fifo.name, bindings=bindings) + monkeypatch.setattr(snapshot, "MAX_READ_BYTES", 32) + path.write_bytes(b"x" * 33) + with pytest.raises(snapshot.SnapshotValidationError): + snapshot.build_trusted_snapshot_package(trusted_root=tmp_path, relative_path=path.name, bindings=bindings) + + +def test_serializer_and_readback_share_the_same_byte_limit(monkeypatch, tmp_path: Path) -> None: + monkeypatch.setattr(snapshot, "MAX_READ_BYTES", 8) + with pytest.raises(snapshot.SnapshotValidationError): + snapshot.canonical_json_bytes({"oversized": "fixture"}) + path, bindings = _write_package(tmp_path) + monkeypatch.setattr(snapshot, "MAX_READ_BYTES", path.stat().st_size + 1) + monkeypatch.setattr(snapshot.os, "read", lambda *_: b"") + with pytest.raises(snapshot.SnapshotValidationError): + snapshot.build_trusted_snapshot_package(trusted_root=tmp_path, relative_path=path.name, bindings=bindings) + + +def test_missing_o_nofollow_fails_closed(monkeypatch, tmp_path: Path) -> None: + path, bindings = _write_package(tmp_path) + monkeypatch.delattr(snapshot.os, "O_NOFOLLOW", raising=False) + with pytest.raises(snapshot.SnapshotValidationError): + snapshot.build_trusted_snapshot_package(trusted_root=tmp_path, relative_path=path.name, bindings=bindings) + + +def test_rejects_huge_json_integer_before_conversion(tmp_path: Path) -> None: + raw = b'{"size":' + b"9" * (snapshot.MAX_JSON_INT_DIGITS + 1) + b"}" + (tmp_path / "fixture.json").write_bytes(raw) + bindings = snapshot.ExternalBindings( + source_sha256=HASH_A, + calendar_sha256=HASH_B, + manifest_sha256=HASH_C, + content_sha256=hashlib.sha256(raw).hexdigest(), + expected_sessions=SESSIONS, + ) + with pytest.raises(snapshot.SnapshotValidationError): + snapshot.build_trusted_snapshot_package(trusted_root=tmp_path, relative_path="fixture.json", bindings=bindings) + + +def test_factory_revalidates_mutated_external_bindings(tmp_path: Path) -> None: + path, bindings = _write_package(tmp_path) + object.__setattr__(bindings, "source_sha256", "invalid") + with pytest.raises(snapshot.SnapshotValidationError): + snapshot.build_trusted_snapshot_package(trusted_root=tmp_path, relative_path=path.name, bindings=bindings) + path, bindings = _write_package(tmp_path) + object.__setattr__(bindings, "expected_sessions", ("not-a-date",)) + with pytest.raises(snapshot.SnapshotValidationError): + snapshot.build_trusted_snapshot_package(trusted_root=tmp_path, relative_path=path.name, bindings=bindings) + + +def test_validator_rechecks_mutable_escape_hatches_against_external_bindings(tmp_path: Path) -> None: + package = _build(tmp_path) + object.__setattr__(package, "canonical_bytes", package.canonical_bytes + b" ") + with pytest.raises(snapshot.SnapshotValidationError): + snapshot.validate_trusted_snapshot_package(package) + package = _build(tmp_path) + object.__setattr__(package.external_bindings, "content_sha256", HASH_D) + with pytest.raises(snapshot.SnapshotValidationError): + snapshot.validate_trusted_snapshot_package(package) + package = _build(tmp_path) + object.__setattr__(package.external_bindings, "expected_sessions", ("not-a-date",)) + with pytest.raises(snapshot.SnapshotValidationError): + snapshot.validate_trusted_snapshot_package(package) + + +def test_external_bindings_are_typed_and_exact() -> None: + with pytest.raises(snapshot.SnapshotValidationError): + snapshot.ExternalBindings(HASH_A, HASH_B, HASH_C, HASH_D, ["2024-01-02"]) # type: ignore[arg-type] + with pytest.raises(snapshot.SnapshotValidationError): + snapshot.ExternalBindings(HASH_A, HASH_B, HASH_C, HASH_D, ("2024-01-02", "2024-01-02"))