diff --git a/docs/weekly_period_bundle_contract.md b/docs/weekly_period_bundle_contract.md new file mode 100644 index 0000000..7eccbaa --- /dev/null +++ b/docs/weekly_period_bundle_contract.md @@ -0,0 +1,43 @@ +# `pert.weekly.period_bundle.v1` + +This is a pure contract library for a later trusted Actions acquisition +harness. It performs no filesystem, GitHub API, upload, download, or producer +work. The existing `pert.weekly.period_lock.v1` remains the period source of +truth. + +## Immutable models + +- `BundleContext` fixes the original `run_id`, `source_attempt=1`, exact + reviewed workflow SHA, producer SHA, expected period lock, and external + artifact evidence (`name`, `id`, digest, retention). +- `BundleWire` contains only canonical `period_lock.json`, + `input_snapshot.json`, and `bundle_manifest.json` bytes plus the external + artifact evidence. Its SHA-256 properties cover all three byte members. +- `RerunContext` accepts only current `run_attempt=2` and reuses the original + context; it requires the same `run_id`, repository, and reviewed workflow + SHA as the original context and cannot derive a new period or snapshot. + +The repository is an explicit `owner/name` binding in the manifest. Untrusted +bundle members are bounded before parsing: lock 64 KiB, snapshot 256 KiB, +manifest 64 KiB, snapshot depth 16, object keys 64, list items 256, strings +4096 characters, and source artifacts 64. Exceeding any bound fails closed. + +The canonical manifest binds the bundle version, artifact name, original run +identity, workflow and producer SHAs, lock/snapshot versions, and exact lock +and snapshot hashes. Artifact id/digest/retention are external Actions +evidence and are required to match the immutable `BundleContext`; manifest +self-hashing is intentionally avoided. + +## Verification rules + +`verify_period_bundle()` reconstructs the expected lock, snapshot, and manifest +from the complete context and requires byte-for-byte equality. It does not +accept a manifest recomputed from a replacement lock or snapshot. Duplicate, +unknown, missing, reordered, malformed, noncanonical, unsafe-integer, wrong +run/attempt/SHA/artifact, and tampered values fail closed with sanitized +`WeeklyBundleError` codes. `verify_bundle_collection()` requires exactly one +bundle. Unexpected programming/system exceptions are not broad-caught. + +This library is PR-1A only. A privileged workflow that has `actions:read` must +be a later PR sourced from the trusted default branch and must never execute +PR-controlled code. diff --git a/src/political_event_tracking_research/weekly_period_bundle.py b/src/political_event_tracking_research/weekly_period_bundle.py new file mode 100644 index 0000000..e78b93d --- /dev/null +++ b/src/political_event_tracking_research/weekly_period_bundle.py @@ -0,0 +1,430 @@ +"""Pure, exact-binding models for the trusted weekly bundle preflight. + +This module has no filesystem, GitHub, or workflow access. Privileged +artifact acquisition is intentionally a later integration boundary. +""" + +from __future__ import annotations + +import hashlib +import json +import math +import re +from collections.abc import Mapping +from dataclasses import dataclass + +from .weekly_period_lock import ( + LOCK_VERSION, + MAX_SAFE_JSON_INTEGER, + PeriodLockError, + PoliticalEventWeeklyPeriodLockV1, + SourceSnapshotArtifact, + parse_period_lock_bytes, + serialize_period_lock, +) + +BUNDLE_VERSION = "pert.weekly.period_bundle.v1" +SNAPSHOT_VERSION = "pert.weekly.input_snapshot.v1" +_RUN_ID_RE = re.compile(r"^[1-9][0-9]*$") +_SHA1_RE = re.compile(r"^[0-9a-f]{40}$") +_ARTIFACT_DIGEST_RE = re.compile(r"^sha256:[0-9a-f]{64}$") +_ARTIFACT_NAME_RE = re.compile(r"^pert-weekly-period-lock-[1-9][0-9]*$") +_REPOSITORY_RE = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$") +MAX_LOCK_BYTES = 64 * 1024 +MAX_SNAPSHOT_BYTES = 256 * 1024 +MAX_MANIFEST_BYTES = 64 * 1024 +MAX_SNAPSHOT_DEPTH = 16 +MAX_SNAPSHOT_OBJECT_KEYS = 64 +MAX_SNAPSHOT_LIST_LENGTH = 256 +MAX_SNAPSHOT_STRING_LENGTH = 4096 +MAX_SOURCE_ARTIFACTS = 64 +_SNAPSHOT_KEYS = frozenset( + { + "snapshot_version", + "source_run_id", + "source_attempt", + "workflow_sha", + "producer_sha", + "source_snapshot_id", + "source_snapshot_digest", + "source_provenance", + "source_artifacts", + } +) +_ARTIFACT_KEYS = frozenset({"path", "sha256", "row_count"}) +_MANIFEST_KEYS = frozenset( + { + "bundle_version", + "artifact_name", + "repository", + "source_run_id", + "source_attempt", + "workflow_sha", + "producer_sha", + "lock_version", + "snapshot_version", + "lock_sha256", + "snapshot_sha256", + } +) + + +class WeeklyBundleError(ValueError): + """Stable, sanitized bundle contract error.""" + + def __init__(self, code: str) -> None: + self.code = code + super().__init__(code) + + +def _error(code: str) -> WeeklyBundleError: + return WeeklyBundleError(code) + + +def _string(value: object, pattern: re.Pattern[str], code: str) -> str: + if type(value) is not str or not pattern.fullmatch(value): + raise _error(code) + return value + + +def _safe_int(value: object, code: str) -> int: + if type(value) is not int or not 0 <= value <= MAX_SAFE_JSON_INTEGER: + raise _error(code) + return value + + +def _snapshot_tree(value: object, depth: int = 0) -> object: + if depth > MAX_SNAPSHOT_DEPTH: + raise _error("bundle_snapshot_depth_exceeded") + if isinstance(value, Mapping): + if len(value) > MAX_SNAPSHOT_OBJECT_KEYS: + raise _error("bundle_snapshot_object_oversized") + result: dict[str, object] = {} + for key, item in value.items(): + if type(key) is not str: + raise _error("bundle_snapshot_invalid") + if key in result: + raise _error("bundle_snapshot_duplicate_key") + result[key] = _snapshot_tree(item, depth + 1) + return result + if type(value) is list: + if len(value) > MAX_SNAPSHOT_LIST_LENGTH: + raise _error("bundle_snapshot_list_oversized") + return [_snapshot_tree(item, depth + 1) for item in value] + if value is None or type(value) is bool: + return value + if type(value) is str: + if len(value) > MAX_SNAPSHOT_STRING_LENGTH: + raise _error("bundle_snapshot_string_oversized") + return value + if type(value) is int: + return _safe_int(value, "bundle_snapshot_invalid") + if type(value) is float and math.isfinite(value): + return value + raise _error("bundle_snapshot_invalid") + + +def _canonical_json(value: object, code: str) -> bytes: + try: + return json.dumps( + value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False + ).encode("utf-8") + except (TypeError, ValueError, UnicodeError, OverflowError, RecursionError): + raise _error(code) from None + + +def _validate_snapshot(value: object) -> dict[str, object]: + if not isinstance(value, dict) or set(value) != _SNAPSHOT_KEYS: + raise _error("bundle_snapshot_invalid") + if value["snapshot_version"] != SNAPSHOT_VERSION: + raise _error("bundle_snapshot_version_invalid") + _string(value["source_run_id"], _RUN_ID_RE, "bundle_snapshot_run_invalid") + if type(value["source_attempt"]) is not int or value["source_attempt"] != 1: + raise _error("bundle_snapshot_attempt_invalid") + _string(value["workflow_sha"], _SHA1_RE, "bundle_snapshot_workflow_invalid") + _string(value["producer_sha"], _SHA1_RE, "bundle_snapshot_producer_invalid") + _string(value["source_snapshot_id"], re.compile(r"^[a-z][a-z0-9_]*$"), "bundle_snapshot_id_invalid") + _string(value["source_snapshot_digest"], re.compile(r"^[0-9a-f]{64}$"), "bundle_snapshot_digest_invalid") + _string(value["source_provenance"], re.compile(r"^[a-z][a-z0-9_]*$"), "bundle_snapshot_provenance_invalid") + artifacts = value["source_artifacts"] + if type(artifacts) is not list or not artifacts or len(artifacts) > MAX_SOURCE_ARTIFACTS: + raise _error("bundle_snapshot_artifacts_invalid") + for item in artifacts: + if not isinstance(item, dict) or set(item) != _ARTIFACT_KEYS: + raise _error("bundle_snapshot_artifacts_invalid") + try: + SourceSnapshotArtifact(item["path"], item["sha256"], item["row_count"]) + except PeriodLockError: + raise _error("bundle_snapshot_artifacts_invalid") from None + return value + + +def _snapshot_bytes(value: Mapping[str, object]) -> bytes: + try: + tree = _snapshot_tree(value) + result = _canonical_json(_validate_snapshot(tree), "bundle_snapshot_invalid") + if len(result) > MAX_SNAPSHOT_BYTES: + raise _error("bundle_snapshot_oversized") + return result + except WeeklyBundleError: + raise + except (TypeError, ValueError, UnicodeError, OverflowError, RecursionError): + raise _error("bundle_snapshot_invalid") from None + + +def _parse_snapshot(raw: bytes) -> dict[str, object]: + if type(raw) is not bytes: + raise _error("bundle_snapshot_invalid") + if len(raw) > MAX_SNAPSHOT_BYTES: + raise _error("bundle_snapshot_oversized") + + def pairs(items: list[tuple[str, object]]) -> dict[str, object]: + result: dict[str, object] = {} + for key, item in items: + if key in result: + raise _error("bundle_snapshot_duplicate_key") + result[key] = item + return result + + def reject_constant(_: str) -> None: + raise _error("bundle_snapshot_invalid") + + try: + value = json.loads(raw.decode("utf-8"), object_pairs_hook=pairs, parse_constant=reject_constant) + except WeeklyBundleError: + raise + except (UnicodeError, json.JSONDecodeError, TypeError, ValueError, RecursionError): + raise _error("bundle_snapshot_invalid") from None + validated = _validate_snapshot(_snapshot_tree(value)) + if _canonical_json(validated, "bundle_snapshot_invalid") != raw: + raise _error("bundle_snapshot_noncanonical") + return validated + + +def _sha256(raw: bytes) -> str: + return hashlib.sha256(raw).hexdigest() + + +def _expected_snapshot(lock: PoliticalEventWeeklyPeriodLockV1, workflow_sha: str) -> dict[str, object]: + return { + "snapshot_version": SNAPSHOT_VERSION, + "source_run_id": lock.source_run_id, + "source_attempt": lock.source_attempt, + "workflow_sha": workflow_sha, + "producer_sha": lock.producer_ref, + "source_snapshot_id": lock.source_snapshot_id, + "source_snapshot_digest": lock.source_snapshot_digest, + "source_provenance": lock.source_provenance, + "source_artifacts": [ + {"path": item.path, "sha256": item.sha256, "row_count": item.row_count} for item in lock.source_artifacts + ], + } + + +@dataclass(frozen=True, slots=True) +class ArtifactEvidence: + name: str + artifact_id: int + digest: str + retention_days: int + + def __post_init__(self) -> None: + _string(self.name, _ARTIFACT_NAME_RE, "bundle_artifact_name_invalid") + _safe_int(self.artifact_id, "bundle_artifact_id_invalid") + if self.artifact_id == 0 or type(self.digest) is not str or not _ARTIFACT_DIGEST_RE.fullmatch(self.digest): + raise _error("bundle_artifact_metadata_invalid") + if type(self.retention_days) is not int or not 1 <= self.retention_days <= 90: + raise _error("bundle_retention_invalid") + + +@dataclass(frozen=True, slots=True) +class BundleContext: + run_id: str + repository: str + workflow_sha: str + producer_sha: str + artifact: ArtifactEvidence + period_lock: PoliticalEventWeeklyPeriodLockV1 + + def __post_init__(self) -> None: + _string(self.run_id, _RUN_ID_RE, "bundle_run_invalid") + _string(self.repository, _REPOSITORY_RE, "bundle_repository_invalid") + _string(self.workflow_sha, _SHA1_RE, "bundle_workflow_invalid") + _string(self.producer_sha, _SHA1_RE, "bundle_producer_invalid") + if type(self.artifact) is not ArtifactEvidence: + raise _error("bundle_artifact_invalid") + if type(self.period_lock) is not PoliticalEventWeeklyPeriodLockV1: + raise _error("bundle_lock_type_invalid") + if self.period_lock.source_run_id != self.run_id or self.period_lock.source_attempt != 1: + raise _error("bundle_lock_context_mismatch") + if self.period_lock.producer_ref != self.producer_sha: + raise _error("bundle_producer_mismatch") + if self.artifact.name != f"pert-weekly-period-lock-{self.run_id}": + raise _error("bundle_artifact_name_mismatch") + + +@dataclass(frozen=True, slots=True) +class RerunContext: + original: BundleContext + current_run_id: str + current_repository: str + current_workflow_sha: str + current_run_attempt: int + + def __post_init__(self) -> None: + if ( + type(self.original) is not BundleContext + or type(self.current_run_attempt) is not int + or self.current_run_attempt != 2 + or type(self.current_run_id) is not str + or self.current_run_id != self.original.run_id + or type(self.current_repository) is not str + or self.current_repository != self.original.repository + or type(self.current_workflow_sha) is not str + or self.current_workflow_sha != self.original.workflow_sha + ): + raise _error("bundle_rerun_attempt_invalid") + + +@dataclass(frozen=True, slots=True) +class BundleWire: + lock_bytes: bytes + snapshot_bytes: bytes + manifest_bytes: bytes + artifact: ArtifactEvidence | None = None + + @property + def bundle_version(self) -> str: + return BUNDLE_VERSION + + @property + def lock_sha256(self) -> str: + return _sha256(self.lock_bytes) + + @property + def snapshot_sha256(self) -> str: + return _sha256(self.snapshot_bytes) + + @property + def manifest_sha256(self) -> str: + return _sha256(self.manifest_bytes) + + +def _manifest_bytes(context: BundleContext, lock_bytes: bytes, snapshot_bytes: bytes) -> bytes: + return _canonical_json( + { + "bundle_version": BUNDLE_VERSION, + "artifact_name": context.artifact.name, + "repository": context.repository, + "source_run_id": context.run_id, + "source_attempt": 1, + "workflow_sha": context.workflow_sha, + "producer_sha": context.producer_sha, + "lock_version": LOCK_VERSION, + "snapshot_version": SNAPSHOT_VERSION, + "lock_sha256": _sha256(lock_bytes), + "snapshot_sha256": _sha256(snapshot_bytes), + }, + "bundle_manifest_invalid", + ) + + +def build_period_bundle( + context: BundleContext, + period_lock: PoliticalEventWeeklyPeriodLockV1, + snapshot: Mapping[str, object], +) -> BundleWire: + if type(context) is not BundleContext or type(period_lock) is not PoliticalEventWeeklyPeriodLockV1: + raise _error("bundle_input_invalid") + if period_lock != context.period_lock: + raise _error("bundle_lock_mismatch") + try: + lock_bytes = serialize_period_lock(period_lock) + except PeriodLockError: + raise _error("bundle_lock_invalid") from None + snapshot_bytes = _snapshot_bytes(snapshot) + expected_snapshot_bytes = _snapshot_bytes(_expected_snapshot(period_lock, context.workflow_sha)) + if snapshot_bytes != expected_snapshot_bytes: + raise _error("bundle_snapshot_mismatch") + return BundleWire( + lock_bytes, snapshot_bytes, _manifest_bytes(context, lock_bytes, snapshot_bytes), context.artifact + ) + + +def verify_period_bundle(bundle: BundleWire, expected: BundleContext) -> BundleWire: + if type(bundle) is not BundleWire or type(expected) is not BundleContext: + raise _error("bundle_input_invalid") + if bundle.artifact != expected.artifact: + raise _error("bundle_artifact_mismatch") + if type(bundle.lock_bytes) is not bytes or len(bundle.lock_bytes) > MAX_LOCK_BYTES: + raise _error("bundle_lock_oversized" if type(bundle.lock_bytes) is bytes else "bundle_lock_invalid") + if type(bundle.snapshot_bytes) is not bytes or len(bundle.snapshot_bytes) > MAX_SNAPSHOT_BYTES: + raise _error( + "bundle_snapshot_oversized" if type(bundle.snapshot_bytes) is bytes else "bundle_snapshot_invalid" + ) + if type(bundle.manifest_bytes) is not bytes or len(bundle.manifest_bytes) > MAX_MANIFEST_BYTES: + raise _error( + "bundle_manifest_oversized" if type(bundle.manifest_bytes) is bytes else "bundle_manifest_invalid" + ) + try: + parsed_lock = parse_period_lock_bytes(bundle.lock_bytes) + except PeriodLockError: + raise _error("bundle_lock_invalid") from None + if parsed_lock != expected.period_lock or serialize_period_lock(parsed_lock) != bundle.lock_bytes: + raise _error("bundle_lock_mismatch") + parsed_snapshot = _parse_snapshot(bundle.snapshot_bytes) + expected_snapshot_bytes = _snapshot_bytes(_expected_snapshot(expected.period_lock, expected.workflow_sha)) + if ( + bundle.snapshot_bytes != expected_snapshot_bytes + or _canonical_json(parsed_snapshot, "bundle_snapshot_invalid") != bundle.snapshot_bytes + ): + raise _error("bundle_snapshot_mismatch") + expected_manifest = _manifest_bytes(expected, bundle.lock_bytes, bundle.snapshot_bytes) + if bundle.manifest_bytes != expected_manifest: + raise _error("bundle_manifest_mismatch") + _parse_manifest(bundle.manifest_bytes) + return bundle + + +def _parse_manifest(raw: bytes) -> dict[str, object]: + if type(raw) is not bytes: + raise _error("bundle_manifest_invalid") + if len(raw) > MAX_MANIFEST_BYTES: + raise _error("bundle_manifest_oversized") + + def pairs(items: list[tuple[str, object]]) -> dict[str, object]: + result: dict[str, object] = {} + for key, item in items: + if key in result: + raise _error("bundle_manifest_duplicate_key") + result[key] = item + return result + + try: + value = json.loads(raw.decode("utf-8"), object_pairs_hook=pairs) + except WeeklyBundleError: + raise + except (UnicodeError, json.JSONDecodeError, TypeError, ValueError, RecursionError): + raise _error("bundle_manifest_invalid") from None + if not isinstance(value, dict) or set(value) != _MANIFEST_KEYS: + raise _error("bundle_manifest_invalid") + return value + + +def verify_rerun_context(bundle: BundleWire, rerun: RerunContext) -> BundleWire: + if type(rerun) is not RerunContext: + raise _error("bundle_rerun_context_invalid") + if ( + rerun.current_run_id != rerun.original.run_id + or rerun.current_repository != rerun.original.repository + or rerun.current_workflow_sha != rerun.original.workflow_sha + or rerun.current_run_attempt != 2 + ): + raise _error("bundle_rerun_attempt_invalid") + return verify_period_bundle(bundle, rerun.original) + + +def verify_bundle_collection(bundles: tuple[BundleWire, ...], expected: BundleContext) -> BundleWire: + if type(bundles) is not tuple or len(bundles) != 1: + raise _error("bundle_collection_shape_invalid") + return verify_period_bundle(bundles[0], expected) diff --git a/tests/test_weekly_period_bundle.py b/tests/test_weekly_period_bundle.py new file mode 100644 index 0000000..b7b7f4f --- /dev/null +++ b/tests/test_weekly_period_bundle.py @@ -0,0 +1,269 @@ +from __future__ import annotations + +import hashlib +import json +from datetime import date + +import pytest + +from political_event_tracking_research.weekly_period_bundle import ( + BUNDLE_VERSION, + MAX_SNAPSHOT_BYTES, + MAX_SNAPSHOT_DEPTH, + MAX_SNAPSHOT_STRING_LENGTH, + MAX_SOURCE_ARTIFACTS, + ArtifactEvidence, + BundleContext, + BundleWire, + RerunContext, + WeeklyBundleError, + build_period_bundle, + verify_bundle_collection, + verify_period_bundle, + verify_rerun_context, +) +from political_event_tracking_research.weekly_period_lock import ( + PoliticalEventWeeklyPeriodLockV1, + SourceSnapshotArtifact, + serialize_period_lock, +) + + +RUN_ID = "29420000001" +WORKFLOW_SHA = "a" * 40 +PRODUCER_SHA = "b" * 40 +ARTIFACT_DIGEST = "sha256:" + "c" * 64 + + +def lock() -> PoliticalEventWeeklyPeriodLockV1: + return PoliticalEventWeeklyPeriodLockV1( + period_start=date(2026, 7, 6), + period_end_exclusive=date(2026, 7, 13), + as_of=date(2026, 7, 12), + workflow_ref=( + "QuantStrategyLab/PoliticalEventTrackingResearch/.github/workflows/" + "source_event_pipeline.yml@refs/heads/main" + ), + source_run_id=RUN_ID, + source_attempt=1, + producer_ref=PRODUCER_SHA, + source_snapshot_id="pert_weekly_input_snapshot_20260712", + source_snapshot_digest="d" * 64, + source_provenance="official_political_event_tracking_research_v1", + source_artifacts=( + SourceSnapshotArtifact("data/live/source_events.csv", "e" * 64, 11), + SourceSnapshotArtifact("data/live/source_manifest.json", "f" * 64, 1), + ), + ) + + +def snapshot() -> dict[str, object]: + current = lock() + return { + "snapshot_version": "pert.weekly.input_snapshot.v1", + "source_run_id": RUN_ID, + "source_attempt": 1, + "workflow_sha": WORKFLOW_SHA, + "producer_sha": PRODUCER_SHA, + "source_snapshot_id": current.source_snapshot_id, + "source_snapshot_digest": current.source_snapshot_digest, + "source_provenance": current.source_provenance, + "source_artifacts": [ + {"path": item.path, "sha256": item.sha256, "row_count": item.row_count} + for item in current.source_artifacts + ], + } + + +def context(**overrides: object) -> BundleContext: + values: dict[str, object] = { + "run_id": RUN_ID, + "repository": "QuantStrategyLab/PoliticalEventTrackingResearch", + "workflow_sha": WORKFLOW_SHA, + "producer_sha": PRODUCER_SHA, + "artifact": ArtifactEvidence( + name=f"pert-weekly-period-lock-{RUN_ID}", + artifact_id=8342569267, + digest=ARTIFACT_DIGEST, + retention_days=30, + ), + "period_lock": lock(), + } + values.update(overrides) + return BundleContext(**values) + + +def test_bundle_round_trip_and_full_binding() -> None: + bundle = build_period_bundle(context(), lock(), snapshot()) + assert bundle.bundle_version == BUNDLE_VERSION + assert verify_period_bundle(bundle, context()) == bundle + assert bundle.lock_sha256 == hashlib.sha256(bundle.lock_bytes).hexdigest() + assert bundle.snapshot_sha256 == hashlib.sha256(bundle.snapshot_bytes).hexdigest() + assert bundle.manifest_sha256 == hashlib.sha256(bundle.manifest_bytes).hexdigest() + + +@pytest.mark.parametrize("member", ["lock", "snapshot", "manifest"]) +def test_tampered_member_bytes_fail_closed(member: str) -> None: + bundle = build_period_bundle(context(), lock(), snapshot()) + values = { + "lock_bytes": bundle.lock_bytes, + "snapshot_bytes": bundle.snapshot_bytes, + "manifest_bytes": bundle.manifest_bytes, + } + key = f"{member}_bytes" + values[key] = values[key] + b" " + tampered = BundleWire(**values) + with pytest.raises(WeeklyBundleError, match="bundle_.*invalid|bundle_.*mismatch"): + verify_period_bundle(tampered, context()) + + +def test_replaced_canonical_lock_fails_even_if_manifest_is_rebuilt() -> None: + original = build_period_bundle(context(), lock(), snapshot()) + changed_lock = PoliticalEventWeeklyPeriodLockV1( + date(2026, 6, 29), + date(2026, 7, 6), + date(2026, 7, 5), + lock().workflow_ref, + RUN_ID, + 1, + PRODUCER_SHA, + lock().source_snapshot_id, + lock().source_snapshot_digest, + lock().source_provenance, + lock().source_artifacts, + ) + changed = build_period_bundle(context(period_lock=changed_lock), changed_lock, snapshot()) + forged = BundleWire(changed.lock_bytes, changed.snapshot_bytes, changed.manifest_bytes, changed.artifact) + with pytest.raises(WeeklyBundleError, match="bundle_lock_mismatch"): + verify_period_bundle(forged, context()) + assert original.lock_bytes != changed.lock_bytes + + +@pytest.mark.parametrize("field", ["run_id", "workflow_sha", "producer_sha", "artifact"]) +def test_context_mismatch_fails_closed(field: str) -> None: + bundle = build_period_bundle(context(), lock(), snapshot()) + values: dict[str, object] = { + "run_id": "29420000002", + "workflow_sha": "1" * 40, + "producer_sha": "2" * 40, + "artifact": ArtifactEvidence("pert-weekly-period-lock-29420000001", 8342569268, "sha256:" + "1" * 64, 30), + } + with pytest.raises(WeeklyBundleError, match="bundle_.*mismatch"): + verify_period_bundle(bundle, context(**{field: values[field]})) + + +def test_same_run_attempt_two_context_is_explicit() -> None: + bundle = build_period_bundle(context(), lock(), snapshot()) + rerun = RerunContext( + context(), + current_run_id=RUN_ID, + current_repository="QuantStrategyLab/PoliticalEventTrackingResearch", + current_workflow_sha=WORKFLOW_SHA, + current_run_attempt=2, + ) + assert verify_rerun_context(bundle, rerun) == bundle + for field, value in ( + ("current_run_id", "29420000002"), + ("current_repository", "QuantStrategyLab/Other"), + ("current_workflow_sha", "1" * 40), + ): + with pytest.raises(WeeklyBundleError, match="bundle_rerun_attempt_invalid"): + RerunContext( + context(), + current_run_id=value if field == "current_run_id" else RUN_ID, + current_repository=value if field == "current_repository" else "QuantStrategyLab/PoliticalEventTrackingResearch", + current_workflow_sha=value if field == "current_workflow_sha" else WORKFLOW_SHA, + current_run_attempt=2, + ) + for attempt in (1, 3): + with pytest.raises(WeeklyBundleError, match="bundle_rerun_attempt_invalid"): + RerunContext( + context(), + current_run_id=RUN_ID, + current_repository="QuantStrategyLab/PoliticalEventTrackingResearch", + current_workflow_sha=WORKFLOW_SHA, + current_run_attempt=attempt, + ) + + +def test_missing_or_multiple_bundle_collection_fails_closed() -> None: + bundle = build_period_bundle(context(), lock(), snapshot()) + assert verify_bundle_collection((bundle,), context()) == bundle + for values in ((), (bundle, bundle)): + with pytest.raises(WeeklyBundleError, match="bundle_collection_shape_invalid"): + verify_bundle_collection(values, context()) + + +@pytest.mark.parametrize( + "values", + [ + ("wrong", 1, ARTIFACT_DIGEST, 30), + (f"pert-weekly-period-lock-{RUN_ID}", True, ARTIFACT_DIGEST, 30), + (f"pert-weekly-period-lock-{RUN_ID}", 1, "sha256:bad", 30), + (f"pert-weekly-period-lock-{RUN_ID}", 1, ARTIFACT_DIGEST, 0), + ], +) +def test_artifact_metadata_is_strict(values: tuple[object, ...]) -> None: + with pytest.raises(WeeklyBundleError): + BundleContext(RUN_ID, "QuantStrategyLab/PoliticalEventTrackingResearch", WORKFLOW_SHA, PRODUCER_SHA, ArtifactEvidence(*values), lock()) + + +@pytest.mark.parametrize("mutation", ["unknown", "missing", "duplicate", "unsafe_int"]) +def test_snapshot_wire_shape_is_strict(mutation: str) -> None: + base = snapshot() + if mutation == "unknown": + base["debug"] = "leak" + elif mutation == "missing": + del base["workflow_sha"] + elif mutation == "duplicate": + wire = json.dumps(base, sort_keys=True, separators=(",", ":")).replace( + '"workflow_sha":"' + WORKFLOW_SHA + '"', + '"workflow_sha":"' + WORKFLOW_SHA + '","workflow_sha":"' + WORKFLOW_SHA + '"', + ).encode() + with pytest.raises(WeeklyBundleError, match="bundle_snapshot_invalid"): + build_period_bundle(context(), lock(), wire) + return + else: + base["source_snapshot_id"] = 2**53 + with pytest.raises(WeeklyBundleError, match="bundle_snapshot_invalid"): + build_period_bundle(context(), lock(), base) + + +@pytest.mark.parametrize("kind", ["oversized_bytes", "deep", "artifacts", "string"]) +def test_snapshot_resource_bounds_fail_closed(kind: str) -> None: + bundle = build_period_bundle(context(), lock(), snapshot()) + if kind == "oversized_bytes": + wire = BundleWire( + bundle.lock_bytes, + b"{" + b"a" * MAX_SNAPSHOT_BYTES, + bundle.manifest_bytes, + bundle.artifact, + ) + else: + value = snapshot() + if kind == "deep": + nested: object = "x" + for _ in range(MAX_SNAPSHOT_DEPTH + 1): + nested = [nested] + value["source_provenance"] = nested + elif kind == "artifacts": + value["source_artifacts"] = value["source_artifacts"] * (MAX_SOURCE_ARTIFACTS + 1) + else: + value["source_snapshot_id"] = "a" * (MAX_SNAPSHOT_STRING_LENGTH + 1) + with pytest.raises(WeeklyBundleError, match="bundle_snapshot_"): + build_period_bundle(context(), lock(), value) + return + with pytest.raises(WeeklyBundleError, match="bundle_snapshot_oversized"): + verify_period_bundle(wire, context()) + + +@pytest.mark.parametrize("exception", [RuntimeError, SystemExit, MemoryError]) +def test_unexpected_exceptions_are_not_broad_caught(monkeypatch: pytest.MonkeyPatch, exception: type[BaseException]) -> None: + import political_event_tracking_research.weekly_period_bundle as module + + def fail(*_: object, **__: object) -> bytes: + raise exception("programming failure") + + monkeypatch.setattr(module.json, "dumps", fail) + with pytest.raises(exception, match="programming failure"): + build_period_bundle(context(), lock(), snapshot())