From 3c96141c9b15c77bfe08f2946ca540d450bd2bfb Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Mon, 27 Jul 2026 09:38:56 +0800 Subject: [PATCH 1/2] harden TQQQ snapshot numeric and atomic boundaries Co-Authored-By: Codex --- .../tqqq_r1_snapshot.py | 101 +++++++++++++++++- tests/test_tqqq_r1_snapshot.py | 24 +++++ 2 files changed, 120 insertions(+), 5 deletions(-) diff --git a/src/us_equity_snapshot_pipelines/tqqq_r1_snapshot.py b/src/us_equity_snapshot_pipelines/tqqq_r1_snapshot.py index 82441b3..286bcfd 100644 --- a/src/us_equity_snapshot_pipelines/tqqq_r1_snapshot.py +++ b/src/us_equity_snapshot_pipelines/tqqq_r1_snapshot.py @@ -2,20 +2,22 @@ from __future__ import annotations +import ctypes +import errno import hashlib -from io import BytesIO import json import math import os import shutil +import sys import tempfile from dataclasses import dataclass +from io import BytesIO from pathlib import Path from typing import Any import pandas as pd - CONTRACT_VERSION = "tqqq_r1_qqq_tqqq_immutable_snapshot.v2" SYMBOLS = ("QQQ", "TQQQ") REQUESTED_LOWER_BOUND = "2010-01-01" @@ -23,6 +25,7 @@ PLUGIN = "ABSENT_DISABLED" MODE = "core_only" OUTPUT_FILENAMES = ("prices.csv", "manifest.json", "validation.json", "sha256sums.json") +_MAX_FLOAT_INTEGER_DECIMAL = format(sys.float_info.max, ".0f") class SnapshotValidationError(ValueError): @@ -47,6 +50,47 @@ def _invalid(message: str) -> None: raise SnapshotValidationError(message) +def _admit_raw_numeric(value: object) -> None: + """Reject non-canonical or out-of-range values before pandas numeric coercion.""" + if isinstance(value, (bool, complex)) or pd.api.types.is_bool(value) or pd.api.types.is_complex(value): + _invalid("invalid adjusted_close numeric form") + if isinstance(value, int) and not isinstance(value, bool): + text = str(value) + if value <= 0 or len(text) > len(_MAX_FLOAT_INTEGER_DECIMAL) or ( + len(text) == len(_MAX_FLOAT_INTEGER_DECIMAL) and text > _MAX_FLOAT_INTEGER_DECIMAL + ): + _invalid("adjusted_close exceeds finite numeric range") + return + if isinstance(value, str): + text = value.strip() + if text.lower() in {"true", "false"}: + _invalid("boolean adjusted_close is not allowed") + if text.isdigit(): + if text != "0" and text.startswith("0"): + _invalid("noncanonical adjusted_close integer") + if text == "0" or len(text) > len(_MAX_FLOAT_INTEGER_DECIMAL) or ( + len(text) == len(_MAX_FLOAT_INTEGER_DECIMAL) and text > _MAX_FLOAT_INTEGER_DECIMAL + ): + _invalid("adjusted_close exceeds finite numeric range") + return + try: + number = float(text) + except (TypeError, ValueError, OverflowError) as exc: + raise SnapshotValidationError("invalid adjusted_close numeric form") from exc + if not math.isfinite(number) or number <= 0: + _invalid("adjusted_close exceeds finite numeric range") + return + if isinstance(value, float): + if not math.isfinite(value) or value <= 0: + _invalid("adjusted_close exceeds finite numeric range") + return + try: + text = str(value) + except Exception as exc: + raise SnapshotValidationError("invalid adjusted_close numeric form") from exc + _admit_raw_numeric(text) + + def _normalize_session(value: object) -> pd.Timestamp: if pd.api.types.is_bool(value): _invalid("invalid session") @@ -123,7 +167,12 @@ def _normalized_prices( _invalid("boolean adjusted_close is not allowed") if pd.api.types.is_datetime64_any_dtype(adjusted_close) or pd.api.types.is_timedelta64_dtype(adjusted_close): _invalid("datetime-like adjusted_close is not allowed") - normalized[PRICE_FIELD] = pd.to_numeric(adjusted_close, errors="coerce") + for value in adjusted_close.array: + _admit_raw_numeric(value) + try: + normalized[PRICE_FIELD] = pd.to_numeric(adjusted_close, errors="coerce") + except (TypeError, ValueError, OverflowError) as exc: + raise SnapshotValidationError("invalid adjusted_close numeric form") from exc if ( normalized[PRICE_FIELD].isna().any() or not normalized[PRICE_FIELD].map(math.isfinite).all() @@ -161,6 +210,48 @@ def no_duplicates(pairs: list[tuple[str, Any]]) -> dict[str, Any]: return value +def _publish_noreplace(source: Path, destination: Path) -> None: + """Publish a directory without clobbering an existing destination.""" + if sys.platform.startswith("linux"): + libc = ctypes.CDLL(None, use_errno=True) + renameat2 = getattr(libc, "renameat2", None) + if renameat2 is None or not hasattr(os, "O_NOFOLLOW") or not hasattr(os, "O_DIRECTORY"): + raise SnapshotValidationError("required no-clobber capability unavailable") + renameat2.argtypes = [ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p, ctypes.c_uint] + renameat2.restype = ctypes.c_int + parent_fd = os.open(destination.parent, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) + try: + result = renameat2( + parent_fd, + source.name.encode(), + parent_fd, + destination.name.encode(), + 1, + ) + finally: + os.close(parent_fd) + if result != 0: + error = ctypes.get_errno() + if error == errno.EEXIST: + raise SnapshotValidationError(f"immutable output already exists: {destination}") + raise SnapshotValidationError("atomic no-clobber publish failed") + return + if sys.platform == "darwin": + libc = ctypes.CDLL(None, use_errno=True) + renamex_np = getattr(libc, "renamex_np", None) + if renamex_np is None or not hasattr(os, "O_NOFOLLOW") or not hasattr(os, "O_DIRECTORY"): + raise SnapshotValidationError("required no-clobber capability unavailable") + renamex_np.argtypes = [ctypes.c_char_p, ctypes.c_char_p, ctypes.c_uint] + renamex_np.restype = ctypes.c_int + result = renamex_np(str(source).encode(), str(destination).encode(), 4) + if result != 0: + error = ctypes.get_errno() + if error == errno.EEXIST: + raise SnapshotValidationError(f"immutable output already exists: {destination}") + raise SnapshotValidationError("atomic no-clobber publish failed") + return + raise SnapshotValidationError("unsupported platform for atomic no-clobber publish") + def _has_exact_type(value: object, expected: object) -> bool: if type(value) is not type(expected): return False @@ -207,7 +298,7 @@ def verify_tqqq_r1_snapshot( try: prices = _normalized_prices( - pd.read_csv(BytesIO(members["prices.csv"])), + pd.read_csv(BytesIO(members["prices.csv"]), dtype={PRICE_FIELD: "string"}), require_exact_columns=True, require_canonical_symbols=True, require_canonical_sessions=True, @@ -280,7 +371,7 @@ def materialize_tqqq_r1_snapshot( ) manifest_sha256 = _sha256(temporary / "manifest.json") verify_tqqq_r1_snapshot(temporary, expected_manifest_sha256=manifest_sha256) - os.replace(temporary, destination) + _publish_noreplace(temporary, destination) except Exception: shutil.rmtree(temporary, ignore_errors=True) raise diff --git a/tests/test_tqqq_r1_snapshot.py b/tests/test_tqqq_r1_snapshot.py index c3ed192..fa25a54 100644 --- a/tests/test_tqqq_r1_snapshot.py +++ b/tests/test_tqqq_r1_snapshot.py @@ -232,3 +232,27 @@ def test_verify_normalizes_csv_parse_failures(tmp_path: Path) -> None: with pytest.raises(snapshot.SnapshotValidationError, match="invalid prices.csv"): snapshot.verify_tqqq_r1_snapshot(output_dir, expected_manifest_sha256=_refresh_trusted_metadata(output_dir)) + + +def test_materialize_rejects_oversized_canonical_integer_before_numeric_conversion(tmp_path: Path) -> None: + prices = _fixture_prices().astype({"adjusted_close": object}) + prices.loc[0, "adjusted_close"] = int("9" * 400) + + with pytest.raises(snapshot.SnapshotValidationError): + snapshot.materialize_tqqq_r1_snapshot(prices, tmp_path / "snapshot") + + +def test_verify_rejects_oversized_canonical_integer_csv(tmp_path: Path) -> None: + result = snapshot.materialize_tqqq_r1_snapshot(_fixture_prices(), tmp_path / "snapshot") + output_dir = result.output_dir + (output_dir / "prices.csv").write_text( + "session,symbol,adjusted_close\n" + + "2010-01-04,QQQ," + "9" * 400 + "\n" + + "2010-01-04,TQQQ,10.5\n" + + "2010-01-05,QQQ,46\n" + + "2010-01-05,TQQQ,11\n", + encoding="utf-8", + ) + + with pytest.raises(snapshot.SnapshotValidationError): + snapshot.verify_tqqq_r1_snapshot(output_dir, expected_manifest_sha256=_refresh_trusted_metadata(output_dir)) From 67219c94688ff81d1b04a76eb33349c623b07092 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Mon, 27 Jul 2026 09:49:20 +0800 Subject: [PATCH 2/2] tighten numeric admission and parent descriptor handling Co-Authored-By: Codex --- src/us_equity_snapshot_pipelines/tqqq_r1_snapshot.py | 10 ++++++++-- tests/test_tqqq_r1_snapshot.py | 8 ++++++++ 2 files changed, 16 insertions(+), 2 deletions(-) diff --git a/src/us_equity_snapshot_pipelines/tqqq_r1_snapshot.py b/src/us_equity_snapshot_pipelines/tqqq_r1_snapshot.py index 286bcfd..d39c865 100644 --- a/src/us_equity_snapshot_pipelines/tqqq_r1_snapshot.py +++ b/src/us_equity_snapshot_pipelines/tqqq_r1_snapshot.py @@ -55,7 +55,12 @@ def _admit_raw_numeric(value: object) -> None: if isinstance(value, (bool, complex)) or pd.api.types.is_bool(value) or pd.api.types.is_complex(value): _invalid("invalid adjusted_close numeric form") if isinstance(value, int) and not isinstance(value, bool): - text = str(value) + if value > 0 and value.bit_length() > 1024: + _invalid("adjusted_close exceeds finite numeric range") + try: + text = str(value) + except ValueError as exc: + raise SnapshotValidationError("adjusted_close exceeds finite numeric range") from exc if value <= 0 or len(text) > len(_MAX_FLOAT_INTEGER_DECIMAL) or ( len(text) == len(_MAX_FLOAT_INTEGER_DECIMAL) and text > _MAX_FLOAT_INTEGER_DECIMAL ): @@ -219,7 +224,8 @@ def _publish_noreplace(source: Path, destination: Path) -> None: raise SnapshotValidationError("required no-clobber capability unavailable") renameat2.argtypes = [ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p, ctypes.c_uint] renameat2.restype = ctypes.c_int - parent_fd = os.open(destination.parent, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) + parent_flags = getattr(os, "O_PATH", os.O_RDONLY) | os.O_DIRECTORY | os.O_NOFOLLOW + parent_fd = os.open(destination.parent, parent_flags) try: result = renameat2( parent_fd, diff --git a/tests/test_tqqq_r1_snapshot.py b/tests/test_tqqq_r1_snapshot.py index fa25a54..8ae631d 100644 --- a/tests/test_tqqq_r1_snapshot.py +++ b/tests/test_tqqq_r1_snapshot.py @@ -256,3 +256,11 @@ def test_verify_rejects_oversized_canonical_integer_csv(tmp_path: Path) -> None: with pytest.raises(snapshot.SnapshotValidationError): snapshot.verify_tqqq_r1_snapshot(output_dir, expected_manifest_sha256=_refresh_trusted_metadata(output_dir)) + + +def test_materialize_rejects_python_int_beyond_digit_limit(tmp_path: Path) -> None: + prices = _fixture_prices().astype({"adjusted_close": object}) + prices.loc[0, "adjusted_close"] = 10**5000 - 1 + + with pytest.raises(snapshot.SnapshotValidationError): + snapshot.materialize_tqqq_r1_snapshot(prices, tmp_path / "snapshot")