Add trusted snapshot package boundary

schemas/soxl_tqqq_clean_cutover_snapshot.v1.schema.json

@@ -0,0 +1,23 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "title": "TrustedSnapshotPackage clean-cutover v1",
+  "type": "object",
+  "additionalProperties": false,
+  "required": ["schema_version", "evidence_generation", "pair_id", "plugin_state", "size_zero", "source_sha256", "calendar_sha256", "external_manifest_sha256", "adjusted_price_field", "timezone", "sessions", "rows", "snapshot_id", "content_sha256"],
+  "properties": {
+    "schema_version": {"const": "soxl_tqqq_clean_cutover_snapshot.v1"},
+    "evidence_generation": {"const": "clean_cutover_v1"},
+    "pair_id": {"const": "QQQ_TQQQ"},
+    "plugin_state": {"const": "ABSENT_DISABLED"},
+    "size_zero": {"const": true},
+    "source_sha256": {"type": "string", "pattern": "^[0-9a-f]{64}$"},
+    "calendar_sha256": {"type": "string", "pattern": "^[0-9a-f]{64}$"},
+    "external_manifest_sha256": {"type": "string", "pattern": "^[0-9a-f]{64}$"},
+    "adjusted_price_field": {"const": "adjusted_close"},
+    "timezone": {"const": "UTC"},
+    "sessions": {"type": "array", "minItems": 1, "uniqueItems": true, "items": {"type": "string", "pattern": "^[0-9]{4}-[0-9]{2}-[0-9]{2}$"}},
+    "rows": {"type": "array", "minItems": 1, "items": {"type": "object", "additionalProperties": false, "required": ["session", "symbol", "adjusted_close"], "properties": {"session": {"type": "string", "pattern": "^[0-9]{4}-[0-9]{2}-[0-9]{2}$"}, "symbol": {"enum": ["QQQ", "TQQQ"]}, "adjusted_close": {"type": "string", "pattern": "^(?:[1-9][0-9]*(\\.[0-9]*[1-9])?|0\\.[0-9]*[1-9])$"}}}},
+    "snapshot_id": {"type": "string", "pattern": "^sha256-[0-9a-f]{64}$"},
+    "content_sha256": {"type": "string", "pattern": "^[0-9a-f]{64}$"}
+  }
+}

src/us_equity_snapshot_pipelines/soxl_tqqq_clean_cutover_snapshot.py

@@ -0,0 +1,206 @@
+"""Offline trusted snapshot boundary for the QQQ/TQQQ clean-cutover package."""
+from __future__ import annotations
+
+import errno
+import hashlib
+import json
+import os
+import stat
+from dataclasses import dataclass
+from datetime import date
+from pathlib import Path
+from typing import Any, ClassVar
+
+SCHEMA_VERSION = "soxl_tqqq_clean_cutover_snapshot.v1"
+EVIDENCE_GENERATION = "clean_cutover_v1"
+_SYMBOLS = {"QQQ", "TQQQ"}
+_REQUIRED = {
+    "schema_version", "evidence_generation", "pair_id", "plugin_state", "size_zero",
+    "source_sha256", "calendar_sha256", "external_manifest_sha256", "adjusted_price_field",
+    "timezone", "sessions", "rows", "snapshot_id", "content_sha256",
+}
+
+
+class SnapshotValidationError(ValueError):
+    """Raised when an immutable snapshot fails the trusted boundary."""
+
+
+def _digest(value: object, label: str) -> str:
+    if type(value) is not str or len(value) != 64 or any(c not in "0123456789abcdef" for c in value):
+        raise SnapshotValidationError(f"invalid {label}")
+    return value
+
+
+@dataclass(frozen=True)
+class ExternalBindings:
+    source_sha256: str
+    calendar_sha256: str
+    manifest_sha256: str
+    content_sha256: str
+
+    def __post_init__(self) -> None:
+        _digest(self.source_sha256, "source_sha256")
+        _digest(self.calendar_sha256, "calendar_sha256")
+        _digest(self.manifest_sha256, "manifest_sha256")
+        _digest(self.content_sha256, "content_sha256")
+
+
+def _canonical_json(value: dict[str, Any]) -> bytes:
+    return (json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n").encode("utf-8")
+
+
+def _pairs(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
+    result: dict[str, Any] = {}
+    for key, value in pairs:
+        if key in result:
+            raise SnapshotValidationError("duplicate JSON key")
+        result[key] = value
+    return result
+
+
+def _canonical_decimal(value: object) -> str:
+    if type(value) is not str or not value or value.startswith(("+", "-")):
+        raise SnapshotValidationError("adjusted_close must be canonical")
+    if value.startswith("0") and value != "0" and not value.startswith("0."):
+        raise SnapshotValidationError("adjusted_close must be canonical")
+    if value.count(".") > 1 or any(c not in "0123456789." for c in value):
+        raise SnapshotValidationError("adjusted_close must be canonical")
+    if value.endswith(".") or ("." in value and value.endswith("0")) or value == "0":
+        raise SnapshotValidationError("adjusted_close must be canonical positive numeric")
+    try:
+        if float(value) <= 0 or not __import__("math").isfinite(float(value)):
+            raise ValueError
+    except ValueError as exc:
+        raise SnapshotValidationError("adjusted_close must be canonical positive numeric") from exc
+    return value
+
+
+def _validate(payload: object, bindings: ExternalBindings) -> dict[str, Any]:
+    if not isinstance(payload, dict) or set(payload) != _REQUIRED:
+        raise SnapshotValidationError("snapshot fields do not match canonical contract")
+    constants = {
+        "schema_version": SCHEMA_VERSION, "evidence_generation": EVIDENCE_GENERATION,
+        "pair_id": "QQQ_TQQQ", "plugin_state": "ABSENT_DISABLED", "size_zero": True,
+        "adjusted_price_field": "adjusted_close", "timezone": "UTC",
+        "source_sha256": bindings.source_sha256, "calendar_sha256": bindings.calendar_sha256,
+        "external_manifest_sha256": bindings.manifest_sha256,
+        "snapshot_id": f"sha256-{bindings.manifest_sha256}",
+    }
+    for key, expected in constants.items():
+        if type(expected) is bool:
+            valid = type(payload.get(key)) is bool and payload.get(key) is expected
+        else:
+            valid = payload.get(key) == expected
+        if not valid:
+            raise SnapshotValidationError(f"{key} mismatch")
+    sessions = payload["sessions"]
+    if not isinstance(sessions, list) or not sessions or any(type(s) is not str for s in sessions) or sessions != sorted(set(sessions)):
+        raise SnapshotValidationError("sessions must be sorted and unique")
+    for session in sessions:
+        if type(session) is not str or len(session) != 10:
+            raise SnapshotValidationError("session must be canonical")
+        try:
+            if date.fromisoformat(session).isoformat() != session:
+                raise ValueError
+        except ValueError as exc:
+            raise SnapshotValidationError("session must be canonical") from exc
+    rows = payload["rows"]
+    if not isinstance(rows, list) or not rows:
+        raise SnapshotValidationError("rows must be non-empty")
+    expected_pairs: list[tuple[str, str]] = []
+    for row in rows:
+        if not isinstance(row, dict) or set(row) != {"session", "symbol", "adjusted_close"}:
+            raise SnapshotValidationError("row fields do not match canonical contract")
+        session, symbol = row["session"], row["symbol"]
+        if type(session) is not str or type(symbol) is not str or session not in sessions or symbol not in _SYMBOLS:
+            raise SnapshotValidationError("row identity mismatch")
+        _canonical_decimal(row["adjusted_close"])
+        expected_pairs.append((session, symbol))
+    if expected_pairs != sorted(expected_pairs) or len(expected_pairs) != len(set(expected_pairs)):
+        raise SnapshotValidationError("rows must be sorted and unique")
+    for session in sessions:
+        if {(s, sym) for s, sym in expected_pairs if s == session} != {(session, "QQQ"), (session, "TQQQ")}:
+            raise SnapshotValidationError("each session must contain QQQ and TQQQ")
+    content = payload["content_sha256"]
+    _digest(content, "content_sha256")
+    unsigned = dict(payload)
+    unsigned.pop("content_sha256")
+    expected_content = hashlib.sha256(_canonical_json(unsigned)).hexdigest()
+    if content != expected_content:
+        raise SnapshotValidationError("content digest mismatch")
+    return payload
+
+
+@dataclass(frozen=True, init=False)
+class TrustedSnapshotPackage:
+    """Canonical, externally-bound snapshot package; construct only via ``read``."""
+
+    path: Path
+    snapshot_id: str
+    _bytes: bytes
+    _TOKEN: ClassVar[object] = object()
+
+    def __init__(self, *args: object, **kwargs: object) -> None:
+        raise TypeError("TrustedSnapshotPackage must be created by read()")
+
+    @classmethod
+    def _create(cls, path: Path, raw: bytes, payload: dict[str, Any]) -> TrustedSnapshotPackage:
+        self = object.__new__(cls)
+        object.__setattr__(self, "path", path)
+        object.__setattr__(self, "snapshot_id", payload["snapshot_id"])
+        object.__setattr__(self, "_bytes", raw)
+        return self
+
+    @classmethod
+    def read(cls, path: str | os.PathLike[str], *, root: str | os.PathLike[str], bindings: ExternalBindings) -> TrustedSnapshotPackage:
+        root_path = Path(root).resolve()
+        candidate = Path(path)
+        lexical = candidate if candidate.is_absolute() else Path.cwd() / candidate
+        lexical = Path(os.path.abspath(lexical))
+        try:
+            relative = lexical.relative_to(root_path)
+        except ValueError as exc:
+            raise SnapshotValidationError("path escapes root") from exc
+        target = root_path / relative
+        if not relative.parts:
+            raise SnapshotValidationError("snapshot path must be a file")
+        try:
+            root_fd = os.open(root_path, os.O_RDONLY | os.O_DIRECTORY | getattr(os, "O_NOFOLLOW", 0))
+        except OSError as exc:
+            raise SnapshotValidationError("snapshot root open failed") from exc
+        fd = root_fd
+        try:
+            for part in relative.parts[:-1]:
+                next_fd = os.open(part, os.O_RDONLY | os.O_DIRECTORY | getattr(os, "O_NOFOLLOW", 0), dir_fd=fd)
+                os.close(fd)
+                fd = next_fd
+            parent_fd = fd
+            fd = os.open(relative.parts[-1], os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0), dir_fd=parent_fd)
+            os.close(parent_fd)
+        except OSError as exc:
+            os.close(fd)
+            message = "symlink path component" if exc.errno == errno.ELOOP else "snapshot readback failed"
+            raise SnapshotValidationError(message) from exc
+        try:
+            if not stat.S_ISREG(os.fstat(fd).st_mode):
+                raise SnapshotValidationError("snapshot must be regular file")
+            raw = b""
+            while chunk := os.read(fd, 1024 * 1024):
+                raw += chunk
+        finally:
+            os.close(fd)
+        if hashlib.sha256(raw).hexdigest() != bindings.content_sha256:
+            raise SnapshotValidationError("content digest mismatch")
+        try:
+            payload = json.loads(raw.decode("utf-8"), object_pairs_hook=_pairs)
+        except SnapshotValidationError:
+            raise
+        except (UnicodeDecodeError, json.JSONDecodeError, ValueError) as exc:
+            raise SnapshotValidationError("invalid snapshot JSON") from exc
+        checked = _validate(payload, bindings)
+        if _canonical_json(checked) != raw:
+            raise SnapshotValidationError("non-canonical readback")
+        return cls._create(target, raw, checked)
+
+    def to_bytes(self) -> bytes:
+        return self._bytes

tests/test_soxl_tqqq_clean_cutover_snapshot.py

@@ -0,0 +1,104 @@
+from __future__ import annotations
+
+import hashlib
+import json
+from pathlib import Path
+
+import pytest
+
+from us_equity_snapshot_pipelines.soxl_tqqq_clean_cutover_snapshot import (
+    ExternalBindings,
+    SnapshotValidationError,
+    TrustedSnapshotPackage,
+)
+
+SOURCE = "1" * 64
+CALENDAR = "2" * 64
+MANIFEST = "3" * 64
+
+
+def payload() -> dict:
+    return {
+        "schema_version": "soxl_tqqq_clean_cutover_snapshot.v1",
+        "evidence_generation": "clean_cutover_v1",
+        "pair_id": "QQQ_TQQQ",
+        "plugin_state": "ABSENT_DISABLED",
+        "size_zero": True,
+        "source_sha256": SOURCE,
+        "calendar_sha256": CALENDAR,
+        "external_manifest_sha256": MANIFEST,
+        "adjusted_price_field": "adjusted_close",
+        "timezone": "UTC",
+        "sessions": ["2026-07-24"],
+        "rows": [
+            {"session": "2026-07-24", "symbol": "QQQ", "adjusted_close": "45.25"},
+            {"session": "2026-07-24", "symbol": "TQQQ", "adjusted_close": "12.5"},
+        ],
+        "snapshot_id": f"sha256-{MANIFEST}",
+    }
+
+
+def write_snapshot(path: Path, value: dict | None = None) -> None:
+    value = value or payload()
+    unsigned = (json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + "\n").encode()
+    value = {**value, "content_sha256": hashlib.sha256(unsigned).hexdigest()}
+    path.write_bytes((json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + "\n").encode())
+
+
+def bindings(path: Path) -> ExternalBindings:
+    return ExternalBindings(source_sha256=SOURCE, calendar_sha256=CALENDAR, manifest_sha256=MANIFEST, content_sha256=hashlib.sha256(path.read_bytes()).hexdigest())
+
+
+def test_readback_and_canonical_serializer(tmp_path: Path) -> None:
+    path = tmp_path / "snapshot.json"
+    write_snapshot(path)
+    package = TrustedSnapshotPackage.read(path, root=tmp_path, bindings=bindings(path))
+    assert package.snapshot_id == f"sha256-{MANIFEST}"
+    assert package.to_bytes() == path.read_bytes()
+
+
+def test_external_digest_binding_and_file_safety(tmp_path: Path) -> None:
+    path = tmp_path / "snapshot.json"
+    write_snapshot(path)
+    with pytest.raises(SnapshotValidationError, match="source_sha256"):
+        TrustedSnapshotPackage.read(path, root=tmp_path, bindings=ExternalBindings("9" * 64, CALENDAR, MANIFEST, hashlib.sha256(path.read_bytes()).hexdigest()))
+    link = tmp_path / "link.json"
+    link.symlink_to(path)
+    with pytest.raises(SnapshotValidationError, match="symlink"):
+        TrustedSnapshotPackage.read(link, root=tmp_path, bindings=bindings(path))
+
+
+def test_content_digest_is_external_binding(tmp_path: Path) -> None:
+    path = tmp_path / "snapshot.json"
+    write_snapshot(path)
+    supplied = ExternalBindings(SOURCE, CALENDAR, MANIFEST, "9" * 64)
+    with pytest.raises(SnapshotValidationError, match="content digest"):
+        TrustedSnapshotPackage.read(path, root=tmp_path, bindings=supplied)
+
+
+def test_invalid_root_and_row_identity_fail_closed(tmp_path: Path) -> None:
+    with pytest.raises(SnapshotValidationError, match="must be a file"):
+        TrustedSnapshotPackage.read(tmp_path, root=tmp_path, bindings=ExternalBindings(SOURCE, CALENDAR, MANIFEST, "0" * 64))
+    path = tmp_path / "snapshot.json"
+    value = payload()
+    value["rows"][0]["symbol"] = []
+    write_snapshot(path, value)
+    with pytest.raises(SnapshotValidationError, match="row identity"):
+        TrustedSnapshotPackage.read(path, root=tmp_path, bindings=bindings(path))
+
+
+def test_duplicate_keys_and_noncanonical_bytes_rejected(tmp_path: Path) -> None:
+    path = tmp_path / "snapshot.json"
+    raw = json.dumps(payload(), sort_keys=True, separators=(",", ":")).replace("\"pair_id\":\"QQQ_TQQQ\"", "\"pair_id\":\"QQQ_TQQQ\",\"pair_id\":\"QQQ_TQQQ\"")
+    path.write_text(raw)
+    with pytest.raises(SnapshotValidationError, match="duplicate"):
+        TrustedSnapshotPackage.read(path, root=tmp_path, bindings=bindings(path))
+    write_snapshot(path)
+    path.write_bytes(path.read_bytes().replace(b"\"12.5\"", b"12.50"))
+    with pytest.raises(SnapshotValidationError, match="canonical"):
+        TrustedSnapshotPackage.read(path, root=tmp_path, bindings=bindings(path))
+
+
+def test_public_boundary_rejects_raw_inputs() -> None:
+    with pytest.raises(TypeError):
+        TrustedSnapshotPackage(b"raw")  # type: ignore[arg-type]