diff --git a/src/research_signal_context_pipelines/__init__.py b/src/research_signal_context_pipelines/__init__.py index 99b98fe..61bc4d4 100644 --- a/src/research_signal_context_pipelines/__init__.py +++ b/src/research_signal_context_pipelines/__init__.py @@ -2,6 +2,7 @@ from .overlay_backtest import OverlayPolicy, backtest_overlay from .context_bundle import DEFAULT_UNIVERSE, build_context_bundle, build_context_from_source +from .latest_linkage import validate_latest_signal from .price_history import PriceExtractionSummary, write_filtered_price_history from .schema import SignalValidationError, validate_signal from .theme_momentum import ( @@ -24,6 +25,7 @@ "load_symbol_theme_exposure", "load_theme_taxonomy", "validate_signal", + "validate_latest_signal", "validate_theme_momentum_snapshot", "write_filtered_price_history", "write_theme_momentum_snapshot", diff --git a/src/research_signal_context_pipelines/latest_linkage.py b/src/research_signal_context_pipelines/latest_linkage.py new file mode 100644 index 0000000..3b35d3f --- /dev/null +++ b/src/research_signal_context_pipelines/latest_linkage.py @@ -0,0 +1,198 @@ +from __future__ import annotations + +import hashlib +import json +import datetime as dt +import errno +import os +import re +import stat +from pathlib import Path +from typing import Any + +from .schema import SignalValidationError, validate_signal +from .theme_momentum import validate_theme_momentum_snapshot + + +_SHA256_RE = re.compile(r"^[0-9a-fA-F]{64}$") +_OPEN_SUPPORTS_DIR_FD = os.open in getattr(os, "supports_dir_fd", set()) +MAX_JSON_ARTIFACT_BYTES = 4 * 1024 * 1024 + + +def validate_latest_signal( + payload: dict[str, Any], + *, + signal_base_dir: str | Path | None = None, + source_override: str | Path | None = None, +) -> None: + """Validate a v2 signal and its single, point-in-time theme snapshot.""" + try: + validate_signal(payload) + except SignalValidationError: + raise + except (TypeError, ValueError, KeyError) as exc: + raise SignalValidationError(f"invalid signal: {exc}") from exc + + if payload.get("schema_version") != "2": + raise SignalValidationError("latest signal schema_version must be '2'") + + evidence = payload["evidence"] + sources = evidence["sources"] + if len(sources) != 1: + raise SignalValidationError("latest signal evidence.sources must declare exactly one source") + declared_source = sources[0] + if not isinstance(declared_source, str) or not declared_source.strip(): + raise SignalValidationError("latest signal evidence.sources must contain a path") + + try: + base = Path(signal_base_dir).resolve() if signal_base_dir is not None else None + except OSError as exc: + raise SignalValidationError("invalid signal_base_dir") from exc + source_path = _resolve_source_path(declared_source, base) + if source_override is not None: + override_path = _resolve_source_path(str(source_override), base) + if override_path != source_path: + raise SignalValidationError("source override must be the same declaration file") + + raw = _read_declaration(source_path, base, declared_source) + _validate_source_hash(evidence, payload, declared_source, source_path, raw) + snapshot = _decode_declaration(raw) + try: + validate_theme_momentum_snapshot(snapshot) + except SignalValidationError: + raise + except (TypeError, ValueError, KeyError) as exc: + raise SignalValidationError(f"invalid theme momentum snapshot: {exc}") from exc + + if snapshot.get("schema_version") != "2": + raise SignalValidationError("theme momentum snapshot schema_version must be '2'") + if snapshot.get("mode") != "theme_momentum_snapshot": + raise SignalValidationError("source must declare theme_momentum_snapshot") + if snapshot.get("artifact_type") != "medium_horizon_theme_context": + raise SignalValidationError("theme snapshot artifact_type must be 'medium_horizon_theme_context'") + if snapshot.get("as_of") != payload.get("as_of"): + raise SignalValidationError("signal and theme snapshot as_of must match") + if _parse_datetime(snapshot["generated_at"], "theme snapshot generated_at") > _parse_datetime( + payload["generated_at"], "signal generated_at" + ): + raise SignalValidationError("theme snapshot generated_at must not be later than signal generated_at") + + +def _resolve_source_path(source: str, base: Path | None) -> Path: + raw_path = Path(source) + if not raw_path.is_absolute() and base is None: + raise SignalValidationError("relative source requires signal_base_dir") + try: + resolved = raw_path.resolve() if raw_path.is_absolute() else (base / raw_path).resolve() + except OSError as exc: + raise SignalValidationError("unable to resolve source path") from exc + if base is not None: + try: + resolved.relative_to(base) + except ValueError as exc: + raise SignalValidationError("source must remain within signal_base_dir") from exc + return resolved + + +def _read_declaration(path: Path, base: Path | None, declared_source: str) -> bytes: + if ( + getattr(os, "O_NOFOLLOW", None) is None + or getattr(os, "O_DIRECTORY", None) is None + or getattr(os, "O_NONBLOCK", None) is None + ): + raise SignalValidationError("secure descriptor-based source reading is unavailable") + if not _OPEN_SUPPORTS_DIR_FD: + raise SignalValidationError("secure openat source reading is unavailable") + fd = -1 + try: + if base is None: + fd = os.open(Path(declared_source), os.O_RDONLY | os.O_NOFOLLOW | getattr(os, "O_NONBLOCK", 0)) + else: + raw_path = Path(declared_source) + try: + relative = raw_path.relative_to(base) if raw_path.is_absolute() else raw_path + except ValueError as exc: + raise SignalValidationError("absolute source must be beneath signal_base_dir") from exc + parts = relative.parts + if not parts or any(part in {"", ".", ".."} for part in parts): + raise SignalValidationError("source path must be a file beneath signal_base_dir") + fd = os.open(base, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) + for part in parts[:-1]: + next_fd = os.open(part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=fd) + os.close(fd) + fd = next_fd + final_fd = -1 + try: + final_fd = os.open( + parts[-1], os.O_RDONLY | os.O_NOFOLLOW | getattr(os, "O_NONBLOCK", 0), dir_fd=fd + ) + finally: + parent_fd = fd + fd = final_fd + os.close(parent_fd) + metadata = os.fstat(fd) + if not stat.S_ISREG(metadata.st_mode): + raise SignalValidationError("declaration source must be a regular file") + if metadata.st_size > MAX_JSON_ARTIFACT_BYTES: + raise SignalValidationError("declaration source exceeds maximum size") + with os.fdopen(fd, "rb") as stream: + fd = -1 + return stream.read() + except SignalValidationError: + raise + except OSError as exc: + if exc.errno == errno.ELOOP: + raise SignalValidationError("no-follow source read rejected a symlink") from exc + raise SignalValidationError(f"unable to securely read declaration file: {path}") from exc + finally: + if fd >= 0: + os.close(fd) + + +def _decode_declaration(raw: bytes) -> dict[str, Any]: + try: + value = json.loads(raw) + except (json.JSONDecodeError, UnicodeDecodeError, TypeError) as exc: + raise SignalValidationError("declaration file must contain valid JSON") from exc + if not isinstance(value, dict): + raise SignalValidationError("declaration file must contain a JSON object") + return value + + +def _validate_source_hash( + evidence: dict[str, Any], + payload: dict[str, Any], + declared_source: str, + source_path: Path, + raw: bytes, +) -> None: + locations = [container["source_hashes"] for container in (evidence, payload) if "source_hashes" in container] + if not locations: + return + if len(locations) != 1: + raise SignalValidationError("source_hashes must be declared in exactly one location") + hashes = locations[0] + if not isinstance(hashes, dict): + raise SignalValidationError("source_hashes must be an object") + allowed_keys = {declared_source, str(source_path)} + if len(hashes) != 1 or not set(hashes).issubset(allowed_keys): + raise SignalValidationError("source_hashes must cover only the declared source") + expected = next(iter(hashes.values())) + if not isinstance(expected, str) or not _SHA256_RE.fullmatch(expected): + raise SignalValidationError("source_hashes must contain a SHA-256 digest") + actual = hashlib.sha256(raw).hexdigest() + if actual.lower() != expected.lower(): + raise SignalValidationError("source SHA-256 hash mismatch") + + +def _parse_datetime(value: Any, name: str) -> dt.datetime: + if not isinstance(value, str) or not value.strip(): + raise SignalValidationError(f"{name} must be an ISO datetime") + normalized = value[:-1] + "+00:00" if value.endswith("Z") else value + try: + parsed = dt.datetime.fromisoformat(normalized) + except ValueError as exc: + raise SignalValidationError(f"{name} must be an ISO datetime") from exc + if parsed.tzinfo is None: + raise SignalValidationError(f"{name} must be an ISO datetime with an explicit timezone") + return parsed.astimezone(dt.timezone.utc) diff --git a/tests/test_latest_linkage.py b/tests/test_latest_linkage.py new file mode 100644 index 0000000..00ec7e8 --- /dev/null +++ b/tests/test_latest_linkage.py @@ -0,0 +1,242 @@ +from __future__ import annotations + +import copy +import hashlib +import json +import os +from pathlib import Path + +import pytest + +import research_signal_context_pipelines.latest_linkage as linkage +from research_signal_context_pipelines import SignalValidationError, validate_latest_signal, validate_signal + + +def signal_payload(source: str = "theme_momentum_snapshot.json") -> dict: + return { + "schema_version": "2", + "model_version": "shadow-v2", + "scoring_version": "rules-v2", + "as_of": "2026-06-26", + "generated_at": "2026-06-27T00:00:00Z", + "mode": "shadow", + "horizon": "1-3 years", + "universe": ["MU"], + "regime": "mixed", + "risk_flags": [], + "candidate_bias": {"MU": "watch"}, + "confidence": 0.5, + "evidence": { + "sources": [source], + "summary": "strict linkage test", + "data_gaps": [], + }, + "expires_at": "2026-07-26", + "policy": {"execution_allowed": False, "downstream_use": "research only"}, + } + + +def snapshot_payload(as_of: str = "2026-06-26", generated_at: str = "2026-06-27T00:00:00Z") -> dict: + return { + "schema_version": "2", + "as_of": as_of, + "generated_at": generated_at, + "expires_at": "2026-07-26", + "model_version": "theme-v2", + "scoring_version": "momentum-v2", + "mode": "theme_momentum_snapshot", + "artifact_type": "medium_horizon_theme_context", + "theme_ranks": [], + "data_quality": {}, + "policy": {}, + } + + +def write_snapshot( + tmp_path: Path, + *, + as_of: str = "2026-06-26", + generated_at: str = "2026-06-27T00:00:00Z", +) -> Path: + path = tmp_path / "theme_momentum_snapshot.json" + path.write_bytes(json.dumps(snapshot_payload(as_of, generated_at)).encode()) + return path + + +def test_validate_latest_signal_accepts_v2_and_matching_snapshot(tmp_path: Path) -> None: + write_snapshot(tmp_path) + + validate_latest_signal(signal_payload(), signal_base_dir=tmp_path) + + +def test_validate_latest_signal_rejects_v1_without_changing_legacy_validator(tmp_path: Path) -> None: + write_snapshot(tmp_path) + payload = copy.deepcopy(signal_payload()) + payload.update({"schema_version": "1"}) + payload.pop("model_version") + payload.pop("scoring_version") + + validate_signal(payload) + with pytest.raises(SignalValidationError, match="schema_version"): + validate_latest_signal(payload, signal_base_dir=tmp_path) + + +def test_relative_source_requires_base_dir(tmp_path: Path) -> None: + write_snapshot(tmp_path) + + with pytest.raises(SignalValidationError, match="signal_base_dir"): + validate_latest_signal(signal_payload()) + + +def test_source_path_escape_and_symlink_escape_are_rejected(tmp_path: Path) -> None: + write_snapshot(tmp_path) + outside = tmp_path.parent / "outside.json" + outside.write_bytes(json.dumps(snapshot_payload()).encode()) + (tmp_path / "link.json").symlink_to(outside) + + for source in ("../outside.json", "link.json"): + with pytest.raises(SignalValidationError, match="within signal_base_dir"): + validate_latest_signal(signal_payload(source), signal_base_dir=tmp_path) + + +def test_final_symlink_is_rejected_without_following_it(tmp_path: Path) -> None: + write_snapshot(tmp_path) + (tmp_path / "alias.json").symlink_to(tmp_path / "theme_momentum_snapshot.json") + + with pytest.raises(SignalValidationError, match="no-follow|symlink"): + validate_latest_signal(signal_payload("alias.json"), signal_base_dir=tmp_path) + + +def test_absolute_declared_symlink_is_rejected_without_resolving_identity(tmp_path: Path) -> None: + write_snapshot(tmp_path) + alias = tmp_path / "absolute-alias.json" + alias.symlink_to(tmp_path / "theme_momentum_snapshot.json") + + with pytest.raises(SignalValidationError, match="no-follow|symlink"): + validate_latest_signal(signal_payload(str(alias))) + + +def test_descriptor_read_rejects_symlink_swap_after_bounds_check(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + path = write_snapshot(tmp_path) + outside = tmp_path.parent / "outside.json" + outside.write_bytes(json.dumps(snapshot_payload()).encode()) + original = linkage._resolve_source_path + + def swap_after_resolve(source: str, base: Path | None) -> Path: + resolved = original(source, base) + if source == "theme_momentum_snapshot.json": + path.unlink() + path.symlink_to(outside) + return resolved + + monkeypatch.setattr(linkage, "_resolve_source_path", swap_after_resolve) + with pytest.raises(SignalValidationError, match="no-follow|symlink"): + validate_latest_signal(signal_payload(), signal_base_dir=tmp_path) + + +def test_override_must_match_declared_source(tmp_path: Path) -> None: + declared = write_snapshot(tmp_path) + other = tmp_path / "other.json" + other.write_bytes(json.dumps(snapshot_payload()).encode()) + + with pytest.raises(SignalValidationError, match="override"): + validate_latest_signal(signal_payload(), signal_base_dir=tmp_path, source_override=other) + validate_latest_signal(signal_payload(), signal_base_dir=tmp_path, source_override=declared) + + +def test_declaration_is_read_once_and_same_bytes_are_hashed_and_decoded(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + path = write_snapshot(tmp_path) + raw = path.read_bytes() + payload = signal_payload() + payload["evidence"]["source_hashes"] = {"theme_momentum_snapshot.json": hashlib.sha256(raw).hexdigest()} + calls = 0 + original_open = os.open + + def open_once(*args: object, **kwargs: object) -> int: + nonlocal calls + fd = original_open(*args, **kwargs) + calls += 1 + return fd + + monkeypatch.setattr(os, "open", open_once) + validate_latest_signal(payload, signal_base_dir=tmp_path) + assert calls == 2 + + +@pytest.mark.parametrize("generated_at", ["2026-06-27", "2026-06-27T00:00:00"]) +def test_generated_at_requires_time_and_explicit_timezone(tmp_path: Path, generated_at: str) -> None: + payload = signal_payload() + payload["generated_at"] = generated_at + write_snapshot(tmp_path) + + with pytest.raises(SignalValidationError, match="ISO datetime"): + validate_latest_signal(payload, signal_base_dir=tmp_path) + + payload = signal_payload() + write_snapshot(tmp_path, generated_at=generated_at) + with pytest.raises(SignalValidationError, match="ISO datetime"): + validate_latest_signal(payload, signal_base_dir=tmp_path) + + +def test_non_regular_and_oversized_declarations_fail_closed(tmp_path: Path) -> None: + fifo = tmp_path / "theme_momentum_snapshot.json" + os.mkfifo(fifo) + with pytest.raises(SignalValidationError, match="regular file"): + validate_latest_signal(signal_payload(), signal_base_dir=tmp_path) + + fifo.unlink() + fifo.write_bytes(b"{}") + with fifo.open("ab") as stream: + stream.truncate(linkage.MAX_JSON_ARTIFACT_BYTES + 1) + with pytest.raises(SignalValidationError, match="maximum size"): + validate_latest_signal(signal_payload(), signal_base_dir=tmp_path) + + +def test_bad_hash_and_as_of_mismatch_are_validation_errors(tmp_path: Path) -> None: + path = write_snapshot(tmp_path) + payload = signal_payload() + payload["evidence"]["source_hashes"] = {"theme_momentum_snapshot.json": "0" * 64} + with pytest.raises(SignalValidationError, match="SHA-256"): + validate_latest_signal(payload, signal_base_dir=tmp_path) + + +def test_source_hashes_must_cover_only_the_declared_source(tmp_path: Path) -> None: + path = write_snapshot(tmp_path) + digest = hashlib.sha256(path.read_bytes()).hexdigest() + + for hashes in ( + {}, + {"theme_momentum_snapshot.json": digest, str(path): digest}, + {"theme_momentum_snapshot.json": digest, "other.json": digest}, + ): + payload = signal_payload() + payload["evidence"]["source_hashes"] = hashes + with pytest.raises(SignalValidationError, match="source_hashes"): + validate_latest_signal(payload, signal_base_dir=tmp_path) + + +def test_strict_linkage_requires_snapshot_contract_and_point_in_time_order(tmp_path: Path) -> None: + path = write_snapshot(tmp_path, generated_at="2026-06-27T00:00:01Z") + payload = signal_payload() + with pytest.raises(SignalValidationError, match="generated_at"): + validate_latest_signal(payload, signal_base_dir=tmp_path) + + path.write_bytes(json.dumps({**snapshot_payload(), "artifact_type": "other"}).encode()) + with pytest.raises(SignalValidationError, match="artifact_type"): + validate_latest_signal(payload, signal_base_dir=tmp_path) + + payload = signal_payload() + path.write_bytes(json.dumps(snapshot_payload("2026-06-25")).encode()) + with pytest.raises(SignalValidationError, match="as_of"): + validate_latest_signal(payload, signal_base_dir=tmp_path) + + +def test_io_json_and_snapshot_validation_errors_are_normalized(tmp_path: Path) -> None: + payload = signal_payload() + with pytest.raises(SignalValidationError): + validate_latest_signal(payload, signal_base_dir=tmp_path) + + path = tmp_path / "theme_momentum_snapshot.json" + path.write_bytes(b"not-json") + with pytest.raises(SignalValidationError, match="JSON"): + validate_latest_signal(payload, signal_base_dir=tmp_path)