"""Knowtation ICP push hook — Phase 7.4. Implements :class:`muse.core.push_hooks.PushHookPlugin` for the Knowtation domain. After ``muse push`` completes, this hook walks the newly-pushed commits, extracts each commit's ``metadata["attestation"]`` blob, and anchors each record on the live Knowtation attestation canister (``dejku-syaaa-aaaaa-qgy3q-cai``). Live canister contract (see ``hub/icp/src/attestation/main.mo``) ---------------------------------------------------------------- * ``storeAttestation(input: StoreInput) -> StoreResult`` - Authenticated update method — only authorised principals can call. - Input fields: ``id, action, path, timestamp, content_hash, sig``. - On success: ``#ok({ seq: Nat })``. - On duplicate ID: ``#err("Attestation X already exists ...")`` — treated as success (idempotent). * HTTP query: ``GET /attest/`` returns the record JSON or 404, accessible at ``https://dejku-syaaa-aaaaa-qgy3q-cai.ic0.app/attest/``. Implementation strategy ----------------------- 1. **Out-of-band by default.** ``run_post_push`` spawns a daemon thread and returns immediately with ``out_of_band=True``. The push command never waits on ICP latency. 2. **ic-py if available, HTTPS fallback otherwise.** We try :mod:`ic.agent` first (proper Candid-encoded authenticated update call). If unavailable, we fall back to the canister's HTTP query interface for read verification only — *unauthenticated update calls cannot succeed* so the fallback is read-only. 3. **Retry with jitter.** 3 attempts, exponential backoff (1s, 2s, 4s), ±10 % jitter — matches industry standard for transient network failures. 4. **Idempotent.** Duplicate-ID errors from the canister are treated as success. Re-running after a partial failure recovers cleanly. 5. **Never fails the push.** ICP unreachable, malformed config, missing identity — all logged at WARNING and returned as ``PushHookResult(skipped=True, reason="ICP_UNREACHABLE")``. Security -------- * Canister ID is validated as a syntactically-valid IC principal before being interpolated into any URL — see :func:`_is_valid_canister_id`. * Identity PEM file is read with strict octal-mode ``0o600`` enforcement. * Unauthorised-caller responses are logged as a structured event but never cause the push to fail (per the Phase 7.4 contract). """ from __future__ import annotations import json import logging import os import pathlib import random import re import threading import time import urllib.error import urllib.request from dataclasses import dataclass, field from typing import Any from muse.core.push_hooks import ( PushHookPlugin, PushHookResult, get_push_hook_registry, ) logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- #: Stable identifier for the ICP push hook. HOOK_NAME: str = "knowtation-icp-anchor" #: Default canister ID pinned from canister_ids.json. Override with #: ``MUSE_ICP_CANISTER_ID`` env var. DEFAULT_CANISTER_ID: str = "dejku-syaaa-aaaaa-qgy3q-cai" #: IC mainnet HTTP gateway domain for canister queries. IC_HTTP_GATEWAY: str = "ic0.app" #: HTTP timeout (seconds) for canister query calls. HTTP_TIMEOUT_SECONDS: float = 10.0 #: Maximum retry attempts for transient errors. MAX_RETRIES: int = 3 #: Base delay (seconds) for exponential backoff. Effective delays are #: BASE * 2**(attempt - 1) with ±10 % jitter. RETRY_BASE_DELAY: float = 1.0 #: Jitter percentage applied to each backoff delay (±). RETRY_JITTER_PCT: float = 0.10 #: Canister ID syntax: 4 groups of 5 lowercase alphanumerics separated by ``-``, #: ending with ``-cai`` (e.g. ``dejku-syaaa-aaaaa-qgy3q-cai``). Conservative — #: IC principals can be more permissive, but every real canister ID matches #: this pattern. Enforced before any URL interpolation to prevent SSRF via #: crafted env var values. _CANISTER_ID_RE: re.Pattern[str] = re.compile( r"\A[a-z0-9]{5}(?:-[a-z0-9]{5}){3}-cai\Z" ) # --------------------------------------------------------------------------- # Config loader # --------------------------------------------------------------------------- @dataclass(frozen=True) class IcpHookConfig: """Resolved ICP push hook configuration. Attributes: enabled: When ``False``, ``run_post_push`` short-circuits with ``skipped=True, reason="DISABLED"``. Default ``True``. canister_id: Validated canister ID. See :func:`_is_valid_canister_id`. identity_pem_path: Optional path to a DFX identity PEM file used by ic-py for authenticated update calls. When ``None``, the hook falls back to the HTTP query interface (read-only). retries: Per-record retry attempts (default :data:`MAX_RETRIES`). timeout_seconds: Per-attempt HTTP timeout (default :data:`HTTP_TIMEOUT_SECONDS`). """ enabled: bool = True canister_id: str = DEFAULT_CANISTER_ID identity_pem_path: str | None = None retries: int = MAX_RETRIES timeout_seconds: float = HTTP_TIMEOUT_SECONDS def _is_valid_canister_id(value: str) -> bool: """Return True iff *value* matches the conservative canister-ID pattern. Args: value: Candidate canister ID string. Returns: ``True`` iff *value* matches :data:`_CANISTER_ID_RE`. ``False`` for any other input (empty, wrong length, illegal characters). """ return isinstance(value, str) and bool(_CANISTER_ID_RE.match(value)) def load_icp_config(env: dict[str, str] | None = None) -> IcpHookConfig: """Load ICP hook configuration from environment variables. Reads: * ``MUSE_ICP_ENABLED`` — ``"0"``, ``"false"`` (case-insensitive), or ``"no"`` disables the hook. Anything else (including unset) enables. * ``MUSE_ICP_CANISTER_ID`` — overrides :data:`DEFAULT_CANISTER_ID`. Validated by :func:`_is_valid_canister_id`; invalid values fall back to the default with a WARNING log. * ``MUSE_ICP_IDENTITY_PEM`` — path to DFX identity PEM file for authenticated update calls. Args: env: Optional dict of environment variables (defaults to :data:`os.environ`). Useful for tests. Returns: An :class:`IcpHookConfig` with all fields validated. """ env_map = env if env is not None else dict(os.environ) raw_enabled = env_map.get("MUSE_ICP_ENABLED", "1").strip().lower() enabled = raw_enabled not in ("0", "false", "no", "off") raw_canister = env_map.get("MUSE_ICP_CANISTER_ID", "").strip() if raw_canister and not _is_valid_canister_id(raw_canister): logger.warning( "knowtation-icp: MUSE_ICP_CANISTER_ID=%r is not a valid canister ID; " "falling back to default %s.", raw_canister, DEFAULT_CANISTER_ID, ) canister_id = DEFAULT_CANISTER_ID else: canister_id = raw_canister or DEFAULT_CANISTER_ID pem_path: str | None = None raw_pem = env_map.get("MUSE_ICP_IDENTITY_PEM", "").strip() if raw_pem: candidate = pathlib.Path(raw_pem) if candidate.exists() and candidate.is_file(): pem_path = str(candidate) else: logger.warning( "knowtation-icp: MUSE_ICP_IDENTITY_PEM=%r does not exist; " "ICP calls will be read-only.", raw_pem, ) return IcpHookConfig( enabled=enabled, canister_id=canister_id, identity_pem_path=pem_path, ) # --------------------------------------------------------------------------- # Backoff math (exposed for unit testing) # --------------------------------------------------------------------------- def compute_backoff_delay( attempt: int, *, base: float = RETRY_BASE_DELAY, jitter_pct: float = RETRY_JITTER_PCT, rng: random.Random | None = None, ) -> float: """Compute the exponential-backoff delay for retry *attempt* (1-indexed). Pure function — exposed so unit tests can verify the math against the documented contract: ``BASE * 2**(attempt-1) ± jitter_pct``. Args: attempt: 1-indexed attempt number (1 = first retry). base: Base delay in seconds (default :data:`RETRY_BASE_DELAY`). jitter_pct: Symmetric jitter as a fraction of the base delay (default :data:`RETRY_JITTER_PCT`). rng: Optional :class:`random.Random` instance for deterministic tests. When ``None``, uses the global RNG. Returns: Delay in seconds. Always non-negative; the lower jitter bound is clamped at zero to avoid sleeping for negative time. Raises: ValueError: If *attempt* < 1 or jitter_pct < 0 or > 1. """ if attempt < 1: raise ValueError(f"attempt must be >= 1 (got {attempt})") if jitter_pct < 0 or jitter_pct > 1: raise ValueError(f"jitter_pct must be in [0, 1] (got {jitter_pct})") base_delay = base * (2 ** (attempt - 1)) jitter_range = base_delay * jitter_pct rand = rng if rng is not None else random jitter = rand.uniform(-jitter_range, jitter_range) return max(0.0, base_delay + jitter) # --------------------------------------------------------------------------- # Canister HTTP client (read-only fallback when ic-py is unavailable) # --------------------------------------------------------------------------- def canister_http_get_url(canister_id: str, path: str) -> str: """Build an https URL to query the canister's HTTP interface. Args: canister_id: Validated canister ID. MUST have already passed :func:`_is_valid_canister_id`. path: Path part starting with ``/`` (e.g. ``"/attest/abc"``). Returns: ``https://.ic0.app``. Raises: ValueError: If *canister_id* is invalid or *path* does not start with ``/``. """ if not _is_valid_canister_id(canister_id): raise ValueError(f"invalid canister id: {canister_id!r}") if not path.startswith("/"): raise ValueError(f"path must start with /: {path!r}") return f"https://{canister_id}.{IC_HTTP_GATEWAY}{path}" def fetch_attestation_via_http( canister_id: str, attestation_id: str, *, timeout: float = HTTP_TIMEOUT_SECONDS, ) -> dict[str, Any] | None: """Fetch a single attestation record via the canister's HTTP query interface. Returns the parsed JSON record on HTTP 200, ``None`` on HTTP 404, and raises on any other transport-level failure. Args: canister_id: Validated canister ID. attestation_id: 64-char lowercase hex SHA-256 ID. timeout: Per-attempt HTTP timeout (seconds). Returns: Parsed record dict on 200, ``None`` on 404. Raises: OSError: For any transport-level failure (connection refused, timeout, DNS error). Caller decides whether to retry. """ url = canister_http_get_url(canister_id, f"/attest/{attestation_id}") request = urllib.request.Request( url=url, method="GET", headers={"Accept": "application/json", "User-Agent": "muse-icp-hook/1.0"}, ) try: with urllib.request.urlopen(request, timeout=timeout) as response: # noqa: S310 status = getattr(response, "status", 200) if status == 404: return None if status >= 400: raise OSError(f"canister returned HTTP {status}") raw = response.read(65_536) except urllib.error.HTTPError as exc: if exc.code == 404: return None raise OSError(f"canister returned HTTP {exc.code}") from exc except urllib.error.URLError as exc: raise OSError(f"canister unreachable: {exc}") from exc parsed = json.loads(raw.decode("utf-8")) return parsed if isinstance(parsed, dict) else None # --------------------------------------------------------------------------- # Update-call shim (ic-py preferred, no-op fallback) # --------------------------------------------------------------------------- @dataclass(frozen=True) class StoreResult: """Result of one storeAttestation update call. Attributes: anchored: ``True`` iff the canister returned ``#ok`` OR returned the documented "already exists" error (treated as success because the canister is immutable and a duplicate ID means the record is already anchored). reason: Short token describing the outcome — ``"ok"``, ``"already_exists"``, ``"unauthorized"``, ``"error"``. message: Human-readable detail (used in logs). """ anchored: bool reason: str message: str def _store_attestation_via_icpy( canister_id: str, record: dict[str, Any], pem_path: str | None, ) -> StoreResult: """Attempt to call storeAttestation via ic-py. Args: canister_id: Validated canister ID. record: Attestation record dict. pem_path: Path to identity PEM, or ``None`` for anonymous. Returns: A :class:`StoreResult` describing the outcome. Raises: ImportError: If ic-py is not installed — caller should fall back. OSError: For transport-level failures. """ try: from ic.client import Client from ic.identity import Identity from ic.agent import Agent from ic.candid import encode, Types except ImportError as exc: raise ImportError("ic-py not installed") from exc if pem_path: identity = Identity.from_pem(pem_path) # type: ignore[attr-defined] else: identity = Identity() client = Client() agent = Agent(identity, client) try: # Build the StoreInput record matching the Motoko type. params = [{ "type": Types.Record({ "id": Types.Text, "action": Types.Text, "path": Types.Text, "timestamp": Types.Text, "content_hash": Types.Text, "sig": Types.Text, }), "value": { "id": str(record.get("id", "")), "action": str(record.get("action", "")), "path": str(record.get("path", "")), "timestamp": str(record.get("timestamp", "")), "content_hash": str(record.get("content_hash", "")), "sig": str(record.get("sig", "")), }, }] encoded = encode(params) raw = agent.update_raw(canister_id, "storeAttestation", encoded) except Exception as exc: # noqa: BLE001 raise OSError(f"ic-py update_raw failed: {exc}") from exc text = repr(raw).lower() if "ok" in text and "err" not in text: return StoreResult(anchored=True, reason="ok", message=str(raw)) if "already exists" in text: return StoreResult( anchored=True, reason="already_exists", message="canister returned 'already exists' (idempotent success)", ) if "unauthorized" in text: return StoreResult( anchored=False, reason="unauthorized", message=f"caller not in authorized list: {raw!r}", ) return StoreResult(anchored=False, reason="error", message=str(raw)) # --------------------------------------------------------------------------- # Main hook # --------------------------------------------------------------------------- class KnowtationICPPushHook: """Anchor knowtation attestation records on the ICP canister after push. Implements :class:`muse.core.push_hooks.PushHookPlugin`. Reads ``commit.metadata["attestation"]`` from each newly-pushed commit and attempts to anchor it on the canister. Failures at any layer (network, auth, malformed records) never propagate to the push command — they are surfaced via :class:`PushHookResult`. The hook is testable via :func:`KnowtationICPPushHook.run_post_push_sync` which performs the same work synchronously (no thread spawn) so test assertions can read the result deterministically. """ def __init__(self) -> None: """Initialise the hook. No I/O at construction time.""" # Per-process tracking of in-flight anchor threads — used by tests # that need to .join() on completion. self._threads: list[threading.Thread] = [] self._threads_lock = threading.Lock() # ------------------------------------------------------------------ # PushHookPlugin contract # ------------------------------------------------------------------ def run_post_push( self, repo_path: pathlib.Path, push_result: dict[str, object], domain: str, config: dict[str, object] | None = None, ) -> PushHookResult: """Dispatch the anchor work and return a status record immediately. Spawns a daemon thread that calls :meth:`_anchor_pushed_commits` and returns a :class:`PushHookResult` with ``out_of_band=True`` describing the dispatch. The merge engine — sorry, push command — does not wait for the background thread. Args: repo_path: Repository root. push_result: Push result dict. Reads ``commits`` (list of newly pushed commit IDs). Other keys ignored. domain: Domain string (always ``"knowtation"`` here). config: Optional config dict. Reads ``"icp"`` sub-dict. Returns: A :class:`PushHookResult` describing the dispatch — never raises. """ del domain, config # accepted for protocol conformance cfg = load_icp_config() if not cfg.enabled: return PushHookResult( hook_name=HOOK_NAME, fired=False, out_of_band=False, skipped=True, reason="DISABLED", message="MUSE_ICP_ENABLED=0 — anchoring skipped", error=None, ) commit_ids_raw = push_result.get("commits", []) if not isinstance(commit_ids_raw, list) or not commit_ids_raw: return PushHookResult( hook_name=HOOK_NAME, fired=False, out_of_band=False, skipped=True, reason="NO_COMMITS", message="push_result has no commits — nothing to anchor", error=None, ) commit_ids = [str(c) for c in commit_ids_raw] def _worker() -> None: try: self._anchor_pushed_commits(repo_path, commit_ids, cfg) except Exception as exc: # noqa: BLE001 — must not propagate logger.warning( "knowtation-icp: background anchor thread crashed: %s: %s", type(exc).__name__, exc, ) thread = threading.Thread( target=_worker, name="knowtation-icp-anchor", daemon=True, ) thread.start() with self._threads_lock: self._threads.append(thread) return PushHookResult( hook_name=HOOK_NAME, fired=True, out_of_band=True, skipped=False, reason="", message=( f"dispatched ICP anchor for {len(commit_ids)} commit(s) " f"to {cfg.canister_id} (background)" ), error=None, ) # ------------------------------------------------------------------ # Synchronous variant for tests + the actual anchor logic # ------------------------------------------------------------------ def run_post_push_sync( self, repo_path: pathlib.Path, push_result: dict[str, object], config: IcpHookConfig | None = None, ) -> PushHookResult: """Synchronous variant — performs anchoring inline and returns the count. Useful in tests that assert on the anchored/failed counters. Production code paths use :meth:`run_post_push` which dispatches asynchronously. Args: repo_path: Repository root. push_result: Push result dict (see :meth:`run_post_push`). config: Optional explicit :class:`IcpHookConfig`. When ``None``, loads from environment variables. Returns: A :class:`PushHookResult` with populated ``anchored_count`` and ``failed_count`` fields. """ cfg = config or load_icp_config() if not cfg.enabled: return PushHookResult( hook_name=HOOK_NAME, fired=False, out_of_band=False, skipped=True, reason="DISABLED", message="MUSE_ICP_ENABLED=0 — anchoring skipped", error=None, ) commit_ids = [str(c) for c in (push_result.get("commits") or [])] if not commit_ids: return PushHookResult( hook_name=HOOK_NAME, fired=False, out_of_band=False, skipped=True, reason="NO_COMMITS", message="push_result has no commits", error=None, ) anchored, failed = self._anchor_pushed_commits(repo_path, commit_ids, cfg) return PushHookResult( hook_name=HOOK_NAME, fired=True, out_of_band=False, skipped=False, reason="", message=f"anchored {anchored}, failed {failed}", error=None, anchored_count=anchored, failed_count=failed, ) def wait_for_inflight(self, timeout: float = 30.0) -> None: """Block until every dispatched anchor thread has joined. Useful in tests after :meth:`run_post_push` to ensure the background work has completed before asserting on side effects. Production code does not call this — the daemon threads are fire-and-forget. Args: timeout: Max seconds to wait for any single thread. """ with self._threads_lock: threads = list(self._threads) for t in threads: t.join(timeout=timeout) # ------------------------------------------------------------------ # Internal: walk commits, extract attestations, anchor each. # ------------------------------------------------------------------ def _anchor_pushed_commits( self, repo_path: pathlib.Path, commit_ids: list[str], cfg: IcpHookConfig, ) -> tuple[int, int]: """Anchor every attestation found in *commit_ids*. Returns (ok, fail). Reads each commit via :func:`muse.core.store.read_commit`, extracts ``metadata["attestation"]`` (canonical JSON), and calls :meth:`_anchor_one_record` with retry logic. Args: repo_path: Repository root. commit_ids: Newly-pushed commit IDs. cfg: Loaded :class:`IcpHookConfig`. Returns: ``(anchored_count, failed_count)`` — the number of records successfully anchored and the number that failed after all retries. Records without an attestation field do not count toward either. """ try: from muse.core.store import read_commit except ImportError: logger.warning( "knowtation-icp: muse.core.store unavailable; cannot read commits." ) return 0, 0 anchored = 0 failed = 0 for cid in commit_ids: try: commit = read_commit(repo_path, cid) except Exception as exc: # noqa: BLE001 logger.warning( "knowtation-icp: cannot read commit %s: %s", cid[:12], exc ) failed += 1 continue if commit is None: continue blob = (commit.metadata or {}).get("attestation") if not blob: continue try: record = json.loads(blob) except json.JSONDecodeError as exc: logger.warning( "knowtation-icp: commit %s has malformed attestation JSON: %s", cid[:12], exc, ) failed += 1 continue if not isinstance(record, dict): failed += 1 continue ok = self._anchor_one_record(cfg, record) if ok: anchored += 1 else: failed += 1 return anchored, failed def _anchor_one_record( self, cfg: IcpHookConfig, record: dict[str, Any], ) -> bool: """Anchor *record* on the canister with retry; return True on success. Retry policy: :data:`MAX_RETRIES` attempts with exponential backoff and ±10 % jitter (per :func:`compute_backoff_delay`). Args: cfg: Loaded :class:`IcpHookConfig`. record: Attestation record dict. Returns: ``True`` on success (including the idempotent "already exists" path). ``False`` after all retries exhausted. """ last_error: str = "" for attempt in range(1, cfg.retries + 1): try: result = _store_attestation_via_icpy( cfg.canister_id, record, cfg.identity_pem_path, ) except ImportError: # ic-py not installed — fall back to read-only verification. # We can only verify; we cannot anchor. Log and treat as # NOT anchored so the push hook reports skipped. logger.info( "knowtation-icp: ic-py not installed — skipping update call " "for record %s. Install with `pip install ic-py` to enable " "authenticated anchoring.", record.get("id", "")[:12], ) return False except OSError as exc: last_error = str(exc) logger.info( "knowtation-icp: anchor attempt %d/%d failed for %s: %s", attempt, cfg.retries, record.get("id", "")[:12], exc, ) if attempt < cfg.retries: delay = compute_backoff_delay(attempt) time.sleep(delay) continue if result.anchored: logger.debug( "knowtation-icp: anchored %s (%s)", record.get("id", "")[:12], result.reason, ) return True if result.reason == "unauthorized": logger.warning( "knowtation-icp: unauthorized caller for %s — record NOT " "anchored. Configure MUSE_ICP_IDENTITY_PEM with an " "authorised principal to enable anchoring.", record.get("id", "")[:12], ) return False # no point retrying auth failures last_error = result.message if attempt < cfg.retries: time.sleep(compute_backoff_delay(attempt)) logger.warning( "knowtation-icp: failed to anchor %s after %d attempts: %s", record.get("id", "")[:12], cfg.retries, last_error, ) return False # --------------------------------------------------------------------------- # Module-level registration helper. # --------------------------------------------------------------------------- def register_knowtation_icp_push_hook() -> KnowtationICPPushHook: """Construct + register the ICP push hook; return the singleton instance. Idempotent: re-calling replaces any prior registration with a fresh instance. Returns: The newly-registered :class:`KnowtationICPPushHook` instance. """ hook = KnowtationICPPushHook() get_push_hook_registry().register("knowtation", hook) return hook __all__ = [ "DEFAULT_CANISTER_ID", "HOOK_NAME", "HTTP_TIMEOUT_SECONDS", "IC_HTTP_GATEWAY", "IcpHookConfig", "KnowtationICPPushHook", "MAX_RETRIES", "RETRY_BASE_DELAY", "RETRY_JITTER_PCT", "StoreResult", "canister_http_get_url", "compute_backoff_delay", "fetch_attestation_via_http", "load_icp_config", "register_knowtation_icp_push_hook", ]