"""Generic attestation provider framework — Phase 7.1. A small extension point that lets domain plugins anchor commit provenance in *external trust roots* — HMAC AIR endpoints, blockchain canisters, TSA timestamping authorities, etc. — without coupling the Muse engine to any specific provider implementation. The first consumer is the Knowtation domain (see :mod:`muse.plugins.knowtation.attestation`), which implements an HMAC AIR provider that is functionally equivalent to the JS ``knowtation/lib/air.mjs`` module. The second is the ICP push hook (Phase 7.4) that anchors the attestation record on the live Knowtation attestation canister (``dejku-syaaa-aaaaa-qgy3q-cai``). Design contract — security-critical ----------------------------------- 1. **Misuse-resistance: partial verifications are hard errors.** :meth:`MuseAttestationProvider.verify` MUST raise :class:`AttestationVerifyError` whenever it cannot complete the verification chain (e.g. ICP unreachable, signature parse failure, missing key material). Returning ``VerifyResult(valid=False)`` is reserved for the case of a *known-bad* record where the verifier was able to complete the check. Silent ``True`` returns are *never* permitted. 2. **Determinism.** :meth:`MuseAttestationProvider.compute_attestation` MUST be deterministic for the same ``(snapshot_id, commit_meta)`` pair — no random salts in the interface, no wall-clock timestamps in the ID. This enables idempotent re-attestation: replaying a commit on a fresh node produces the same attestation ID, so the canister will reject the second write with ``"already exists"`` and the system stays in a consistent state. 3. **Concurrency safety.** :meth:`MuseAttestationProvider.anchor` and :meth:`MuseAttestationProvider.verify` MUST be safe to call from concurrent threads without sharing mutable state. The registry itself is serialised by an internal :class:`threading.Lock` so concurrent registrations cannot lose entries. 4. **Write-once receipts.** :class:`AnchorReceipt.tx_id` is write-once after :meth:`MuseAttestationProvider.anchor` returns. This matches the live ICP canister design — the attestation ledger is immutable and append-only; re-anchoring the same record raises :class:`ImmutableReceiptError` rather than silently mutating state. 5. **No silent exception swallowing.** Errors at the framework layer are propagated or wrapped in typed errors (:class:`AttestationError` and subclasses). Provider implementations are free to convert their transport-level errors into these typed errors but MUST NOT log-and-ignore. Threat model documented inline ------------------------------ * **Replay across branches** — :func:`muse.core.provenance.provenance_payload` binds the commit_id, so the same payload cannot be re-used for a different commit. Different commits have different commit_ids by construction. * **Anchor forgery** — the canister's ``storeAttestation`` requires an authorised principal; unauthorised calls are rejected at the canister boundary. The :meth:`MuseAttestationProvider.verify` method MUST cross-check the receipt against the canister's HTTP query interface to rule out spoofed receipts. * **Identity-chain depth manipulation** — :class:`VerifyResult.depth` records how many identity-record hops were validated during verification (e.g. for org commits with quorum signing — Phase 7.7). A low depth on a high-stakes commit is an audit signal but not a verification failure on its own. """ from __future__ import annotations import threading from typing import Protocol, TypedDict, runtime_checkable # --------------------------------------------------------------------------- # Typed records — flat TypedDicts are JSON-serialisable for storage in # CommitRecord.metadata and for transport over the MCP/CLI boundary. # --------------------------------------------------------------------------- class AttestationRecord(TypedDict, total=False): """A single attestation record produced by a provider. Stored under ``commit.metadata["attestation"]`` after a successful ``muse commit --attest``. All fields are strings to keep the record JSON-clean and forward-compatible across provider implementations. Attributes: id: Globally-unique attestation identifier — typically ``SHA-256(snapshot_id + "\\x00" + path)`` so the ID is deterministic and content-addressed. Must be stable across provider invocations with the same inputs. action: The action being attested (``"write"``, ``"export"``, ``"merge"``, …). Provider-defined; the Knowtation HMAC provider uses the same vocabulary as ``knowtation/lib/air.mjs``. path: The vault-relative or repo-relative path being attested. Empty string when the attestation is repo-scoped rather than file-scoped. timestamp: ISO-8601 UTC timestamp of attestation creation. Bound into the signature so back-dating cannot alter verification. content_hash: SHA-256 hex of the content being attested. Lower-case, 64 hex characters. sig: Provider-specific signature/MAC over the canonical record bytes. Format and verification semantics are owned by the provider. algorithm: Stable identifier for the signature algorithm (e.g. ``"hmac-sha256"``, ``"ed25519"``). Required so verifiers can reject records using a deprecated algorithm. provider_id: The domain string the provider is registered under (e.g. ``"knowtation"``). Required so verifiers can dispatch to the correct provider on disconnected systems. """ id: str action: str path: str timestamp: str content_hash: str sig: str algorithm: str provider_id: str class AnchorReceipt(TypedDict, total=False): """Receipt returned by :meth:`MuseAttestationProvider.anchor`. Embeds the off-chain transaction identifier and a public anchor URL that third parties can use to independently verify the record without contacting the original signer. Attributes: tx_id: Transaction identifier from the anchor (e.g. ICP canister seq number, blockchain tx hash). Write-once: re-anchoring the same record MUST raise :class:`ImmutableReceiptError`, not produce a second receipt. anchor_url: HTTPS URL where the record can be independently fetched (e.g. ``https://.ic0.app/attest/``). MUST be HTTPS — verifiers MUST reject non-HTTPS anchor URLs. anchored_at: ISO-8601 UTC timestamp of when the anchor was written. For audit trails; not bound into the signature. provider_id: Same domain string as the corresponding :class:`AttestationRecord.provider_id`. Required because a single repo may hold receipts from multiple providers. """ tx_id: str anchor_url: str anchored_at: str provider_id: str class VerifyResult(TypedDict, total=False): """Result of a complete verification check by a provider. A complete verification check is one where the provider was able to validate every link in the chain (signature, anchor receipt, identity record). When the provider cannot complete the chain — for any reason that is not "the record is known-bad" — it MUST raise :class:`AttestationVerifyError` instead of returning this dict. Attributes: valid: ``True`` iff every link was validated and matched. Any value of ``False`` means the provider proved the record bad — it does *not* mean "could not check". ``False``-results MUST be accompanied by a non-empty :attr:`reason`. reason: Human-readable diagnostic. Empty when ``valid=True``. When ``valid=False``, must explain *which* link broke (e.g. ``"hmac signature mismatch"``, ``"anchor URL resolves to different content_hash"``). Logged in audit trails. depth: Identity-chain depth validated. ``0`` for a single-signer commit, ``N`` for an N-deep org-quorum commit. This is an audit signal, not a flow-control flag. provider_id: Same domain string used to register the provider. """ valid: bool reason: str depth: int provider_id: str # --------------------------------------------------------------------------- # Typed error hierarchy — verify() must raise these, never silent True. # --------------------------------------------------------------------------- class AttestationError(Exception): """Base exception for the attestation framework. All framework- and provider-level errors derive from this class so callers can catch the entire family with one ``except AttestationError`` handler. """ class AttestationRequiredError(AttestationError): """Raised when ``air.required=True`` and attestation could not be obtained. Functionally equivalent to the JS ``AttestationRequiredError`` defined in ``knowtation/lib/air.mjs``. Callers (typically the ``muse commit --attest`` code path) MUST surface this as a hard write rejection — the commit must not be persisted to the ref. Attributes: code: Stable error code string ``"ATTESTATION_REQUIRED"``. Used by CLI/MCP layers to map back to a structured error result without string-matching the message. """ code: str = "ATTESTATION_REQUIRED" class AttestationVerifyError(AttestationError): """Raised when :meth:`MuseAttestationProvider.verify` cannot complete. Distinct from a returned ``VerifyResult(valid=False)``: this exception means the verifier was unable to *prove* the record valid or invalid — e.g. the canister was unreachable, the signature could not be parsed, a required key was missing. Callers MUST treat this as "verification status unknown", never as "valid". Attributes: code: Stable error code string for programmatic handling. Defaults to ``"VERIFY_INCOMPLETE"`` but providers may set a more specific code (e.g. ``"ICP_UNREACHABLE"``, ``"KEY_MISSING"``). """ code: str = "VERIFY_INCOMPLETE" def __init__(self, message: str, *, code: str | None = None) -> None: """Construct a verify error with an optional override code. Args: message: Human-readable description of the incomplete verification. code: Optional override for the default ``"VERIFY_INCOMPLETE"`` code. Should be a short uppercase token (e.g. ``"ICP_UNREACHABLE"``). """ super().__init__(message) if code is not None: self.code = code class ImmutableReceiptError(AttestationError): """Raised when re-anchoring a record that already has a receipt. Receipts are write-once by contract — see the canister's ``storeAttestation`` which rejects duplicate IDs with ``#err("Attestation X already exists (records are immutable)")``. The Python framework mirrors that semantics so application code sees the same error regardless of where it was raised. Attributes: code: Stable error code string ``"IMMUTABLE_RECEIPT"``. """ code: str = "IMMUTABLE_RECEIPT" # --------------------------------------------------------------------------- # Provider protocol — runtime_checkable so isinstance() works for tests. # --------------------------------------------------------------------------- @runtime_checkable class MuseAttestationProvider(Protocol): """Structural type for an attestation provider. A provider is any object that exposes the three methods below. Domain plugins typically expose a class implementing this protocol in ``muse/plugins//attestation.py`` and register a singleton via :func:`register_attestation_provider`. Implementations MUST: * Be safe to call from any thread (no shared mutable state between :meth:`anchor` and :meth:`verify`). * Make :meth:`compute_attestation` deterministic for the same input. * Raise :class:`AttestationVerifyError` from :meth:`verify` whenever verification cannot be completed; never return ``valid=True`` on partial-verification. * Raise :class:`ImmutableReceiptError` when :meth:`anchor` is called with a record whose ID is already anchored. """ def compute_attestation( self, snapshot_id: str, commit_meta: dict[str, object], ) -> AttestationRecord: """Compute a deterministic attestation record for a commit. Called from ``muse commit --attest`` after the commit object has been constructed but before it is written to ``.muse/refs/heads/``. Args: snapshot_id: SHA-256 hex of the snapshot being attested. Bound into the deterministic record ID so the same snapshot always produces the same attestation. commit_meta: A free-form dict of commit metadata. Provider implementations should treat it as opaque except for keys they explicitly document (e.g. the Knowtation HMAC provider reads ``"path"`` and ``"action"``). Returns: A :class:`AttestationRecord` with all required fields set. The record is *not* yet anchored — call :meth:`anchor` to write it to the external trust root. Raises: AttestationRequiredError: If a required configuration field is missing and the provider's policy requires attestation. """ ... def anchor(self, record: AttestationRecord) -> AnchorReceipt: """Anchor *record* in the provider's external trust root. Typically calls a network endpoint (HTTP POST to AIR daemon, ICP canister update call, blockchain transaction). May spawn a background thread for long-running anchors, in which case the returned receipt's ``tx_id`` may be a placeholder until the background work completes — see :class:`AnchorReceipt` semantics in each provider's docstring. Args: record: A :class:`AttestationRecord` from :meth:`compute_attestation`. Must include all required fields; providers MAY validate this and raise :class:`AttestationError` on malformed input. Returns: An :class:`AnchorReceipt` containing the off-chain transaction ID and a public anchor URL. Raises: ImmutableReceiptError: If a receipt already exists for ``record["id"]`` (write-once semantics). AttestationError: For any other anchoring failure. """ ... def verify( self, record: AttestationRecord, receipt: AnchorReceipt, ) -> VerifyResult: """Verify that *record* and *receipt* are valid and consistent. Performs the full verification chain: 1. Re-compute the canonical bytes of *record* and check the signature against the provider's known signing key. 2. Fetch the anchored copy from ``receipt["anchor_url"]`` and check that it matches *record* (same ID, same content_hash, same path). 3. (Org commits — Phase 7.7) Walk the identity chain and validate each member's signature; record the depth in the result. Args: record: The :class:`AttestationRecord` to verify. receipt: The :class:`AnchorReceipt` returned by :meth:`anchor`. Returns: A :class:`VerifyResult` with ``valid=True`` iff every link in the chain validated. ``valid=False`` iff the verifier proved the record bad — never returned for "could not check". Raises: AttestationVerifyError: If the verifier cannot complete the check (network unreachable, key missing, malformed record, etc.). Distinct from a returned ``valid=False``, which means the verifier *proved* the record bad. """ ... # --------------------------------------------------------------------------- # Registry — same threading pattern as muse/core/merge_hooks.py # --------------------------------------------------------------------------- _REGISTRY_LOCK: threading.Lock = threading.Lock() _PROVIDERS: dict[str, MuseAttestationProvider] = {} def register_attestation_provider( domain: str, provider: MuseAttestationProvider, ) -> None: """Register *provider* under *domain* in the process-wide registry. Idempotent — calling this function more than once with the same domain replaces the previous registration. Safe to call eagerly at import time (e.g. from ``muse/plugins//__init__.py``) as well as lazily on first need. Concurrency: Serialised by a module-level :class:`threading.Lock` so concurrent registrations from different threads cannot lose entries. Domain isolation: Domains are isolated — :func:`get_attestation_provider` for domain A will never return the provider registered for domain B, even if domain A has no registered provider. This guarantees that a misconfigured commit cannot accidentally invoke a provider for an unrelated domain. Args: domain: Non-empty domain name string (e.g. ``"knowtation"``). Case-sensitive. provider: An object implementing the :class:`MuseAttestationProvider` protocol. Validated structurally via :func:`isinstance` against the runtime-checkable protocol — duck-typing accepted because only the three public methods are ever called. Raises: ValueError: If *domain* is empty or not a string. TypeError: If *provider* does not conform to :class:`MuseAttestationProvider`. """ if not isinstance(domain, str) or not domain: raise ValueError("domain must be a non-empty string") if not isinstance(provider, MuseAttestationProvider): raise TypeError( f"provider must implement MuseAttestationProvider " f"(got {type(provider).__name__})" ) with _REGISTRY_LOCK: _PROVIDERS[domain] = provider def get_attestation_provider(domain: str) -> MuseAttestationProvider | None: """Return the registered provider for *domain* or ``None``. Never raises. When *domain* has no registered provider, returns ``None`` so the caller can decide whether to treat absence as a soft skip (``muse commit --attest`` when no provider is registered) or a hard failure (``muse verify-commit --attest --required``). Args: domain: Domain name string. Returns: The :class:`MuseAttestationProvider` registered for *domain*, or ``None`` when no provider has been registered. Domain isolation is enforced — this function never falls back to a provider registered under a different domain. """ with _REGISTRY_LOCK: return _PROVIDERS.get(domain) def unregister_attestation_provider(domain: str) -> bool: """Remove the provider registered under *domain*, if any. Useful in tests that need to restore a clean registry state between cases. Production code should not need to call this. Args: domain: Domain name string. Returns: ``True`` if a provider was removed, ``False`` when no provider was registered for *domain*. """ with _REGISTRY_LOCK: return _PROVIDERS.pop(domain, None) is not None def _registered_domains() -> list[str]: """Return a snapshot of registered domains, sorted for determinism. Internal helper for tests and diagnostic CLIs. Not part of the public API — production callers should use :func:`get_attestation_provider` to look up by name. """ with _REGISTRY_LOCK: return sorted(_PROVIDERS.keys()) # --------------------------------------------------------------------------- # Phase 7.7 — Identity / quorum types and provider Protocol # --------------------------------------------------------------------------- class IdentityRecord(TypedDict, total=False): """A single identity record from the identity domain. Identity records describe agents, humans, and orgs. Org records carry a ``quorum_threshold`` and a list of member ``handles`` so that downstream quorum verification can fan out and check member signatures. Attributes: handle: Stable identity handle. Org handles MUST start with ``@``; single-actor handles MUST NOT. Case-sensitive. kind: One of ``"human"``, ``"agent"``, ``"org"``. Verifiers MUST reject quorum signing for any handle whose ``kind != "org"``. public_key: Base64url-encoded raw Ed25519 public key (32 bytes → 43 chars no padding). Empty for org records (orgs have no standalone signing key — they're verified through their members). members: For org records — list of member handles. Each member MUST itself resolve to an :class:`IdentityRecord` (recursive lookup, depth-bounded by the verifier). quorum_threshold: For org records — minimum number of valid member signatures required to consider the quorum satisfied. MUST be an integer in ``[1, len(members)]``. spawned_by: Optional handle of the identity that spawned this one (for audit trails of the SPAWNS edge in the identity graph). revoked_at: Optional ISO-8601 timestamp when the identity was revoked. Verifiers MUST treat any non-empty ``revoked_at`` as "identity invalid" — signatures by revoked identities do not count toward quorum. """ handle: str kind: str public_key: str members: list[str] quorum_threshold: int spawned_by: str revoked_at: str class IdentityChainResult(TypedDict, total=False): """Result of walking an identity chain to validate a handle. Returned by :func:`muse.core.verify.verify_identity_chain` and surfaced in :class:`QuorumVerifyResult.depth`. Attributes: valid: ``True`` iff every node in the walked chain resolved and was non-revoked. ``False`` for any verifiable failure. depth: Number of identity-graph hops walked (1 for a direct lookup, N for an N-deep org-of-orgs chain). Zero means the root handle could not even be resolved. reason: Human-readable diagnostic. Empty when ``valid=True``. cycle_detected: ``True`` iff a cycle was detected during the walk (member A → member B → member A). Cycles are a security failure — the identity domain rejects them at write time, but the verifier re-checks defensively in case a malicious actor was able to inject a cycle into a stale local cache. """ valid: bool depth: int reason: str cycle_detected: bool @runtime_checkable class MuseIdentityProvider(Protocol): """Protocol for resolving identity handles into :class:`IdentityRecord`s. Distinct from :class:`MuseAttestationProvider` because identity lookups have different concurrency, caching, and trust-root semantics than attestation anchoring. A domain MAY register both providers (knowtation does — see :mod:`muse.plugins.knowtation.quorum`). Implementations MUST: * Be safe to call from any thread. * Be deterministic given a fixed backing store (the same handle always resolves to the same record between cache invalidations). * Return ``None`` for unknown handles — never raise for "not found". * Raise :class:`AttestationVerifyError` for transport/IO errors that prevent the resolver from determining whether the handle exists. """ def get_identity(self, handle: str) -> "IdentityRecord | None": """Resolve *handle* to its :class:`IdentityRecord`, or ``None``. Args: handle: Identity handle string. Case-sensitive. Returns: The :class:`IdentityRecord` for *handle*, or ``None`` if no such identity exists in the provider's backing store. Raises: AttestationVerifyError: If the provider cannot determine whether the handle exists (e.g. backing store unreachable). Callers MUST surface this as ``partial=True`` in the outer verification result, never silently treat as ``None``. """ ... _IDENTITY_REGISTRY_LOCK: threading.Lock = threading.Lock() _IDENTITY_PROVIDERS: dict[str, MuseIdentityProvider] = {} def register_identity_provider(domain: str, provider: MuseIdentityProvider) -> None: """Register *provider* under *domain* in the identity provider registry. Mirrors :func:`register_attestation_provider` exactly — same locking, same domain-isolation guarantees. Kept as a separate registry so that a domain can implement attestation without identity (or vice versa). Args: domain: Non-empty domain name. provider: Object implementing :class:`MuseIdentityProvider`. Raises: ValueError: If *domain* is empty. TypeError: If *provider* does not implement :class:`MuseIdentityProvider`. """ if not isinstance(domain, str) or not domain: raise ValueError("domain must be a non-empty string") if not isinstance(provider, MuseIdentityProvider): raise TypeError( f"provider must implement MuseIdentityProvider " f"(got {type(provider).__name__})" ) with _IDENTITY_REGISTRY_LOCK: _IDENTITY_PROVIDERS[domain] = provider def get_identity_provider(domain: str) -> "MuseIdentityProvider | None": """Return the identity provider for *domain*, or ``None``. Domain isolation is enforced — never falls back across domains. Args: domain: Domain name string. Returns: :class:`MuseIdentityProvider` or ``None``. """ with _IDENTITY_REGISTRY_LOCK: return _IDENTITY_PROVIDERS.get(domain) def unregister_identity_provider(domain: str) -> bool: """Remove the identity provider registered under *domain*, if any. Args: domain: Domain name string. Returns: ``True`` when a provider was removed; ``False`` when none was registered. """ with _IDENTITY_REGISTRY_LOCK: return _IDENTITY_PROVIDERS.pop(domain, None) is not None __all__ = [ "AnchorReceipt", "AttestationError", "AttestationRecord", "AttestationRequiredError", "AttestationVerifyError", "IdentityChainResult", "IdentityRecord", "ImmutableReceiptError", "MuseAttestationProvider", "MuseIdentityProvider", "VerifyResult", "get_attestation_provider", "get_identity_provider", "register_attestation_provider", "register_identity_provider", "unregister_attestation_provider", "unregister_identity_provider", ]