diff --git a/scripts/post_shadow_signal_request.py b/scripts/post_shadow_signal_request.py index 0527de5..4be49e2 100644 --- a/scripts/post_shadow_signal_request.py +++ b/scripts/post_shadow_signal_request.py @@ -4,9 +4,11 @@ import argparse from collections.abc import Mapping import datetime as dt +import hashlib import json import os from pathlib import Path +import re import sys import urllib.error import urllib.parse @@ -16,6 +18,7 @@ DEFAULT_API_URL = "https://api.github.com" DEFAULT_LABEL = "long-horizon-shadow" +COMMIT_SHA_PATTERN = re.compile(r"[0-9a-fA-F]{40}") def github_request(method: str, url: str, token: str, payload: dict[str, Any] | None = None) -> Any: @@ -75,6 +78,15 @@ def load_context_bundle(path: str | None) -> dict[str, Any] | None: return json.loads(Path(path).read_text(encoding="utf-8")) +def build_immutable_provenance(context_path: Path, producer_commit_sha: str) -> dict[str, str]: + if not COMMIT_SHA_PATTERN.fullmatch(producer_commit_sha): + raise ValueError("producer commit must be an immutable 40-hex commit") + return { + "producer_commit_sha": producer_commit_sha, + "input_digest": f"sha256:{hashlib.sha256(context_path.read_bytes()).hexdigest()}", + } + + def resolve_as_of_date(raw_as_of_date: str | None, context_bundle: Mapping[str, Any] | None) -> str: if raw_as_of_date: return raw_as_of_date @@ -114,6 +126,7 @@ def build_issue_body( provider: str, bridge_repository: str, context_bundle: Mapping[str, Any] | None = None, + immutable_provenance: Mapping[str, str] | None = None, ) -> str: sections = [ [ @@ -141,6 +154,41 @@ def build_issue_body( "", ], [context_markdown(context_bundle)], + [ + "", + "## Immutable Provenance Contract", + "", + "Any new signal must update `latest_signal.json` and its sibling manifest in the same PR.", + "The manifest must use this v2 contract; v1 is legacy_untrusted and cannot authorize downstream use.", + "", + "```json", + json.dumps( + { + "manifest_type": "research_signal_context", + "schema_version": 2, + "artifact": { + "path": "data/output/latest_signal.json", + "sha256": "sha256 of exact latest_signal.json bytes", + }, + "as_of": "copy exactly from signal", + "generated_at": "copy exactly from signal", + "expires_at": "copy exactly from signal", + "mode": "copy exactly from signal", + "producer": { + "repository": "QuantStrategyLab/ResearchSignalContextPipelines", + "commit_sha": (immutable_provenance or {}).get("producer_commit_sha", "required 40-hex commit"), + }, + "input_digest": (immutable_provenance or {}).get("input_digest", "required sha256 digest"), + "policy": {"execution_allowed": False}, + }, + ensure_ascii=True, + indent=2, + sort_keys=True, + ), + "```", + "", + "Do not add a publisher or publication self-commit field. Do not upgrade existing v1 artifacts by hand.", + ], [ "", "Do not infer historical AI signals that were not generated point-in-time.", @@ -201,16 +249,28 @@ def main() -> int: if not token: print("GITHUB_TOKEN is required", file=sys.stderr) return 1 + if not args.context_file: + print("immutable provenance requires --context-file", file=sys.stderr) + return 1 context_bundle = load_context_bundle(args.context_file) as_of_date = resolve_as_of_date(args.as_of_date, context_bundle) title = build_issue_title(as_of_date) + try: + immutable_provenance = build_immutable_provenance( + Path(args.context_file), + os.environ.get("GITHUB_SHA", args.source_ref), + ) + except (OSError, ValueError) as exc: + print(f"immutable provenance unavailable: {exc}", file=sys.stderr) + return 1 body = build_issue_body( as_of_date=as_of_date, source_ref=args.source_ref, provider=args.provider, bridge_repository=args.bridge_repository, context_bundle=context_bundle, + immutable_provenance=immutable_provenance, ) try: action, issue_number, issue_url = upsert_issue( diff --git a/scripts/validate_latest_signal.py b/scripts/validate_latest_signal.py index bfd98cc..24f2b09 100644 --- a/scripts/validate_latest_signal.py +++ b/scripts/validate_latest_signal.py @@ -2,9 +2,13 @@ from __future__ import annotations import argparse +from collections.abc import Mapping +import hashlib import json +import re import sys from pathlib import Path +from typing import Any ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "src")) @@ -13,6 +17,70 @@ DEFAULT_SIGNAL_PATH = Path("data/output/latest_signal.json") +EXPECTED_MANIFEST_TYPE = "research_signal_context" +EXPECTED_ARTIFACT_PATH = "data/output/latest_signal.json" +EXPECTED_PRODUCER_REPOSITORY = "QuantStrategyLab/ResearchSignalContextPipelines" +COMMIT_SHA_PATTERN = re.compile(r"[0-9a-fA-F]{40}") +SHA256_PATTERN = re.compile(r"[0-9a-f]{64}") + + +def _require_mapping(value: Any, name: str) -> Mapping[str, Any]: + if not isinstance(value, Mapping): + raise SignalValidationError(f"{name} must be an object") + return value + + +def _require_string(value: Any, name: str) -> str: + if not isinstance(value, str) or not value: + raise SignalValidationError(f"{name} must be a non-empty string") + return value + + +def validate_manifest_v2( + signal_path: Path, + signal: Mapping[str, Any], + manifest: Mapping[str, Any], +) -> None: + manifest = _require_mapping(manifest, "manifest") + schema_version = manifest.get("schema_version") + if schema_version in ("1", 1): + raise SignalValidationError("legacy_untrusted: manifest schema_version 1 cannot satisfy immutable provenance") + if schema_version != 2 or isinstance(schema_version, bool): + raise SignalValidationError("manifest schema_version must be 2") + if manifest.get("manifest_type") != EXPECTED_MANIFEST_TYPE: + raise SignalValidationError(f"manifest_type must be {EXPECTED_MANIFEST_TYPE!r}") + + artifact = _require_mapping(manifest.get("artifact"), "artifact") + if artifact.get("path") != EXPECTED_ARTIFACT_PATH: + raise SignalValidationError(f"artifact.path must be {EXPECTED_ARTIFACT_PATH!r}") + artifact_sha256 = _require_string(artifact.get("sha256"), "artifact.sha256") + if not SHA256_PATTERN.fullmatch(artifact_sha256): + raise SignalValidationError("artifact.sha256 must be a lowercase SHA-256 hex digest") + if artifact_sha256 != hashlib.sha256(signal_path.read_bytes()).hexdigest(): + raise SignalValidationError("artifact.sha256 does not match signal bytes") + + for field in ("as_of", "generated_at", "expires_at", "mode"): + if manifest.get(field) != signal.get(field): + raise SignalValidationError(f"{field} must exactly match the signal") + + producer = _require_mapping(manifest.get("producer"), "producer") + if producer.get("repository") != EXPECTED_PRODUCER_REPOSITORY: + raise SignalValidationError(f"producer.repository must be {EXPECTED_PRODUCER_REPOSITORY!r}") + producer_commit_sha = _require_string(producer.get("commit_sha"), "producer.commit_sha") + if not COMMIT_SHA_PATTERN.fullmatch(producer_commit_sha): + raise SignalValidationError("producer.commit_sha must be an immutable 40-hex commit") + + input_digest = _require_string(manifest.get("input_digest"), "input_digest") + if not input_digest.startswith("sha256:") or not SHA256_PATTERN.fullmatch(input_digest.removeprefix("sha256:")): + raise SignalValidationError("input_digest must be sha256:<64 lowercase hex>") + + policy = _require_mapping(manifest.get("policy"), "policy") + if policy.get("execution_allowed") is not False: + raise SignalValidationError("policy.execution_allowed must be false") + + for field in ("publication_commit", "artifact_commit", "publisher_commit"): + if field in manifest or field in producer: + raise SignalValidationError(f"{field} is not permitted in immutable provenance manifests") def main() -> int: @@ -31,10 +99,17 @@ def main() -> int: payload = json.loads(path.read_text(encoding="utf-8")) try: validate_signal(payload) + manifest_path = path.with_suffix(".manifest.json") + if manifest_path.exists(): + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + validate_manifest_v2(path, payload, manifest) except SignalValidationError as exc: - raise SystemExit(f"invalid signal artifact: {exc}") from exc + raise SystemExit(f"invalid or untrusted signal artifact: {exc}") from exc - print(f"valid signal artifact: {path}") + if path.with_suffix(".manifest.json").exists(): + print(f"valid signal artifact with immutable provenance: {path}") + else: + print(f"valid signal schema only; immutable provenance unavailable: {path}") return 0 diff --git a/tests/test_signal_validation.py b/tests/test_signal_validation.py index b2c33fe..d1e16c1 100644 --- a/tests/test_signal_validation.py +++ b/tests/test_signal_validation.py @@ -1,11 +1,14 @@ from __future__ import annotations +import hashlib import json from pathlib import Path import pytest from research_signal_context_pipelines import SignalValidationError, validate_signal +from scripts import post_shadow_signal_request as shadow_issue +from scripts import validate_latest_signal as signal_validator ROOT = Path(__file__).resolve().parents[1] @@ -137,3 +140,72 @@ def test_committed_latest_signal_covers_advisor_long_context() -> None: assert payload.get("symbol_theme_exposure") covered_symbols = set(payload.get("symbol_bias", {})) | set(payload.get("symbol_theme_exposure", {})) assert {"MU", "INTC", "AMD", "VRT", "DELL"} <= covered_symbols + + +def _write_v2_artifact_pair(tmp_path: Path) -> tuple[Path, dict]: + signal_path = tmp_path / "latest_signal.json" + signal = load_example() + signal_bytes = json.dumps(signal, sort_keys=True).encode("utf-8") + signal_path.write_bytes(signal_bytes) + manifest = { + "manifest_type": "research_signal_context", + "schema_version": 2, + "artifact": { + "path": "data/output/latest_signal.json", + "sha256": hashlib.sha256(signal_bytes).hexdigest(), + }, + "as_of": signal["as_of"], + "generated_at": signal["generated_at"], + "expires_at": signal["expires_at"], + "mode": signal["mode"], + "producer": { + "repository": "QuantStrategyLab/ResearchSignalContextPipelines", + "commit_sha": "a" * 40, + }, + "input_digest": "sha256:" + "b" * 64, + "policy": {"execution_allowed": False}, + } + return signal_path, manifest + + +def test_manifest_v2_rejects_stale_signal_digest(tmp_path: Path) -> None: + signal_path, manifest = _write_v2_artifact_pair(tmp_path) + manifest["artifact"]["sha256"] = "0" * 64 + + with pytest.raises(SignalValidationError, match="artifact.sha256"): + signal_validator.validate_manifest_v2(signal_path, load_example(), manifest) + + +def test_manifest_v2_rejects_missing_required_input_digest(tmp_path: Path) -> None: + signal_path, manifest = _write_v2_artifact_pair(tmp_path) + del manifest["input_digest"] + + with pytest.raises(SignalValidationError, match="input_digest"): + signal_validator.validate_manifest_v2(signal_path, load_example(), manifest) + + +def test_manifest_v2_rejects_mutable_producer_ref(tmp_path: Path) -> None: + signal_path, manifest = _write_v2_artifact_pair(tmp_path) + manifest["producer"]["commit_sha"] = "main" + + with pytest.raises(SignalValidationError, match="producer.commit_sha"): + signal_validator.validate_manifest_v2(signal_path, load_example(), manifest) + + +def test_manifest_v1_is_explicitly_legacy_untrusted(tmp_path: Path) -> None: + signal_path, _ = _write_v2_artifact_pair(tmp_path) + + with pytest.raises(SignalValidationError, match="legacy_untrusted"): + signal_validator.validate_manifest_v2(signal_path, load_example(), {"schema_version": "1"}) + + +def test_shadow_request_binds_context_digest_to_immutable_commit(tmp_path: Path) -> None: + context_path = tmp_path / "context.json" + context_path.write_bytes(b'{"as_of":"2026-05-29"}') + + provenance = shadow_issue.build_immutable_provenance(context_path, "c" * 40) + + assert provenance == { + "producer_commit_sha": "c" * 40, + "input_digest": "sha256:" + hashlib.sha256(context_path.read_bytes()).hexdigest(), + }