"""Knowtation HMAC AIR attestation provider — Phase 7.2. Python port of ``knowtation/lib/air.mjs``: implements :class:`muse.core.attestation.MuseAttestationProvider` for the Knowtation domain by calling the local AIR HMAC daemon, then anchoring the resulting record on the immutable ICP attestation canister via Phase 7.4's push hook. Wire format — strict cross-language parity with air.mjs ------------------------------------------------------- The HTTP body sent to the AIR endpoint is a JSON object with **exactly three keys, always present, in canonical alphabetical order**:: {"action": "", "content_hash": "", "path": ""} This matches the upgraded ``knowtation/lib/air.mjs`` which now also forwards ``content_hash``. The cross-language parity test in ``tests/test_knowtation_attestation.py`` verifies that 200 random fixtures serialise identically on both sides — see :func:`build_air_request_body`. Determinism ----------- The :attr:`AttestationRecord.id` is computed as:: SHA-256(snapshot_id + "\\x00" + note_path).hexdigest() — deterministic, content-addressed, and globally unique within a commit graph. Two replays of the same commit on different nodes produce identical IDs, so the canister rejects the second write with the documented ``"already exists"`` error and the system stays consistent. Inbox bypass ------------ Notes whose vault-relative path matches :func:`is_inbox_path` are skipped — the same regex as JS ``isInboxPath``: literal ``inbox``, ``inbox/...``, or ``projects//inbox(/...)?``. Required-mode rejection ----------------------- When ``config.air.required=True`` and the AIR endpoint returns non-OK or is unreachable, :meth:`KnowtationHmacAttestationProvider.compute_attestation` raises :class:`muse.core.attestation.AttestationRequiredError` — same code, same semantics as the JS ``AttestationRequiredError``. Verification ------------ :meth:`KnowtationHmacAttestationProvider.verify` recomputes the canonical HMAC bytes locally using a shared HMAC key loaded from ``MUSE_AIR_HMAC_KEY`` (or ``config.air.hmac_key_path``) and constant-time compares against ``record["sig"]``. Missing-key and partial-verify scenarios raise :class:`muse.core.attestation.AttestationVerifyError` with the appropriate code — never silently return ``valid=True``. """ from __future__ import annotations import hashlib import hmac import json import logging import os import pathlib import re import urllib.error import urllib.request from typing import TypedDict from muse.core.attestation import ( AnchorReceipt, AttestationError, AttestationRecord, AttestationRequiredError, AttestationVerifyError, ImmutableReceiptError, MuseAttestationProvider, VerifyResult, register_attestation_provider, ) logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Configuration # --------------------------------------------------------------------------- #: Domain name this provider registers under. PROVIDER_DOMAIN: str = "knowtation" #: Stable algorithm identifier embedded in every record. ALGORITHM: str = "hmac-sha256" #: Stable identifier for the placeholder sig used when the AIR endpoint is #: unconfigured/unreachable AND ``air.required=False``. Exposed as a #: constant so verifiers can recognise placeholders without string-matching. PLACEHOLDER_SIG: str = "air-placeholder-write" #: HTTP timeout for AIR endpoint POSTs (seconds). HTTP_TIMEOUT_SECONDS: float = 10.0 #: Field separator for canonical record bytes — same null-byte convention as #: muse.core.provenance.provenance_payload to prevent separator-injection #: attacks from field values containing the separator. _CANONICAL_SEP: str = "\x00" #: Maximum HTTP response body bytes accepted from the AIR endpoint. #: 64 KiB is generous for a JSON ID response and bounds memory. _MAX_RESPONSE_BYTES: int = 65_536 #: Inbox-path regex — mirrors JS ``isInboxPath`` in air.mjs. #: Matches: ``inbox``, ``inbox/...``, ``projects//inbox(/...)?``. _INBOX_RE: re.Pattern[str] = re.compile(r"^projects/[^/]+/inbox(?:/|$)") class AirConfig(TypedDict, total=False): """Subset of ``.muse/air.yaml`` consumed by this provider. Mirrors the shape of ``config.air`` consumed by ``knowtation/lib/air.mjs``, plus the new ``hmac_key`` field used by the Python verifier. Attributes: enabled: When ``False`` (the default), the provider is a no-op: :meth:`compute_attestation` returns a placeholder record without calling the endpoint. When ``True``, the endpoint is consulted. required: When ``True``, an unreachable or non-OK endpoint causes :meth:`compute_attestation` to raise :class:`AttestationRequiredError`. When ``False`` (the default), the provider degrades to a placeholder sig and logs a warning. endpoint: Full HTTPS URL of the AIR daemon's POST endpoint. Falls back to ``$KNOWTATION_AIR_ENDPOINT`` when omitted. hmac_key: Hex-encoded shared HMAC key used by :meth:`verify`. Falls back to ``$MUSE_AIR_HMAC_KEY`` when omitted. When neither source is set, :meth:`verify` raises :class:`AttestationVerifyError(code="KEY_MISSING")`. """ enabled: bool required: bool endpoint: str hmac_key: str # --------------------------------------------------------------------------- # Inbox detection — port of isInboxPath in air.mjs # --------------------------------------------------------------------------- def is_inbox_path(vault_relative_path: str) -> bool: """Return True iff *vault_relative_path* is under an inbox directory. Identical semantics to the JS ``isInboxPath()`` in ``knowtation/lib/air.mjs``:: - "inbox" → True - "inbox/anything" → True - "projects/foo/inbox" → True - "projects/foo/inbox/note.md" → True - "projects/foo/bar/inbox/note.md" → False (only direct children) Backslashes are normalised to forward slashes so paths originating on Windows are classified the same way as POSIX paths. Args: vault_relative_path: Vault-relative path of the note being attested. May use ``/`` or ``\\`` as separators. Returns: ``True`` if the path is in an inbox (and should bypass attestation), ``False`` otherwise. """ n = vault_relative_path.replace("\\", "/") if n == "inbox" or n.startswith("inbox/"): return True return bool(_INBOX_RE.match(n)) # --------------------------------------------------------------------------- # Cross-language wire-format builders — exact parity with air.mjs # --------------------------------------------------------------------------- def build_air_request_body( action: str, path: str, content_hash: str, ) -> bytes: """Build the exact UTF-8 bytes POSTed to the AIR endpoint. The body is a JSON object with **exactly three keys** in canonical alphabetical order: ``action``, ``content_hash``, ``path``. Producing the same byte sequence on both Python and JS sides is what enables the cross-language parity test in ``tests/test_knowtation_attestation.py``. Args: action: The attested action (``"write"``, ``"export"``, …). Must be a non-empty string; control characters are rejected. path: Vault-relative path to the note. May be empty for repo-scoped actions but must be a string. content_hash: SHA-256 hex of the content (lowercase). Empty string is accepted for actions that do not bind to specific bytes (e.g. legacy export records). Returns: UTF-8 encoded JSON bytes. Compact form (no whitespace), keys in alphabetical order, no NaN/Infinity tokens. Raises: ValueError: If any argument is not a ``str`` or contains an embedded null byte (which would break canonical encoding). """ for name, val in (("action", action), ("path", path), ("content_hash", content_hash)): if not isinstance(val, str): raise ValueError(f"{name} must be a string (got {type(val).__name__})") if "\x00" in val: raise ValueError(f"{name} must not contain embedded null bytes") payload = {"action": action, "content_hash": content_hash, "path": path} return json.dumps( payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False, ).encode("utf-8") def compute_attestation_id(snapshot_id: str, note_path: str) -> str: """Compute the deterministic attestation ID for a (snapshot, note) pair. Definition:: SHA-256(snapshot_id + "\\x00" + note_path).hexdigest() Lower-case hex, 64 characters. The null-byte separator prevents the classic concatenation collision (``"abc" + "def"`` vs ``"ab" + "cdef"``). Args: snapshot_id: SHA-256 hex of the snapshot containing the note. note_path: Vault-relative path of the note. Returns: 64-character lowercase hex SHA-256 digest. Raises: ValueError: If either argument is not a string or contains a null byte (which would create separator-injection collisions). """ for name, val in (("snapshot_id", snapshot_id), ("note_path", note_path)): if not isinstance(val, str): raise ValueError(f"{name} must be a string (got {type(val).__name__})") if "\x00" in val: raise ValueError(f"{name} must not contain embedded null bytes") raw = (snapshot_id + _CANONICAL_SEP + note_path).encode("utf-8") return hashlib.sha256(raw).hexdigest() def canonical_record_bytes(record: AttestationRecord) -> bytes: """Serialise a record into canonical bytes for HMAC signing/verification. Format (null-byte separated, fixed field order):: action || \\x00 || path || \\x00 || content_hash || \\x00 || timestamp || \\x00 || id Note: the ``sig`` field is intentionally excluded — it would create a self-reference loop. ``algorithm`` and ``provider_id`` are excluded because they are constants for this provider and including them would not add discrimination over the algorithm choice already encoded by the verifier's choice of HMAC-SHA256. Args: record: An :class:`AttestationRecord` produced by this provider. Missing ``action``/``path``/``content_hash``/``timestamp``/``id`` keys are treated as empty strings. Returns: UTF-8 encoded canonical bytes ready for HMAC. """ fields = [ record.get("action", ""), record.get("path", ""), record.get("content_hash", ""), record.get("timestamp", ""), record.get("id", ""), ] return _CANONICAL_SEP.join(fields).encode("utf-8") def _hmac_sha256_hex(key_hex: str, data: bytes) -> str: """Compute HMAC-SHA256 over *data* using a hex-encoded key. Args: key_hex: Hex-encoded shared HMAC key. Decoded with :func:`bytes.fromhex` so non-hex input raises :class:`ValueError`. data: Bytes to authenticate. Returns: 64-character lowercase hex HMAC digest. Raises: ValueError: If *key_hex* is empty or not valid hex. """ if not key_hex: raise ValueError("HMAC key is empty") key_bytes = bytes.fromhex(key_hex) return hmac.new(key_bytes, data, hashlib.sha256).hexdigest() # --------------------------------------------------------------------------- # AIR endpoint client — port of attestBeforeWrite in air.mjs # --------------------------------------------------------------------------- def _resolve_endpoint(config: AirConfig | None) -> str: """Resolve the AIR endpoint URL from config or environment. Mirrors the JS resolution: ``config.air.endpoint || process.env.KNOWTATION_AIR_ENDPOINT``. Args: config: Optional :class:`AirConfig` dict. Returns: Endpoint URL string, or ``""`` when neither source set it. """ if config and config.get("endpoint"): return str(config["endpoint"]) return os.environ.get("KNOWTATION_AIR_ENDPOINT", "") def _resolve_hmac_key(config: AirConfig | None) -> str: """Resolve the HMAC verification key from config or environment. Args: config: Optional :class:`AirConfig` dict. Returns: Hex-encoded HMAC key string, or ``""`` when neither source set it. """ if config and config.get("hmac_key"): return str(config["hmac_key"]) return os.environ.get("MUSE_AIR_HMAC_KEY", "") def _validate_endpoint_url(url: str) -> None: """Reject malformed or insecure endpoint URLs. Accepts ``http://localhost*`` (developer-friendly) and ``https://*`` URLs only. Rejects ``file://``, ``ftp://``, ``javascript:``, and other protocols that could turn a misconfigured endpoint into an arbitrary file read or out-of-band request. Args: url: Endpoint URL to validate. Raises: AttestationError: If the URL is empty or not http(s). """ if not url: raise AttestationError("AIR endpoint URL is empty") lower = url.lower().strip() if lower.startswith("https://"): return if lower.startswith("http://localhost") or lower.startswith("http://127.0.0.1"): return raise AttestationError( f"AIR endpoint URL must be https:// (or http://localhost for development); " f"got scheme {url.split('://', 1)[0]!r}" ) class _AirResponse(TypedDict, total=False): """Parsed AIR endpoint response shape.""" id: str air_id: str air_sig: str sig: str timestamp: str def _post_to_air( endpoint: str, body: bytes, *, timeout: float = HTTP_TIMEOUT_SECONDS, ) -> _AirResponse: """POST *body* to *endpoint* and return the parsed JSON response. The endpoint URL is validated by :func:`_validate_endpoint_url` before use; only ``https://`` and ``http://localhost`` URLs are accepted. Args: endpoint: Pre-validated AIR endpoint URL. body: Pre-built request body bytes from :func:`build_air_request_body`. timeout: HTTP socket timeout in seconds. Returns: Parsed JSON response as a dict; missing/non-JSON returns ``{}``. Raises: AttestationError: For any transport- or HTTP-level failure. The caller decides whether to convert to AttestationRequiredError based on ``config.air.required``. """ _validate_endpoint_url(endpoint) request = urllib.request.Request( url=endpoint, data=body, method="POST", headers={ "Content-Type": "application/json", "User-Agent": "muse-attestation/1.0", "Content-Length": str(len(body)), }, ) try: with urllib.request.urlopen( # noqa: S310 — URL pre-validated request, timeout=timeout ) as response: status = getattr(response, "status", 200) if status >= 400: raise AttestationError( f"AIR endpoint returned HTTP {status}" ) raw = response.read(_MAX_RESPONSE_BYTES + 1) if len(raw) > _MAX_RESPONSE_BYTES: raise AttestationError( f"AIR response exceeds {_MAX_RESPONSE_BYTES} byte cap" ) except urllib.error.HTTPError as exc: raise AttestationError( f"AIR endpoint returned HTTP {exc.code}" ) from exc except urllib.error.URLError as exc: raise AttestationError(f"AIR endpoint unreachable: {exc}") from exc except TimeoutError as exc: raise AttestationError(f"AIR endpoint timed out after {timeout}s") from exc try: parsed = json.loads(raw.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError): return {} if not isinstance(parsed, dict): return {} return parsed # --------------------------------------------------------------------------- # Provider implementation # --------------------------------------------------------------------------- class KnowtationHmacAttestationProvider: """HMAC AIR attestation provider for the Knowtation domain. Implements :class:`muse.core.attestation.MuseAttestationProvider`. The provider is constructed once per process (typically via the :func:`register_knowtation_attestation_provider` module-level helper) and then resolved from the registry at commit/verify time. Attributes: config: The :class:`AirConfig` dict used by every method. Stored as the provider's only mutable state; instances should be treated as immutable after construction (no mutation post-init in practice). """ config: AirConfig _anchored_ids: set[str] def __init__(self, config: AirConfig | None = None) -> None: """Initialise the provider with an optional configuration dict. Args: config: Optional :class:`AirConfig` to override env-var defaults. When ``None``, the provider falls back to environment variables for endpoint and HMAC key on every call. """ self.config = dict(config) if config else {} self._anchored_ids = set() # ------------------------------------------------------------------ # MuseAttestationProvider methods # ------------------------------------------------------------------ def compute_attestation( self, snapshot_id: str, commit_meta: dict[str, object], ) -> AttestationRecord: """Compute the attestation record for a knowtation commit. Implementation outline (matches air.mjs ``attestBeforeWrite``): 1. Read ``commit_meta["path"]`` (vault-relative) and ``commit_meta["content_hash"]`` (SHA-256 hex). Action defaults to ``"write"`` if not provided. 2. Compute the deterministic ID via :func:`compute_attestation_id`. 3. If the path is an inbox path (per :func:`is_inbox_path`), return the record with empty ``sig`` and ``algorithm="bypass-inbox"``. 4. If ``config.air.enabled`` is False, return placeholder. 5. Otherwise POST to the AIR endpoint via :func:`_post_to_air`. - On HTTP 2xx: extract ``data.air_sig`` (or ``data.sig``) for the record's ``sig`` field; ``data.id`` (or ``data.air_id``) is cross-checked against our deterministic ID and a mismatch is logged as a WARNING but does not fail. - On any failure with ``required=True``: raise :class:`AttestationRequiredError`. - On any failure with ``required=False``: degrade to placeholder and log a WARNING. Args: snapshot_id: SHA-256 hex of the snapshot containing the note. commit_meta: Dict with at least the keys ``path`` (str) and ``content_hash`` (str). Optional: ``action`` (str, default ``"write"``), ``timestamp`` (str ISO-8601, default ``""``). Other keys are ignored. Returns: An :class:`AttestationRecord` with all required fields set. Raises: AttestationRequiredError: If ``config.air.required=True`` and the endpoint cannot be reached or returns non-OK. ValueError: If ``commit_meta["path"]`` or ``commit_meta["content_hash"]`` is missing or not a string. """ path = commit_meta.get("path", "") content_hash = commit_meta.get("content_hash", "") action = commit_meta.get("action", "write") timestamp = commit_meta.get("timestamp", "") if not isinstance(path, str): raise ValueError("commit_meta['path'] must be a string") if not isinstance(content_hash, str): raise ValueError("commit_meta['content_hash'] must be a string") if not isinstance(action, str): raise ValueError("commit_meta['action'] must be a string") if not isinstance(timestamp, str): raise ValueError("commit_meta['timestamp'] must be a string") deterministic_id = compute_attestation_id(snapshot_id, path) base_record: AttestationRecord = { "id": deterministic_id, "action": action, "path": path, "timestamp": timestamp, "content_hash": content_hash, "sig": "", "algorithm": ALGORITHM, "provider_id": PROVIDER_DOMAIN, } # Inbox bypass — never call the endpoint, never raise. if is_inbox_path(path): base_record["sig"] = "" base_record["algorithm"] = "bypass-inbox" return base_record enabled = bool(self.config.get("enabled", False)) required = bool(self.config.get("required", False)) endpoint = _resolve_endpoint(self.config) if not enabled: base_record["sig"] = PLACEHOLDER_SIG base_record["algorithm"] = "placeholder-disabled" return base_record if not endpoint: if required: raise AttestationRequiredError( "knowtation: air.required=true but no AIR endpoint is " "configured; commit rejected." ) base_record["sig"] = PLACEHOLDER_SIG base_record["algorithm"] = "placeholder-no-endpoint" logger.warning( "knowtation-attestation: AIR enabled but endpoint unconfigured " "and required=false; storing placeholder." ) return base_record body = build_air_request_body(action, path, content_hash) try: data = _post_to_air(endpoint, body) except AttestationError as exc: if required: raise AttestationRequiredError( f"knowtation: AIR endpoint failed and air.required=true; " f"commit rejected. ({exc})" ) from exc base_record["sig"] = PLACEHOLDER_SIG base_record["algorithm"] = "placeholder-endpoint-failure" logger.warning( "knowtation-attestation: AIR endpoint failed (%s) and " "required=false; storing placeholder.", exc, ) return base_record sig = str(data.get("air_sig") or data.get("sig") or "") returned_id = str(data.get("id") or data.get("air_id") or "") if returned_id and returned_id != deterministic_id: logger.warning( "knowtation-attestation: AIR endpoint returned id %r but " "expected deterministic id %r; using deterministic id.", returned_id, deterministic_id, ) if data.get("timestamp"): base_record["timestamp"] = str(data["timestamp"]) if not sig: if required: raise AttestationRequiredError( "knowtation: AIR endpoint returned no signature and " "air.required=true; commit rejected." ) base_record["sig"] = PLACEHOLDER_SIG base_record["algorithm"] = "placeholder-no-sig" return base_record base_record["sig"] = sig return base_record def anchor(self, record: AttestationRecord) -> AnchorReceipt: """Return a local-only receipt — true on-chain anchoring is the push hook. For the HMAC AIR provider, the local AIR daemon already persists the record in its append-only log when :meth:`compute_attestation` returned successfully. Network-anchored persistence to the ICP canister is the responsibility of :class:`muse.plugins.knowtation.icp_push_hook.KnowtationICPPushHook` which fires asynchronously on ``muse push``. This method therefore returns a synthetic receipt whose ``tx_id`` encodes the record ID (so verifiers can correlate) and whose ``anchor_url`` points to the AIR endpoint where the record was persisted. Re-anchoring is rejected with :class:`ImmutableReceiptError` to mirror the canister's write-once semantics — see ``hub/icp/src/attestation/main.mo`` line 415. Args: record: An :class:`AttestationRecord` from :meth:`compute_attestation`. Must have a non-empty ``id``. Returns: An :class:`AnchorReceipt` with ``tx_id="air-local:"`` and ``anchor_url`` pointing to the AIR endpoint. Raises: ImmutableReceiptError: If this provider has already issued a receipt for ``record["id"]`` in the current process. AttestationError: If ``record["id"]`` is missing or empty. """ rid = record.get("id", "") if not rid: raise AttestationError("record['id'] is empty — cannot anchor") if rid in self._anchored_ids: raise ImmutableReceiptError( f"record {rid!r} already anchored locally (write-once)" ) self._anchored_ids.add(rid) endpoint = _resolve_endpoint(self.config) return { "tx_id": f"air-local:{rid}", "anchor_url": endpoint or "", "anchored_at": record.get("timestamp", ""), "provider_id": PROVIDER_DOMAIN, } def verify( self, record: AttestationRecord, receipt: AnchorReceipt, ) -> VerifyResult: """Verify the record's HMAC and that the receipt belongs to it. Verification chain: 1. Cross-check ``record["provider_id"]`` and ``receipt["provider_id"]`` — mismatch raises :class:`AttestationVerifyError`. 2. Recompute the deterministic ID from ``(snapshot_id, path)`` — wait, we don't have snapshot_id at verify-time. Instead we sanity-check the record's own ID format. 3. If ``record["sig"] == PLACEHOLDER_SIG`` → return ``valid=False`` with reason ``"placeholder-sig"`` (provably-bad: this isn't a real attestation). 4. Load the HMAC key from config or env; missing key raises :class:`AttestationVerifyError(code="KEY_MISSING")` — never silently returns ``valid=True``. 5. Recompute HMAC over :func:`canonical_record_bytes` and constant-time compare against ``record["sig"]``. Args: record: The :class:`AttestationRecord` to verify. receipt: The :class:`AnchorReceipt` to cross-check. Returns: A :class:`VerifyResult` with ``valid=True`` iff the HMAC matched and the receipt belongs to the record. ``valid=False`` only for *provably bad* records (placeholder sigs, HMAC mismatch). Raises: AttestationVerifyError: If the verifier cannot complete the check — missing HMAC key, malformed sig, provider_id mismatch. """ if record.get("provider_id") != PROVIDER_DOMAIN: raise AttestationVerifyError( f"record provider_id is {record.get('provider_id')!r}, " f"expected {PROVIDER_DOMAIN!r}", code="PROVIDER_MISMATCH", ) if receipt.get("provider_id") != PROVIDER_DOMAIN: raise AttestationVerifyError( f"receipt provider_id is {receipt.get('provider_id')!r}, " f"expected {PROVIDER_DOMAIN!r}", code="PROVIDER_MISMATCH", ) rid = record.get("id", "") if not rid or len(rid) != 64 or not all(c in "0123456789abcdef" for c in rid): raise AttestationVerifyError( f"record id is not a valid SHA-256 hex digest: {rid!r}", code="INVALID_RECORD_ID", ) sig = record.get("sig", "") if sig == PLACEHOLDER_SIG or record.get("algorithm", "").startswith("placeholder-"): return { "valid": False, "reason": "record contains a placeholder signature, not a real HMAC", "depth": 0, "provider_id": PROVIDER_DOMAIN, } if record.get("algorithm") == "bypass-inbox": return { "valid": False, "reason": "inbox-bypass record cannot be verified (no signature was issued)", "depth": 0, "provider_id": PROVIDER_DOMAIN, } if not sig: raise AttestationVerifyError( "record sig is empty — cannot verify", code="MISSING_SIG", ) key_hex = _resolve_hmac_key(self.config) if not key_hex: raise AttestationVerifyError( "no HMAC key configured (set MUSE_AIR_HMAC_KEY or " "config.air.hmac_key); cannot complete verification", code="KEY_MISSING", ) try: expected = _hmac_sha256_hex(key_hex, canonical_record_bytes(record)) except ValueError as exc: raise AttestationVerifyError( f"HMAC computation failed: {exc}", code="HMAC_COMPUTE_FAILED", ) from exc if hmac.compare_digest(expected, sig): return { "valid": True, "reason": "", "depth": 0, "provider_id": PROVIDER_DOMAIN, } return { "valid": False, "reason": "HMAC signature mismatch", "depth": 0, "provider_id": PROVIDER_DOMAIN, } # --------------------------------------------------------------------------- # Module-level registration helper — same pattern as register_knowtation_hook. # --------------------------------------------------------------------------- def register_knowtation_attestation_provider( config: AirConfig | None = None, ) -> KnowtationHmacAttestationProvider: """Construct + register the Knowtation HMAC provider; return the instance. Idempotent: re-calling replaces the prior registration with a fresh instance configured by *config*. Safe to call eagerly at import time (typical) or lazily once configuration is loaded. Args: config: Optional :class:`AirConfig`. When ``None``, the provider relies on environment variables on every call. Returns: The newly-registered :class:`KnowtationHmacAttestationProvider` instance. """ provider = KnowtationHmacAttestationProvider(config) register_attestation_provider(PROVIDER_DOMAIN, provider) return provider def load_air_config(config_path: pathlib.Path | None = None) -> AirConfig: """Load AIR configuration from a YAML file or environment variables. Reads ``.muse/air.yaml`` by default. Falls back to ``MUSE_AIR_CONFIG`` env var for the path. When neither file exists, returns an empty :class:`AirConfig` that the provider will treat as "disabled". The YAML is parsed with ``yaml.safe_load`` so no arbitrary object instantiation is possible. Args: config_path: Optional explicit path to the YAML file. When omitted, uses ``$MUSE_AIR_CONFIG`` if set, else ``.muse/air.yaml`` relative to the current working directory. Returns: An :class:`AirConfig` dict with whichever fields the file contained; missing fields default to absent. Raises: AttestationError: If the file exists but cannot be parsed or its top-level shape is not a mapping with an ``air`` key. """ import yaml if config_path is None: env_path = os.environ.get("MUSE_AIR_CONFIG", "") if env_path: config_path = pathlib.Path(env_path) else: config_path = pathlib.Path(".muse/air.yaml") if not config_path.exists(): return AirConfig() try: raw_text = config_path.read_text(encoding="utf-8") parsed = yaml.safe_load(raw_text) or {} except (OSError, yaml.YAMLError) as exc: raise AttestationError(f"cannot read AIR config {config_path}: {exc}") from exc if not isinstance(parsed, dict): raise AttestationError(f"AIR config root must be a mapping; got {type(parsed).__name__}") air_section = parsed.get("air", {}) if not isinstance(air_section, dict): raise AttestationError(f"AIR config 'air' key must be a mapping; got {type(air_section).__name__}") cfg: AirConfig = {} if "enabled" in air_section: cfg["enabled"] = bool(air_section["enabled"]) if "required" in air_section: cfg["required"] = bool(air_section["required"]) if isinstance(air_section.get("endpoint"), str): cfg["endpoint"] = air_section["endpoint"] if isinstance(air_section.get("hmac_key"), str): cfg["hmac_key"] = air_section["hmac_key"] return cfg __all__ = [ "ALGORITHM", "AirConfig", "HTTP_TIMEOUT_SECONDS", "KnowtationHmacAttestationProvider", "PLACEHOLDER_SIG", "PROVIDER_DOMAIN", "build_air_request_body", "canonical_record_bytes", "compute_attestation_id", "is_inbox_path", "load_air_config", "register_knowtation_attestation_provider", ]