From 59962ac6ab3c899347d36b0242cc5c25d30da506 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Mon, 27 Jul 2026 13:44:04 +0800 Subject: [PATCH 1/4] Add trusted snapshot package boundary Co-Authored-By: Codex --- ...tqqq_clean_cutover_snapshot.v1.schema.json | 23 +++ .../soxl_tqqq_clean_cutover_snapshot.py | 186 ++++++++++++++++++ .../test_soxl_tqqq_clean_cutover_snapshot.py | 85 ++++++++ 3 files changed, 294 insertions(+) create mode 100644 schemas/soxl_tqqq_clean_cutover_snapshot.v1.schema.json create mode 100644 src/us_equity_snapshot_pipelines/soxl_tqqq_clean_cutover_snapshot.py create mode 100644 tests/test_soxl_tqqq_clean_cutover_snapshot.py 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..4c118da --- /dev/null +++ b/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": "^(0|[1-9][0-9]*)(\\.[0-9]+)?$"}}}}, + "snapshot_id": {"type": "string", "pattern": "^sha256-[0-9a-f]{64}$"}, + "content_sha256": {"type": "string", "pattern": "^[0-9a-f]{64}$"} + } +} 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..e81fd70 --- /dev/null +++ b/src/us_equity_snapshot_pipelines/soxl_tqqq_clean_cutover_snapshot.py @@ -0,0 +1,186 @@ +"""Offline trusted snapshot boundary for the QQQ/TQQQ clean-cutover package.""" +from __future__ import annotations + +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 + + def __post_init__(self) -> None: + _digest(self.source_sha256, "source_sha256") + _digest(self.calendar_sha256, "calendar_sha256") + _digest(self.manifest_sha256, "manifest_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("+") 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 payload.get(key) != expected: + raise SnapshotValidationError(f"{key} mismatch") + sessions = payload["sessions"] + if not isinstance(sessions, list) or not 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 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 + current = root_path + for part in relative.parts: + current = current / part + if current.is_symlink(): + raise SnapshotValidationError("symlink path component") + flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) + try: + fd = os.open(target, flags) + except OSError as exc: + raise SnapshotValidationError("snapshot readback failed") 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) + try: + payload = json.loads(raw.decode("utf-8"), object_pairs_hook=_pairs) + except (UnicodeDecodeError, json.JSONDecodeError) 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 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..527075d --- /dev/null +++ b/tests/test_soxl_tqqq_clean_cutover_snapshot.py @@ -0,0 +1,85 @@ +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() -> ExternalBindings: + return ExternalBindings(source_sha256=SOURCE, calendar_sha256=CALENDAR, manifest_sha256=MANIFEST) + + +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()) + 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)) + link = tmp_path / "link.json" + link.symlink_to(path) + with pytest.raises(SnapshotValidationError, match="symlink"): + TrustedSnapshotPackage.read(link, root=tmp_path, bindings=bindings()) + + +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()) + 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()) + + +def test_public_boundary_rejects_raw_inputs() -> None: + with pytest.raises(TypeError): + TrustedSnapshotPackage(b"raw", bindings()) # type: ignore[arg-type] From 20eb7511051feed617dfb3cdea7b97f3d25f0911 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Mon, 27 Jul 2026 13:44:56 +0800 Subject: [PATCH 2/4] Tighten trusted snapshot lint compliance Co-Authored-By: Codex --- .../soxl_tqqq_clean_cutover_snapshot.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) 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 index e81fd70..2305a52 100644 --- a/src/us_equity_snapshot_pipelines/soxl_tqqq_clean_cutover_snapshot.py +++ b/src/us_equity_snapshot_pipelines/soxl_tqqq_clean_cutover_snapshot.py @@ -56,7 +56,7 @@ def _pairs(pairs: list[tuple[str, Any]]) -> dict[str, Any]: def _canonical_decimal(value: object) -> str: - if type(value) is not str or not value or value.startswith("+") or value.startswith("-"): + 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") @@ -137,7 +137,7 @@ 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": + 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"]) @@ -145,7 +145,7 @@ def _create(cls, path: Path, raw: bytes, payload: dict[str, Any]) -> "TrustedSna return self @classmethod - def read(cls, path: str | os.PathLike[str], *, root: str | os.PathLike[str], bindings: ExternalBindings) -> "TrustedSnapshotPackage": + 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 From e003adb9b011b9262689d12fc161690b997c1126 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Mon, 27 Jul 2026 14:05:19 +0800 Subject: [PATCH 3/4] Harden trusted snapshot remediation findings Co-Authored-By: Codex --- ...tqqq_clean_cutover_snapshot.v1.schema.json | 2 +- .../soxl_tqqq_clean_cutover_snapshot.py | 37 +++++++++++++------ .../test_soxl_tqqq_clean_cutover_snapshot.py | 24 ++++++++---- 3 files changed, 43 insertions(+), 20 deletions(-) diff --git a/schemas/soxl_tqqq_clean_cutover_snapshot.v1.schema.json b/schemas/soxl_tqqq_clean_cutover_snapshot.v1.schema.json index 4c118da..77fe709 100644 --- a/schemas/soxl_tqqq_clean_cutover_snapshot.v1.schema.json +++ b/schemas/soxl_tqqq_clean_cutover_snapshot.v1.schema.json @@ -16,7 +16,7 @@ "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": "^(0|[1-9][0-9]*)(\\.[0-9]+)?$"}}}}, + "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])?$"}}}}, "snapshot_id": {"type": "string", "pattern": "^sha256-[0-9a-f]{64}$"}, "content_sha256": {"type": "string", "pattern": "^[0-9a-f]{64}$"} } 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 index 2305a52..ce5f315 100644 --- a/src/us_equity_snapshot_pipelines/soxl_tqqq_clean_cutover_snapshot.py +++ b/src/us_equity_snapshot_pipelines/soxl_tqqq_clean_cutover_snapshot.py @@ -1,6 +1,7 @@ """Offline trusted snapshot boundary for the QQQ/TQQQ clean-cutover package.""" from __future__ import annotations +import errno import hashlib import json import os @@ -35,11 +36,13 @@ 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: @@ -84,10 +87,14 @@ def _validate(payload: object, bindings: ExternalBindings) -> dict[str, Any]: "snapshot_id": f"sha256-{bindings.manifest_sha256}", } for key, expected in constants.items(): - if payload.get(key) != expected: + 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 sessions != sorted(set(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: @@ -155,16 +162,20 @@ def read(cls, path: str | os.PathLike[str], *, root: str | os.PathLike[str], bin except ValueError as exc: raise SnapshotValidationError("path escapes root") from exc target = root_path / relative - current = root_path - for part in relative.parts: - current = current / part - if current.is_symlink(): - raise SnapshotValidationError("symlink path component") - flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) + root_fd = os.open(root_path, os.O_RDONLY | os.O_DIRECTORY | getattr(os, "O_NOFOLLOW", 0)) + fd = root_fd try: - fd = os.open(target, flags) + 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: - raise SnapshotValidationError("snapshot readback failed") from 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") @@ -173,9 +184,13 @@ def read(cls, path: str | os.PathLike[str], *, root: str | os.PathLike[str], bin 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 (UnicodeDecodeError, json.JSONDecodeError) as exc: + 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: diff --git a/tests/test_soxl_tqqq_clean_cutover_snapshot.py b/tests/test_soxl_tqqq_clean_cutover_snapshot.py index 527075d..9ff385a 100644 --- a/tests/test_soxl_tqqq_clean_cutover_snapshot.py +++ b/tests/test_soxl_tqqq_clean_cutover_snapshot.py @@ -45,14 +45,14 @@ def write_snapshot(path: Path, value: dict | None = None) -> None: path.write_bytes((json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + "\n").encode()) -def bindings() -> ExternalBindings: - return ExternalBindings(source_sha256=SOURCE, calendar_sha256=CALENDAR, manifest_sha256=MANIFEST) +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()) + package = TrustedSnapshotPackage.read(path, root=tmp_path, bindings=bindings(path)) assert package.snapshot_id == f"sha256-{MANIFEST}" assert package.to_bytes() == path.read_bytes() @@ -61,11 +61,19 @@ 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)) + 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()) + 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_duplicate_keys_and_noncanonical_bytes_rejected(tmp_path: Path) -> None: @@ -73,13 +81,13 @@ def test_duplicate_keys_and_noncanonical_bytes_rejected(tmp_path: Path) -> None: 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()) + 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()) + TrustedSnapshotPackage.read(path, root=tmp_path, bindings=bindings(path)) def test_public_boundary_rejects_raw_inputs() -> None: with pytest.raises(TypeError): - TrustedSnapshotPackage(b"raw", bindings()) # type: ignore[arg-type] + TrustedSnapshotPackage(b"raw") # type: ignore[arg-type] From f8c70b30d66fc53aabdb81664da046207c3ea598 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Mon, 27 Jul 2026 14:33:30 +0800 Subject: [PATCH 4/4] Close remaining snapshot boundary findings Co-Authored-By: Codex --- .../soxl_tqqq_clean_cutover_snapshot.v1.schema.json | 2 +- .../soxl_tqqq_clean_cutover_snapshot.py | 9 +++++++-- tests/test_soxl_tqqq_clean_cutover_snapshot.py | 11 +++++++++++ 3 files changed, 19 insertions(+), 3 deletions(-) diff --git a/schemas/soxl_tqqq_clean_cutover_snapshot.v1.schema.json b/schemas/soxl_tqqq_clean_cutover_snapshot.v1.schema.json index 77fe709..7dcf6be 100644 --- a/schemas/soxl_tqqq_clean_cutover_snapshot.v1.schema.json +++ b/schemas/soxl_tqqq_clean_cutover_snapshot.v1.schema.json @@ -16,7 +16,7 @@ "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])?$"}}}}, + "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}$"} } 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 index ce5f315..ab1ab14 100644 --- a/src/us_equity_snapshot_pipelines/soxl_tqqq_clean_cutover_snapshot.py +++ b/src/us_equity_snapshot_pipelines/soxl_tqqq_clean_cutover_snapshot.py @@ -112,7 +112,7 @@ def _validate(payload: object, bindings: ExternalBindings) -> dict[str, Any]: 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 session not in sessions or symbol not in _SYMBOLS: + 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)) @@ -162,7 +162,12 @@ def read(cls, path: str | os.PathLike[str], *, root: str | os.PathLike[str], bin except ValueError as exc: raise SnapshotValidationError("path escapes root") from exc target = root_path / relative - root_fd = os.open(root_path, os.O_RDONLY | os.O_DIRECTORY | getattr(os, "O_NOFOLLOW", 0)) + 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]: diff --git a/tests/test_soxl_tqqq_clean_cutover_snapshot.py b/tests/test_soxl_tqqq_clean_cutover_snapshot.py index 9ff385a..4654955 100644 --- a/tests/test_soxl_tqqq_clean_cutover_snapshot.py +++ b/tests/test_soxl_tqqq_clean_cutover_snapshot.py @@ -76,6 +76,17 @@ def test_content_digest_is_external_binding(tmp_path: Path) -> None: 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\"")