diff --git a/docs/trusted_workflow_identity_raw_contract.md b/docs/trusted_workflow_identity_raw_contract.md new file mode 100644 index 0000000..240b78d --- /dev/null +++ b/docs/trusted_workflow_identity_raw_contract.md @@ -0,0 +1,37 @@ +# Trusted workflow identity raw-byte contract + +This fresh A0 replacement is pure and does not use Python object identity, +singleton state, capability objects, registries, workflows, permissions, +Actions, bundles, or artifact access as a trust boundary. + +## Fixed identity + +The only accepted constants are: + +- repository: `QuantStrategyLab/PoliticalEventTrackingResearch` +- workflow path: `.github/workflows/pert_weekly_period_lock_harness.yml` +- workflow ref: `QuantStrategyLab/PoliticalEventTrackingResearch/.github/workflows/pert_weekly_period_lock_harness.yml@refs/heads/main` + +`build_trusted_workflow_identity_bytes(reviewed_workflow_sha)` builds canonical +bytes from those module-private constants and an explicitly validated lowercase +40-hex runtime SHA. `validate_trusted_workflow_ref()` accepts only the fixed +workflow ref. + +## Operation boundary + +`validate_trusted_workflow_identity_bytes(raw)` bounds, parses, shape-checks, +reconstructs, canonicalizes, and compares the complete wire object on every +call. It returns only a plain reviewed-SHA value; the value is not authority +and must not be cached as a capability. Any later operation must retain and +revalidate the original canonical bytes. + +The serializer accepts only a `Mapping`, snapshots all entries, validates exact +keys and string types before canonical JSON serialization, and never reads +attributes from arbitrary objects. Duplicate, unknown, missing, noncanonical, +oversized, Unicode/control, wrong identity, and malformed SHA inputs fail with +sanitized `TrustedWorkflowIdentityError` codes. Expected contract exceptions are +sanitized; programming/system exceptions are not broadly caught. + +The later A1 bundle integration may consume this raw contract and bind it to +the period lock/snapshot/manifest. It remains blocked until this foundation is +accepted and merged; no workflow or privileged permission is part of A0. diff --git a/src/political_event_tracking_research/trusted_workflow_identity_raw.py b/src/political_event_tracking_research/trusted_workflow_identity_raw.py new file mode 100644 index 0000000..a182c1a --- /dev/null +++ b/src/political_event_tracking_research/trusted_workflow_identity_raw.py @@ -0,0 +1,143 @@ +"""Canonical raw-byte validation for the future trusted weekly harness. + +No returned Python value is an authority token. Each caller must retain and +revalidate the canonical bytes at its operation boundary. +""" + +from __future__ import annotations + +import json +import re +from collections.abc import Mapping + +IDENTITY_VERSION = "pert.trusted_workflow_identity.v1" +TRUSTED_REPOSITORY = "QuantStrategyLab/PoliticalEventTrackingResearch" +TRUSTED_WORKFLOW_PATH = ".github/workflows/pert_weekly_period_lock_harness.yml" +TRUSTED_WORKFLOW_REF = f"{TRUSTED_REPOSITORY}/{TRUSTED_WORKFLOW_PATH}@refs/heads/main" +MAX_IDENTITY_BYTES = 4096 +_SHA_RE = re.compile(r"^[0-9a-f]{40}$") +_KEYS = frozenset({"identity_version", "repository", "workflow_path", "workflow_ref", "reviewed_workflow_sha"}) + + +class TrustedWorkflowIdentityError(ValueError): + """Stable, sanitized raw identity contract error.""" + + def __init__(self, code: str) -> None: + self.code = code + super().__init__(code) + + +def _error(code: str) -> TrustedWorkflowIdentityError: + return TrustedWorkflowIdentityError(code) + + +def _reviewed_sha(value: object) -> str: + if type(value) is not str or not _SHA_RE.fullmatch(value): + raise _error("reviewed_workflow_sha_invalid") + return value + + +def validate_trusted_workflow_ref(value: object) -> str: + if type(value) is not str or value != TRUSTED_WORKFLOW_REF: + raise _error("trusted_workflow_identity_mismatch") + return TRUSTED_WORKFLOW_REF + + +def _wire_snapshot(value: object) -> dict[str, str]: + if not isinstance(value, Mapping): + raise _error("identity_wire_invalid") + try: + items = list(value.items()) + except (AttributeError, TypeError, ValueError, UnicodeError, OverflowError, RecursionError): + raise _error("identity_wire_invalid") from None + if len(items) != len(_KEYS): + raise _error("identity_shape_invalid") + result: dict[str, str] = {} + for key, item in items: + if type(key) is not str or key in result or type(item) is not str: + raise _error("identity_wire_invalid") + result[key] = item + if set(result) != _KEYS: + raise _error("identity_shape_invalid") + if result["identity_version"] != IDENTITY_VERSION: + raise _error("identity_version_invalid") + if result["repository"] != TRUSTED_REPOSITORY or any( + char in result["repository"] for char in "\x00\n\r\t" + ): + raise _error("identity_mismatch") + if result["workflow_path"] != TRUSTED_WORKFLOW_PATH: + raise _error("identity_mismatch") + validate_trusted_workflow_ref(result["workflow_ref"]) + _reviewed_sha(result["reviewed_workflow_sha"]) + return result + + +def _canonical_bytes(value: Mapping[str, str]) -> bytes: + try: + raw = json.dumps(value, ensure_ascii=True, sort_keys=True, separators=(",", ":"), allow_nan=False).encode( + "ascii" + ) + except (TypeError, ValueError, UnicodeError, OverflowError, RecursionError): + raise _error("identity_serialization_invalid") from None + if len(raw) > MAX_IDENTITY_BYTES: + raise _error("identity_wire_oversized") + return raw + + +def serialize_trusted_workflow_identity(value: object) -> bytes: + """Validate a plain mapping completely, then emit canonical bytes.""" + + return _canonical_bytes(_wire_snapshot(value)) + + +def build_trusted_workflow_identity_bytes(reviewed_workflow_sha: object) -> bytes: + """Build canonical bytes from fixed identity constants and explicit SHA.""" + + sha = _reviewed_sha(reviewed_workflow_sha) + return serialize_trusted_workflow_identity( + { + "identity_version": IDENTITY_VERSION, + "repository": TRUSTED_REPOSITORY, + "workflow_path": TRUSTED_WORKFLOW_PATH, + "workflow_ref": TRUSTED_WORKFLOW_REF, + "reviewed_workflow_sha": sha, + } + ) + + +def parse_trusted_workflow_identity_bytes(raw: object) -> dict[str, str]: + """Parse and reconstruct a plain value; never return an authority object.""" + + if type(raw) is not bytes: + raise _error("identity_wire_invalid") + if len(raw) > MAX_IDENTITY_BYTES: + raise _error("identity_wire_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("identity_duplicate_key") + result[key] = item + return result + + def reject_constant(_: str) -> None: + raise _error("identity_wire_invalid") + + try: + value = json.loads(raw.decode("ascii"), object_pairs_hook=pairs, parse_constant=reject_constant) + except TrustedWorkflowIdentityError: + raise + except (UnicodeError, json.JSONDecodeError, TypeError, ValueError, RecursionError): + raise _error("identity_wire_invalid") from None + snapshot = _wire_snapshot(value) + canonical = _canonical_bytes(snapshot) + if canonical != raw: + raise _error("identity_noncanonical") + return dict(snapshot) + + +def validate_trusted_workflow_identity_bytes(raw: object) -> str: + """Reparse canonical bytes and return only the reviewed SHA value.""" + + return parse_trusted_workflow_identity_bytes(raw)["reviewed_workflow_sha"] diff --git a/tests/test_trusted_workflow_identity_raw.py b/tests/test_trusted_workflow_identity_raw.py new file mode 100644 index 0000000..6b7f323 --- /dev/null +++ b/tests/test_trusted_workflow_identity_raw.py @@ -0,0 +1,149 @@ +from __future__ import annotations + +import json +from dataclasses import dataclass + +import pytest + +import political_event_tracking_research.trusted_workflow_identity_raw as module +from political_event_tracking_research.trusted_workflow_identity_raw import ( + IDENTITY_VERSION, + MAX_IDENTITY_BYTES, + TRUSTED_REPOSITORY, + TRUSTED_WORKFLOW_PATH, + TRUSTED_WORKFLOW_REF, + TrustedWorkflowIdentityError, + build_trusted_workflow_identity_bytes, + serialize_trusted_workflow_identity, + validate_trusted_workflow_identity_bytes, + validate_trusted_workflow_ref, +) + + +SHA = "a" * 40 + + +def wire() -> dict[str, str]: + return { + "identity_version": IDENTITY_VERSION, + "repository": TRUSTED_REPOSITORY, + "workflow_path": TRUSTED_WORKFLOW_PATH, + "workflow_ref": TRUSTED_WORKFLOW_REF, + "reviewed_workflow_sha": SHA, + } + + +def test_fixed_identity_bytes_are_deterministic_and_revalidated() -> None: + raw = build_trusted_workflow_identity_bytes(SHA) + assert validate_trusted_workflow_identity_bytes(raw) == SHA + assert serialize_trusted_workflow_identity(dict(reversed(list(wire().items())))) == raw + assert validate_trusted_workflow_ref(TRUSTED_WORKFLOW_REF) == TRUSTED_WORKFLOW_REF + + +def test_returned_plain_values_are_not_trust_objects() -> None: + value = module.parse_trusted_workflow_identity_bytes(build_trusted_workflow_identity_bytes(SHA)) + value["repository"] = "QuantStrategyLab/Other" + with pytest.raises(TrustedWorkflowIdentityError, match="identity_mismatch"): + serialize_trusted_workflow_identity(value) + + +@pytest.mark.parametrize("field", ["repository", "workflow_path", "workflow_ref"]) +def test_each_identity_field_is_rechecked(field: str) -> None: + value = wire() + value[field] = "wrong" + with pytest.raises(TrustedWorkflowIdentityError, match="identity_mismatch"): + serialize_trusted_workflow_identity(value) + + +def test_reviewed_sha_is_explicit_runtime_evidence() -> None: + value = wire() + value["reviewed_workflow_sha"] = "b" * 40 + assert serialize_trusted_workflow_identity(value) != serialize_trusted_workflow_identity(wire()) + + +@pytest.mark.parametrize("mutation", ["unknown", "missing", "duplicate", "noncanonical", "wrong_version"]) +def test_wire_shape_and_canonical_bytes_fail_closed(mutation: str) -> None: + value = wire() + if mutation == "unknown": + value["debug"] = "unexpected" # type: ignore[typeddict-unknown-key] + raw = json.dumps(value, sort_keys=True, separators=(",", ":")).encode() + elif mutation == "missing": + del value["workflow_path"] + raw = json.dumps(value, sort_keys=True, separators=(",", ":")).encode() + elif mutation == "duplicate": + raw = json.dumps(value, sort_keys=True, separators=(",", ":")).replace( + '"workflow_path":"' + TRUSTED_WORKFLOW_PATH + '"', + '"workflow_path":"' + TRUSTED_WORKFLOW_PATH + '","workflow_path":"' + TRUSTED_WORKFLOW_PATH + '"', + ).encode() + elif mutation == "noncanonical": + raw = json.dumps(value).encode() + else: + value["identity_version"] = "future" + raw = json.dumps(value, sort_keys=True, separators=(",", ":")).encode() + with pytest.raises(TrustedWorkflowIdentityError): + validate_trusted_workflow_identity_bytes(raw) + + +@pytest.mark.parametrize( + "workflow_ref", + [ + "QuantStrategyLab/Other/.github/workflows/pert_weekly_period_lock_harness.yml@refs/heads/main", + TRUSTED_WORKFLOW_REF.replace(".github/workflows/", ".github//workflows/"), + TRUSTED_WORKFLOW_REF.replace("refs/heads/main", "refs/tags/main"), + TRUSTED_WORKFLOW_REF + "\n", + "QuantStrategyLab/PoliticalEventTrackingResearch/.github/workflows/other.yml@refs/heads/main", + ], +) +def test_wrong_workflow_refs_are_sanitized(workflow_ref: str) -> None: + with pytest.raises(TrustedWorkflowIdentityError, match="identity_mismatch"): + validate_trusted_workflow_ref(workflow_ref) + + +@pytest.mark.parametrize("sha", ["a" * 39, "A" * 40, "g" * 40, True, 1, "refs/heads/main"]) +def test_reviewed_sha_requires_lowercase_full_hex(sha: object) -> None: + with pytest.raises(TrustedWorkflowIdentityError, match="reviewed_workflow_sha_invalid"): + build_trusted_workflow_identity_bytes(sha) + + +def test_oversized_and_non_ascii_wire_fail_closed() -> None: + with pytest.raises(TrustedWorkflowIdentityError, match="identity_wire_oversized"): + validate_trusted_workflow_identity_bytes(b"{" + b"a" * MAX_IDENTITY_BYTES) + value = wire() + value["repository"] = "QuantStrategyLab/PoliticalEventTrackingResearch\N{SNOWMAN}" + with pytest.raises(TrustedWorkflowIdentityError, match="identity_mismatch"): + serialize_trusted_workflow_identity(value) + + +@dataclass +class ForgedObject: + repository: str = TRUSTED_REPOSITORY + workflow_path: str = TRUSTED_WORKFLOW_PATH + workflow_ref: str = TRUSTED_WORKFLOW_REF + reviewed_workflow_sha: str = SHA + + +def test_forged_objects_and_attribute_failures_are_sanitized() -> None: + with pytest.raises(TrustedWorkflowIdentityError, match="identity_wire_invalid"): + serialize_trusted_workflow_identity(ForgedObject()) + + class BrokenMapping(dict[str, str]): + def items(self): # type: ignore[no-untyped-def] + raise AttributeError("untrusted attribute") + + with pytest.raises(TrustedWorkflowIdentityError, match="identity_wire_invalid"): + serialize_trusted_workflow_identity(BrokenMapping()) + + +def test_unexpected_runtime_error_is_not_broad_caught(monkeypatch: pytest.MonkeyPatch) -> None: + class BrokenMapping(dict[str, str]): + def items(self): # type: ignore[no-untyped-def] + raise RuntimeError("programming failure") + + with pytest.raises(RuntimeError, match="programming failure"): + serialize_trusted_workflow_identity(BrokenMapping()) + def fail_dumps(*_: object, **__: object) -> str: + raise RuntimeError("programming failure") + + monkeypatch.setattr(module.json, "dumps", fail_dumps) + with pytest.raises(RuntimeError, match="programming failure"): + serialize_trusted_workflow_identity(wire())