From 827598fde693399318cd6559286a97898aac8106 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 15 Jul 2026 21:23:21 +0800 Subject: [PATCH 1/2] add fixed trusted weekly workflow identity Co-Authored-By: Codex --- docs/trusted_workflow_identity_contract.md | 41 +++++ .../trusted_workflow_identity.py | 172 ++++++++++++++++++ tests/test_trusted_workflow_identity.py | 110 +++++++++++ 3 files changed, 323 insertions(+) create mode 100644 docs/trusted_workflow_identity_contract.md create mode 100644 src/political_event_tracking_research/trusted_workflow_identity.py create mode 100644 tests/test_trusted_workflow_identity.py diff --git a/docs/trusted_workflow_identity_contract.md b/docs/trusted_workflow_identity_contract.md new file mode 100644 index 0000000..026c72f --- /dev/null +++ b/docs/trusted_workflow_identity_contract.md @@ -0,0 +1,41 @@ +# Trusted weekly workflow identity v1 + +This is a pure PR-A0 foundation. It performs no filesystem, GitHub Actions, +artifact, workflow, producer, or rerun operation. + +## Fixed identity + +The only supported identity is compiled into the module: + +- 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` + +`trusted_workflow_identity()` is the zero-override API. Callers cannot provide +another repository, path, branch, tag, or ref as a trust root. The reviewed +workflow SHA is separate runtime evidence: it must be exactly 40 lowercase hex +characters and is bound to the fixed identity by +`validate_trusted_workflow_identity()`. + +The workflow path is a planned dedicated trusted harness path. This PR does not +create or modify that workflow and does not grant Actions permissions. + +## Wire contract + +The canonical `pert.trusted_workflow_identity.v1` wire object has exactly: + +```json +{"identity_version":"pert.trusted_workflow_identity.v1","repository":"QuantStrategyLab/PoliticalEventTrackingResearch","reviewed_workflow_sha":"<40 lowercase hex>","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"} +``` + +Unknown, missing, duplicate, noncanonical, Unicode/control, alias, wrong +repository/path/ref, and malformed SHA values fail closed with sanitized +`TrustedWorkflowIdentityError` codes. The parser and serializer use the same +fixed identity; they do not accept legacy or future versions. + +## Next boundary + +The later bundle integration must derive repository/path/ref from this typed +identity, validate `period_lock.workflow_ref` against it, and parse a manifest +structurally before comparing reconstructed canonical bytes. A privileged +workflow and `actions:read` remain outside PR-A0. diff --git a/src/political_event_tracking_research/trusted_workflow_identity.py b/src/political_event_tracking_research/trusted_workflow_identity.py new file mode 100644 index 0000000..7b32114 --- /dev/null +++ b/src/political_event_tracking_research/trusted_workflow_identity.py @@ -0,0 +1,172 @@ +"""Pure, code-reviewed identity for the future trusted weekly harness. + +This module has no workflow, filesystem, GitHub, or artifact access. Runtime +workflow SHA evidence is validated separately from the fixed repository and +workflow path/ref; callers cannot supply an alternate trust root. +""" + +from __future__ import annotations + +import json +import re +from dataclasses import dataclass + +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" +) + +_SHA_RE = re.compile(r"^[0-9a-f]{40}$") +_WIRE_KEYS = frozenset( + { + "identity_version", + "repository", + "workflow_path", + "workflow_ref", + "reviewed_workflow_sha", + } +) + + +class TrustedWorkflowIdentityError(ValueError): + """Stable, sanitized identity contract error.""" + + def __init__(self, code: str) -> None: + self.code = code + super().__init__(code) + + +def _error(code: str) -> TrustedWorkflowIdentityError: + return TrustedWorkflowIdentityError(code) + + +class TrustedWorkflowIdentity: + """Immutable fixed identity; construction is intentionally not public.""" + + __slots__ = ("_repository", "_workflow_path", "_workflow_ref") + + def __new__(cls, *_: object) -> TrustedWorkflowIdentity: + raise TypeError("use trusted_workflow_identity()") + + @classmethod + def _create(cls) -> TrustedWorkflowIdentity: + instance = object.__new__(cls) + object.__setattr__(instance, "_repository", TRUSTED_REPOSITORY) + object.__setattr__(instance, "_workflow_path", TRUSTED_WORKFLOW_PATH) + object.__setattr__(instance, "_workflow_ref", TRUSTED_WORKFLOW_REF) + return instance + + @property + def repository(self) -> str: + return self._repository + + @property + def workflow_path(self) -> str: + return self._workflow_path + + @property + def workflow_ref(self) -> str: + return self._workflow_ref + + +_TRUSTED_IDENTITY = TrustedWorkflowIdentity._create() + + +def trusted_workflow_identity() -> TrustedWorkflowIdentity: + """Return the only supported repository/workflow identity.""" + + return _TRUSTED_IDENTITY + + +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 + + +@dataclass(frozen=True, slots=True) +class TrustedWorkflowEvidence: + """Fixed identity plus independently supplied reviewed workflow SHA.""" + + identity: TrustedWorkflowIdentity + reviewed_workflow_sha: str + + def __post_init__(self) -> None: + if type(self.identity) is not TrustedWorkflowIdentity or self.identity is not _TRUSTED_IDENTITY: + raise _error("trusted_workflow_identity_mismatch") + _reviewed_sha(self.reviewed_workflow_sha) + + +def validate_trusted_workflow_identity( + workflow_ref: object, reviewed_workflow_sha: object +) -> TrustedWorkflowEvidence: + """Validate fixed workflow ref and independent runtime SHA evidence.""" + + if type(workflow_ref) is not str or workflow_ref != TRUSTED_WORKFLOW_REF: + raise _error("trusted_workflow_identity_mismatch") + return TrustedWorkflowEvidence(_TRUSTED_IDENTITY, _reviewed_sha(reviewed_workflow_sha)) + + +def _canonical_bytes(value: dict[str, object]) -> bytes: + try: + return json.dumps(value, ensure_ascii=True, sort_keys=True, separators=(",", ":"), allow_nan=False).encode( + "ascii" + ) + except (TypeError, ValueError, UnicodeError, OverflowError, RecursionError): + raise _error("trusted_workflow_serialization_invalid") from None + + +def serialize_trusted_workflow_evidence(evidence: TrustedWorkflowEvidence) -> bytes: + if type(evidence) is not TrustedWorkflowEvidence or evidence.identity is not _TRUSTED_IDENTITY: + raise _error("trusted_workflow_evidence_invalid") + return _canonical_bytes( + { + "identity_version": IDENTITY_VERSION, + "repository": TRUSTED_REPOSITORY, + "workflow_path": TRUSTED_WORKFLOW_PATH, + "workflow_ref": TRUSTED_WORKFLOW_REF, + "reviewed_workflow_sha": _reviewed_sha(evidence.reviewed_workflow_sha), + } + ) + + +def _parse_wire_value(value: object) -> TrustedWorkflowEvidence: + if type(value) is not dict or set(value) != _WIRE_KEYS: + raise _error("trusted_workflow_shape_invalid") + if value["identity_version"] != IDENTITY_VERSION: + raise _error("trusted_workflow_version_invalid") + if type(value["identity_version"]) is not str: + raise _error("trusted_workflow_version_invalid") + if value["repository"] != TRUSTED_REPOSITORY or type(value["repository"]) is not str: + raise _error("trusted_workflow_identity_mismatch") + if value["workflow_path"] != TRUSTED_WORKFLOW_PATH or type(value["workflow_path"]) is not str: + raise _error("trusted_workflow_identity_mismatch") + if value["workflow_ref"] != TRUSTED_WORKFLOW_REF or type(value["workflow_ref"]) is not str: + raise _error("trusted_workflow_identity_mismatch") + return validate_trusted_workflow_identity(value["workflow_ref"], value["reviewed_workflow_sha"]) + + +def parse_trusted_workflow_evidence(raw: bytes) -> TrustedWorkflowEvidence: + if type(raw) is not bytes: + raise _error("trusted_workflow_wire_invalid") + + 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("trusted_workflow_duplicate_key") + result[key] = item + return result + + try: + value = json.loads(raw.decode("ascii"), object_pairs_hook=pairs, parse_constant=lambda _: (_ for _ in ()).throw(_error("trusted_workflow_wire_invalid"))) + except TrustedWorkflowIdentityError: + raise + except (UnicodeError, json.JSONDecodeError, TypeError, ValueError, RecursionError): + raise _error("trusted_workflow_wire_invalid") from None + evidence = _parse_wire_value(value) + if serialize_trusted_workflow_evidence(evidence) != raw: + raise _error("trusted_workflow_noncanonical") + return evidence diff --git a/tests/test_trusted_workflow_identity.py b/tests/test_trusted_workflow_identity.py new file mode 100644 index 0000000..9b37231 --- /dev/null +++ b/tests/test_trusted_workflow_identity.py @@ -0,0 +1,110 @@ +from __future__ import annotations + +import json + +import pytest + +import political_event_tracking_research.trusted_workflow_identity as module +from political_event_tracking_research.trusted_workflow_identity import ( + IDENTITY_VERSION, + TRUSTED_REPOSITORY, + TRUSTED_WORKFLOW_PATH, + TRUSTED_WORKFLOW_REF, + TrustedWorkflowIdentityError, + parse_trusted_workflow_evidence, + serialize_trusted_workflow_evidence, + trusted_workflow_identity, + validate_trusted_workflow_identity, +) + + +SHA = "a" * 40 + + +def valid_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_has_no_runtime_override() -> None: + identity = trusted_workflow_identity() + assert identity.repository == TRUSTED_REPOSITORY + assert identity.workflow_path == TRUSTED_WORKFLOW_PATH + assert identity.workflow_ref == TRUSTED_WORKFLOW_REF + with pytest.raises(TypeError): + module.TrustedWorkflowIdentity(TRUSTED_REPOSITORY, "other.yml", TRUSTED_WORKFLOW_REF) # type: ignore[call-arg] + + +def test_runtime_sha_and_fixed_workflow_identity_round_trip() -> None: + evidence = validate_trusted_workflow_identity(TRUSTED_WORKFLOW_REF, SHA) + wire = serialize_trusted_workflow_evidence(evidence) + assert parse_trusted_workflow_evidence(wire) == evidence + assert wire == serialize_trusted_workflow_evidence(parse_trusted_workflow_evidence(wire)) + + +@pytest.mark.parametrize( + "workflow_ref", + [ + "QuantStrategyLab/Other/.github/workflows/pert_weekly_period_lock_harness.yml@refs/heads/main", + "QuantStrategyLab/PoliticalEventTrackingResearch/.github/workflows/other.yml@refs/heads/main", + TRUSTED_WORKFLOW_REF.replace("refs/heads/main", "refs/tags/main"), + TRUSTED_WORKFLOW_REF.replace("PoliticalEventTrackingResearch", "politicaleventtrackingresearch"), + TRUSTED_WORKFLOW_REF.replace(".github/workflows/", ".github//workflows/"), + TRUSTED_WORKFLOW_REF + "\n", + SHA, + ], +) +def test_wrong_workflow_identity_is_rejected(workflow_ref: str) -> None: + with pytest.raises(TrustedWorkflowIdentityError, match="trusted_workflow_identity_mismatch"): + validate_trusted_workflow_identity(workflow_ref, SHA) + + +@pytest.mark.parametrize("sha", ["a" * 39, "A" * 40, "g" * 40, "refs/heads/main", True, 1]) +def test_reviewed_sha_is_strict_full_hex(sha: object) -> None: + with pytest.raises(TrustedWorkflowIdentityError, match="reviewed_workflow_sha_invalid"): + validate_trusted_workflow_identity(TRUSTED_WORKFLOW_REF, sha) + + +@pytest.mark.parametrize("mutation", ["unknown", "missing", "wrong_repo", "wrong_path", "wrong_ref", "wrong_sha"]) +def test_wire_is_exact_and_sanitized(mutation: str) -> None: + wire = valid_wire() + if mutation == "unknown": + wire["debug"] = "secret" + elif mutation == "missing": + del wire["workflow_path"] + elif mutation == "wrong_repo": + wire["repository"] = "QuantStrategyLab/Other" + elif mutation == "wrong_path": + wire["workflow_path"] = "other.yml" + elif mutation == "wrong_ref": + wire["workflow_ref"] = TRUSTED_WORKFLOW_REF.replace("main", "feature") + else: + wire["reviewed_workflow_sha"] = "b" * 40 + with pytest.raises(TrustedWorkflowIdentityError): + parse_trusted_workflow_evidence(json.dumps(wire).encode()) + + +def test_duplicate_and_noncanonical_wire_fail_closed() -> None: + wire = json.dumps(valid_wire(), sort_keys=True, separators=(",", ":")) + duplicate = wire.replace( + '"workflow_path":"' + TRUSTED_WORKFLOW_PATH + '"', + '"workflow_path":"' + TRUSTED_WORKFLOW_PATH + '","workflow_path":"' + TRUSTED_WORKFLOW_PATH + '"', + ) + with pytest.raises(TrustedWorkflowIdentityError, match="trusted_workflow_duplicate_key"): + parse_trusted_workflow_evidence(duplicate.encode()) + with pytest.raises(TrustedWorkflowIdentityError, match="trusted_workflow_noncanonical"): + parse_trusted_workflow_evidence(json.dumps(valid_wire()).encode()) + + +def test_unexpected_runtime_error_is_not_broad_caught(monkeypatch: pytest.MonkeyPatch) -> None: + def fail(*_: object, **__: object) -> bytes: + raise RuntimeError("programming failure") + + monkeypatch.setattr(module.json, "dumps", fail) + with pytest.raises(RuntimeError, match="programming failure"): + serialize_trusted_workflow_evidence(validate_trusted_workflow_identity(TRUSTED_WORKFLOW_REF, SHA)) From a1afd72571beaeb91b303cffd05d3b3e23d93e30 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 15 Jul 2026 21:38:30 +0800 Subject: [PATCH 2/2] harden trusted identity invariants Co-Authored-By: Codex --- .../trusted_workflow_identity.py | 60 +++++++++---------- tests/test_trusted_workflow_identity.py | 19 ++++++ 2 files changed, 47 insertions(+), 32 deletions(-) diff --git a/src/political_event_tracking_research/trusted_workflow_identity.py b/src/political_event_tracking_research/trusted_workflow_identity.py index 7b32114..ef97ed5 100644 --- a/src/political_event_tracking_research/trusted_workflow_identity.py +++ b/src/political_event_tracking_research/trusted_workflow_identity.py @@ -42,42 +42,24 @@ def _error(code: str) -> TrustedWorkflowIdentityError: return TrustedWorkflowIdentityError(code) +@dataclass(frozen=True, slots=True, init=False) class TrustedWorkflowIdentity: - """Immutable fixed identity; construction is intentionally not public.""" + """Immutable fixed identity; construction has no override parameters.""" - __slots__ = ("_repository", "_workflow_path", "_workflow_ref") + repository: str + workflow_path: str + workflow_ref: str - def __new__(cls, *_: object) -> TrustedWorkflowIdentity: - raise TypeError("use trusted_workflow_identity()") - - @classmethod - def _create(cls) -> TrustedWorkflowIdentity: - instance = object.__new__(cls) - object.__setattr__(instance, "_repository", TRUSTED_REPOSITORY) - object.__setattr__(instance, "_workflow_path", TRUSTED_WORKFLOW_PATH) - object.__setattr__(instance, "_workflow_ref", TRUSTED_WORKFLOW_REF) - return instance - - @property - def repository(self) -> str: - return self._repository - - @property - def workflow_path(self) -> str: - return self._workflow_path - - @property - def workflow_ref(self) -> str: - return self._workflow_ref - - -_TRUSTED_IDENTITY = TrustedWorkflowIdentity._create() + def __init__(self) -> None: + object.__setattr__(self, "repository", TRUSTED_REPOSITORY) + object.__setattr__(self, "workflow_path", TRUSTED_WORKFLOW_PATH) + object.__setattr__(self, "workflow_ref", TRUSTED_WORKFLOW_REF) def trusted_workflow_identity() -> TrustedWorkflowIdentity: """Return the only supported repository/workflow identity.""" - return _TRUSTED_IDENTITY + return TrustedWorkflowIdentity() def _reviewed_sha(value: object) -> str: @@ -94,7 +76,12 @@ class TrustedWorkflowEvidence: reviewed_workflow_sha: str def __post_init__(self) -> None: - if type(self.identity) is not TrustedWorkflowIdentity or self.identity is not _TRUSTED_IDENTITY: + if ( + type(self.identity) is not TrustedWorkflowIdentity + or self.identity.repository != TRUSTED_REPOSITORY + or self.identity.workflow_path != TRUSTED_WORKFLOW_PATH + or self.identity.workflow_ref != TRUSTED_WORKFLOW_REF + ): raise _error("trusted_workflow_identity_mismatch") _reviewed_sha(self.reviewed_workflow_sha) @@ -106,7 +93,7 @@ def validate_trusted_workflow_identity( if type(workflow_ref) is not str or workflow_ref != TRUSTED_WORKFLOW_REF: raise _error("trusted_workflow_identity_mismatch") - return TrustedWorkflowEvidence(_TRUSTED_IDENTITY, _reviewed_sha(reviewed_workflow_sha)) + return TrustedWorkflowEvidence(TrustedWorkflowIdentity(), _reviewed_sha(reviewed_workflow_sha)) def _canonical_bytes(value: dict[str, object]) -> bytes: @@ -119,8 +106,14 @@ def _canonical_bytes(value: dict[str, object]) -> bytes: def serialize_trusted_workflow_evidence(evidence: TrustedWorkflowEvidence) -> bytes: - if type(evidence) is not TrustedWorkflowEvidence or evidence.identity is not _TRUSTED_IDENTITY: + if type(evidence) is not TrustedWorkflowEvidence: raise _error("trusted_workflow_evidence_invalid") + if ( + evidence.identity.repository != TRUSTED_REPOSITORY + or evidence.identity.workflow_path != TRUSTED_WORKFLOW_PATH + or evidence.identity.workflow_ref != TRUSTED_WORKFLOW_REF + ): + raise _error("trusted_workflow_identity_mismatch") return _canonical_bytes( { "identity_version": IDENTITY_VERSION, @@ -160,8 +153,11 @@ def pairs(items: list[tuple[str, object]]) -> dict[str, object]: result[key] = item return result + def reject_constant(_: str) -> None: + raise _error("trusted_workflow_wire_invalid") + try: - value = json.loads(raw.decode("ascii"), object_pairs_hook=pairs, parse_constant=lambda _: (_ for _ in ()).throw(_error("trusted_workflow_wire_invalid"))) + value = json.loads(raw.decode("ascii"), object_pairs_hook=pairs, parse_constant=reject_constant) except TrustedWorkflowIdentityError: raise except (UnicodeError, json.JSONDecodeError, TypeError, ValueError, RecursionError): diff --git a/tests/test_trusted_workflow_identity.py b/tests/test_trusted_workflow_identity.py index 9b37231..719428d 100644 --- a/tests/test_trusted_workflow_identity.py +++ b/tests/test_trusted_workflow_identity.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +from dataclasses import FrozenInstanceError, replace import pytest @@ -11,6 +12,7 @@ TRUSTED_WORKFLOW_PATH, TRUSTED_WORKFLOW_REF, TrustedWorkflowIdentityError, + TrustedWorkflowEvidence, parse_trusted_workflow_evidence, serialize_trusted_workflow_evidence, trusted_workflow_identity, @@ -38,6 +40,23 @@ def test_fixed_identity_has_no_runtime_override() -> None: assert identity.workflow_ref == TRUSTED_WORKFLOW_REF with pytest.raises(TypeError): module.TrustedWorkflowIdentity(TRUSTED_REPOSITORY, "other.yml", TRUSTED_WORKFLOW_REF) # type: ignore[call-arg] + with pytest.raises(FrozenInstanceError): + identity.repository = "QuantStrategyLab/Other" # type: ignore[misc] + assert trusted_workflow_identity().repository == TRUSTED_REPOSITORY + + +def test_evidence_constructor_and_replace_revalidate_invariants() -> None: + evidence = validate_trusted_workflow_identity(TRUSTED_WORKFLOW_REF, SHA) + assert replace(evidence, reviewed_workflow_sha="b" * 40).reviewed_workflow_sha == "b" * 40 + with pytest.raises(TrustedWorkflowIdentityError, match="reviewed_workflow_sha_invalid"): + replace(evidence, reviewed_workflow_sha="not-a-sha") + + forged_identity = object.__new__(module.TrustedWorkflowIdentity) + object.__setattr__(forged_identity, "repository", "QuantStrategyLab/Other") + object.__setattr__(forged_identity, "workflow_path", TRUSTED_WORKFLOW_PATH) + object.__setattr__(forged_identity, "workflow_ref", TRUSTED_WORKFLOW_REF) + with pytest.raises(TrustedWorkflowIdentityError, match="trusted_workflow_identity_mismatch"): + TrustedWorkflowEvidence(forged_identity, SHA) def test_runtime_sha_and_fixed_workflow_identity_round_trip() -> None: