Add fixed trusted weekly workflow identity foundation

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.

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

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))