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..ef97ed5 --- /dev/null +++ b/src/political_event_tracking_research/trusted_workflow_identity.py @@ -0,0 +1,168 @@ +"""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) + + +@dataclass(frozen=True, slots=True, init=False) +class TrustedWorkflowIdentity: + """Immutable fixed identity; construction has no override parameters.""" + + repository: str + workflow_path: str + workflow_ref: str + + 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 TrustedWorkflowIdentity() + + +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.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) + + +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(TrustedWorkflowIdentity(), _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: + 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, + "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 + + def reject_constant(_: str) -> None: + raise _error("trusted_workflow_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("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..719428d --- /dev/null +++ b/tests/test_trusted_workflow_identity.py @@ -0,0 +1,129 @@ +from __future__ import annotations + +import json +from dataclasses import FrozenInstanceError, replace + +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, + TrustedWorkflowEvidence, + 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] + 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: + 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))