provenance.py
python
sha256:0e9549ec7b463911bc08b7d586dc320b1ac9b1f5c943ee7e3865dcc6cb0f6f83
chore(governance): sync handover+roadmap to 84db8c8 (drift:…
Human
21 hours ago
| 1 | """Provenance envelope validation and signature verification (§P0.3 / §P0.4).""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | from typing import Any |
| 6 | |
| 7 | from tools.honesty.ed25519_util import verify_ed25519_signature |
| 8 | from tools.honesty.muse_registry import MuseAgentKeyRegistry, NullMuseAgentKeyRegistry |
| 9 | from tools.honesty.types import EntryValidationError |
| 10 | |
| 11 | PROVENANCE_KEYS = frozenset({"agent_id", "model_id", "human_ref", "sig", "pubkey"}) |
| 12 | MUSE_REGIMES = frozenset({"muse+git-mirror", "muse-only"}) |
| 13 | SIGNATURE_REQUIRED_KINDS = frozenset({"verdict", "approval_recorded"}) |
| 14 | |
| 15 | |
| 16 | def _require_non_empty_str(value: Any, field: str) -> str: |
| 17 | if not isinstance(value, str) or not value.strip(): |
| 18 | raise EntryValidationError(2, f"{field} must be a non-empty string") |
| 19 | return value |
| 20 | |
| 21 | |
| 22 | def validate_provenance(raw: Any) -> dict[str, Any]: |
| 23 | """Validate and normalize a ``provenance`` object (§P0.3).""" |
| 24 | if not isinstance(raw, dict): |
| 25 | raise EntryValidationError(2, "provenance must be an object") |
| 26 | |
| 27 | extra = set(raw) - PROVENANCE_KEYS |
| 28 | if extra: |
| 29 | raise EntryValidationError(2, f"unknown provenance keys: {sorted(extra)}") |
| 30 | |
| 31 | agent_id = _require_non_empty_str(raw.get("agent_id"), "provenance.agent_id") |
| 32 | model_id = _require_non_empty_str(raw.get("model_id"), "provenance.model_id") |
| 33 | |
| 34 | human_ref = raw.get("human_ref") |
| 35 | if human_ref is not None and not isinstance(human_ref, str): |
| 36 | raise EntryValidationError(2, "provenance.human_ref must be a string or null") |
| 37 | |
| 38 | sig = raw.get("sig") |
| 39 | pubkey = raw.get("pubkey") |
| 40 | has_sig = sig is not None |
| 41 | has_pubkey = pubkey is not None |
| 42 | if has_sig != has_pubkey: |
| 43 | raise EntryValidationError(2, "provenance.sig and provenance.pubkey must both be present or both absent") |
| 44 | |
| 45 | if has_sig: |
| 46 | if not isinstance(sig, str) or not sig.startswith("ed25519:"): |
| 47 | raise EntryValidationError(2, "provenance.sig must be ed25519:<base64>") |
| 48 | if not isinstance(pubkey, str) or not pubkey.startswith("ed25519:"): |
| 49 | raise EntryValidationError(2, "provenance.pubkey must be ed25519:<base64>") |
| 50 | |
| 51 | normalized: dict[str, Any] = { |
| 52 | "agent_id": agent_id, |
| 53 | "model_id": model_id, |
| 54 | } |
| 55 | if human_ref is not None: |
| 56 | normalized["human_ref"] = human_ref |
| 57 | if has_sig: |
| 58 | normalized["sig"] = sig |
| 59 | normalized["pubkey"] = pubkey |
| 60 | return normalized |
| 61 | |
| 62 | |
| 63 | def resolve_verification_pubkey( |
| 64 | provenance: dict[str, Any], |
| 65 | *, |
| 66 | regime: str, |
| 67 | registry: MuseAgentKeyRegistry | None = None, |
| 68 | ) -> str | None: |
| 69 | """Resolve the pubkey token used to verify ``provenance.sig``.""" |
| 70 | embedded = provenance.get("pubkey") |
| 71 | if isinstance(embedded, str) and embedded.startswith("ed25519:"): |
| 72 | return embedded |
| 73 | |
| 74 | if regime in MUSE_REGIMES: |
| 75 | human_ref = provenance.get("human_ref") |
| 76 | agent_id = provenance.get("agent_id") |
| 77 | if isinstance(human_ref, str) and isinstance(agent_id, str): |
| 78 | lookup = registry if registry is not None else NullMuseAgentKeyRegistry() |
| 79 | return lookup.resolve_pubkey(human_ref=human_ref, agent_id=agent_id) |
| 80 | return None |
| 81 | |
| 82 | |
| 83 | def verify_entry_provenance( |
| 84 | entry: dict[str, Any], |
| 85 | *, |
| 86 | regime: str, |
| 87 | registry: MuseAgentKeyRegistry | None = None, |
| 88 | ) -> int: |
| 89 | """Verify optional provenance signature; return ``0`` or ``25``.""" |
| 90 | provenance = entry.get("provenance") |
| 91 | if not isinstance(provenance, dict): |
| 92 | return 0 |
| 93 | |
| 94 | sig = provenance.get("sig") |
| 95 | if not sig: |
| 96 | return 0 |
| 97 | |
| 98 | entry_hash = entry.get("entry_hash") |
| 99 | if not isinstance(entry_hash, str): |
| 100 | return 25 |
| 101 | |
| 102 | pubkey = resolve_verification_pubkey(provenance, regime=regime, registry=registry) |
| 103 | if pubkey is None: |
| 104 | return 25 |
| 105 | |
| 106 | if not verify_ed25519_signature( |
| 107 | pubkey_token=pubkey, |
| 108 | entry_hash_hex=entry_hash.lower(), |
| 109 | sig_token=sig, |
| 110 | ): |
| 111 | return 25 |
| 112 | return 0 |
| 113 | |
| 114 | |
| 115 | def signature_required_for_kind(*, require_agent_signature: bool, kind: str) -> bool: |
| 116 | """Return True when a Muse-backed config mandates ``provenance.sig``.""" |
| 117 | return require_agent_signature and kind in SIGNATURE_REQUIRED_KINDS |
| 118 | |
| 119 | |
| 120 | def provenance_has_signature(body: dict[str, Any]) -> bool: |
| 121 | """Return True when the append body carries a non-empty ``provenance.sig``.""" |
| 122 | provenance = body.get("provenance") |
| 123 | if not isinstance(provenance, dict): |
| 124 | return False |
| 125 | sig = provenance.get("sig") |
| 126 | return isinstance(sig, str) and bool(sig.strip()) |
File History
1 commit
sha256:6abcf1fa82a7a621ccbc945f19acdba5bc0db54569599404a1452fb4a096a199
fix(ISR): default require_independent_second_reviewer to require
Human
minor
⚠
21 hours ago