replace trusted identity with raw byte validation

Co-Authored-By: Codex <noreply@openai.com>

Author: Pigbibi

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.

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"]

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