icp_push_hook.py
python
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8
docs(KD-STAGING): sync governance after KD-6b DONE
Human
14 days ago
| 1 | """Knowtation ICP push hook — Phase 7.4. |
| 2 | |
| 3 | Implements :class:`muse.core.push_hooks.PushHookPlugin` for the Knowtation |
| 4 | domain. After ``muse push`` completes, this hook walks the newly-pushed |
| 5 | commits, extracts each commit's ``metadata["attestation"]`` blob, and |
| 6 | anchors each record on the live Knowtation attestation canister |
| 7 | (``dejku-syaaa-aaaaa-qgy3q-cai``). |
| 8 | |
| 9 | Live canister contract (see ``hub/icp/src/attestation/main.mo``) |
| 10 | ---------------------------------------------------------------- |
| 11 | |
| 12 | * ``storeAttestation(input: StoreInput) -> StoreResult`` |
| 13 | - Authenticated update method — only authorised principals can call. |
| 14 | - Input fields: ``id, action, path, timestamp, content_hash, sig``. |
| 15 | - On success: ``#ok({ seq: Nat })``. |
| 16 | - On duplicate ID: ``#err("Attestation X already exists ...")`` — treated |
| 17 | as success (idempotent). |
| 18 | * HTTP query: ``GET /attest/<id>`` returns the record JSON or 404, accessible |
| 19 | at ``https://dejku-syaaa-aaaaa-qgy3q-cai.ic0.app/attest/<id>``. |
| 20 | |
| 21 | Implementation strategy |
| 22 | ----------------------- |
| 23 | |
| 24 | 1. **Out-of-band by default.** ``run_post_push`` spawns a daemon thread |
| 25 | and returns immediately with ``out_of_band=True``. The push command |
| 26 | never waits on ICP latency. |
| 27 | 2. **ic-py if available, HTTPS fallback otherwise.** We try |
| 28 | :mod:`ic.agent` first (proper Candid-encoded authenticated update call). |
| 29 | If unavailable, we fall back to the canister's HTTP query interface for |
| 30 | read verification only — *unauthenticated update calls cannot succeed* |
| 31 | so the fallback is read-only. |
| 32 | 3. **Retry with jitter.** 3 attempts, exponential backoff |
| 33 | (1s, 2s, 4s), ±10 % jitter — matches industry standard for transient |
| 34 | network failures. |
| 35 | 4. **Idempotent.** Duplicate-ID errors from the canister are treated as |
| 36 | success. Re-running after a partial failure recovers cleanly. |
| 37 | 5. **Never fails the push.** ICP unreachable, malformed config, missing |
| 38 | identity — all logged at WARNING and returned as |
| 39 | ``PushHookResult(skipped=True, reason="ICP_UNREACHABLE")``. |
| 40 | |
| 41 | Security |
| 42 | -------- |
| 43 | |
| 44 | * Canister ID is validated as a syntactically-valid IC principal before |
| 45 | being interpolated into any URL — see :func:`_is_valid_canister_id`. |
| 46 | * Identity PEM file is read with strict octal-mode ``0o600`` enforcement. |
| 47 | * Unauthorised-caller responses are logged as a structured event but never |
| 48 | cause the push to fail (per the Phase 7.4 contract). |
| 49 | """ |
| 50 | |
| 51 | from __future__ import annotations |
| 52 | |
| 53 | import json |
| 54 | import logging |
| 55 | import os |
| 56 | import pathlib |
| 57 | import random |
| 58 | import re |
| 59 | import threading |
| 60 | import time |
| 61 | import urllib.error |
| 62 | import urllib.request |
| 63 | from dataclasses import dataclass, field |
| 64 | from typing import Any |
| 65 | |
| 66 | from muse.core.push_hooks import ( |
| 67 | PushHookPlugin, |
| 68 | PushHookResult, |
| 69 | get_push_hook_registry, |
| 70 | ) |
| 71 | |
| 72 | logger = logging.getLogger(__name__) |
| 73 | |
| 74 | |
| 75 | # --------------------------------------------------------------------------- |
| 76 | # Constants |
| 77 | # --------------------------------------------------------------------------- |
| 78 | |
| 79 | #: Stable identifier for the ICP push hook. |
| 80 | HOOK_NAME: str = "knowtation-icp-anchor" |
| 81 | |
| 82 | #: Default canister ID pinned from canister_ids.json. Override with |
| 83 | #: ``MUSE_ICP_CANISTER_ID`` env var. |
| 84 | DEFAULT_CANISTER_ID: str = "dejku-syaaa-aaaaa-qgy3q-cai" |
| 85 | |
| 86 | #: IC mainnet HTTP gateway domain for canister queries. |
| 87 | IC_HTTP_GATEWAY: str = "ic0.app" |
| 88 | |
| 89 | #: HTTP timeout (seconds) for canister query calls. |
| 90 | HTTP_TIMEOUT_SECONDS: float = 10.0 |
| 91 | |
| 92 | #: Maximum retry attempts for transient errors. |
| 93 | MAX_RETRIES: int = 3 |
| 94 | |
| 95 | #: Base delay (seconds) for exponential backoff. Effective delays are |
| 96 | #: BASE * 2**(attempt - 1) with ±10 % jitter. |
| 97 | RETRY_BASE_DELAY: float = 1.0 |
| 98 | |
| 99 | #: Jitter percentage applied to each backoff delay (±). |
| 100 | RETRY_JITTER_PCT: float = 0.10 |
| 101 | |
| 102 | #: Canister ID syntax: 4 groups of 5 lowercase alphanumerics separated by ``-``, |
| 103 | #: ending with ``-cai`` (e.g. ``dejku-syaaa-aaaaa-qgy3q-cai``). Conservative — |
| 104 | #: IC principals can be more permissive, but every real canister ID matches |
| 105 | #: this pattern. Enforced before any URL interpolation to prevent SSRF via |
| 106 | #: crafted env var values. |
| 107 | _CANISTER_ID_RE: re.Pattern[str] = re.compile( |
| 108 | r"\A[a-z0-9]{5}(?:-[a-z0-9]{5}){3}-cai\Z" |
| 109 | ) |
| 110 | |
| 111 | |
| 112 | # --------------------------------------------------------------------------- |
| 113 | # Config loader |
| 114 | # --------------------------------------------------------------------------- |
| 115 | |
| 116 | |
| 117 | @dataclass(frozen=True) |
| 118 | class IcpHookConfig: |
| 119 | """Resolved ICP push hook configuration. |
| 120 | |
| 121 | Attributes: |
| 122 | enabled: When ``False``, ``run_post_push`` short-circuits with |
| 123 | ``skipped=True, reason="DISABLED"``. Default ``True``. |
| 124 | canister_id: Validated canister ID. See :func:`_is_valid_canister_id`. |
| 125 | identity_pem_path: Optional path to a DFX identity PEM file used by |
| 126 | ic-py for authenticated update calls. When ``None``, the hook |
| 127 | falls back to the HTTP query interface (read-only). |
| 128 | retries: Per-record retry attempts (default :data:`MAX_RETRIES`). |
| 129 | timeout_seconds: Per-attempt HTTP timeout |
| 130 | (default :data:`HTTP_TIMEOUT_SECONDS`). |
| 131 | """ |
| 132 | |
| 133 | enabled: bool = True |
| 134 | canister_id: str = DEFAULT_CANISTER_ID |
| 135 | identity_pem_path: str | None = None |
| 136 | retries: int = MAX_RETRIES |
| 137 | timeout_seconds: float = HTTP_TIMEOUT_SECONDS |
| 138 | |
| 139 | |
| 140 | def _is_valid_canister_id(value: str) -> bool: |
| 141 | """Return True iff *value* matches the conservative canister-ID pattern. |
| 142 | |
| 143 | Args: |
| 144 | value: Candidate canister ID string. |
| 145 | |
| 146 | Returns: |
| 147 | ``True`` iff *value* matches :data:`_CANISTER_ID_RE`. ``False`` |
| 148 | for any other input (empty, wrong length, illegal characters). |
| 149 | """ |
| 150 | return isinstance(value, str) and bool(_CANISTER_ID_RE.match(value)) |
| 151 | |
| 152 | |
| 153 | def load_icp_config(env: dict[str, str] | None = None) -> IcpHookConfig: |
| 154 | """Load ICP hook configuration from environment variables. |
| 155 | |
| 156 | Reads: |
| 157 | |
| 158 | * ``MUSE_ICP_ENABLED`` — ``"0"``, ``"false"`` (case-insensitive), or |
| 159 | ``"no"`` disables the hook. Anything else (including unset) enables. |
| 160 | * ``MUSE_ICP_CANISTER_ID`` — overrides :data:`DEFAULT_CANISTER_ID`. |
| 161 | Validated by :func:`_is_valid_canister_id`; invalid values fall back |
| 162 | to the default with a WARNING log. |
| 163 | * ``MUSE_ICP_IDENTITY_PEM`` — path to DFX identity PEM file for |
| 164 | authenticated update calls. |
| 165 | |
| 166 | Args: |
| 167 | env: Optional dict of environment variables (defaults to |
| 168 | :data:`os.environ`). Useful for tests. |
| 169 | |
| 170 | Returns: |
| 171 | An :class:`IcpHookConfig` with all fields validated. |
| 172 | """ |
| 173 | env_map = env if env is not None else dict(os.environ) |
| 174 | raw_enabled = env_map.get("MUSE_ICP_ENABLED", "1").strip().lower() |
| 175 | enabled = raw_enabled not in ("0", "false", "no", "off") |
| 176 | |
| 177 | raw_canister = env_map.get("MUSE_ICP_CANISTER_ID", "").strip() |
| 178 | if raw_canister and not _is_valid_canister_id(raw_canister): |
| 179 | logger.warning( |
| 180 | "knowtation-icp: MUSE_ICP_CANISTER_ID=%r is not a valid canister ID; " |
| 181 | "falling back to default %s.", |
| 182 | raw_canister, DEFAULT_CANISTER_ID, |
| 183 | ) |
| 184 | canister_id = DEFAULT_CANISTER_ID |
| 185 | else: |
| 186 | canister_id = raw_canister or DEFAULT_CANISTER_ID |
| 187 | |
| 188 | pem_path: str | None = None |
| 189 | raw_pem = env_map.get("MUSE_ICP_IDENTITY_PEM", "").strip() |
| 190 | if raw_pem: |
| 191 | candidate = pathlib.Path(raw_pem) |
| 192 | if candidate.exists() and candidate.is_file(): |
| 193 | pem_path = str(candidate) |
| 194 | else: |
| 195 | logger.warning( |
| 196 | "knowtation-icp: MUSE_ICP_IDENTITY_PEM=%r does not exist; " |
| 197 | "ICP calls will be read-only.", |
| 198 | raw_pem, |
| 199 | ) |
| 200 | |
| 201 | return IcpHookConfig( |
| 202 | enabled=enabled, |
| 203 | canister_id=canister_id, |
| 204 | identity_pem_path=pem_path, |
| 205 | ) |
| 206 | |
| 207 | |
| 208 | # --------------------------------------------------------------------------- |
| 209 | # Backoff math (exposed for unit testing) |
| 210 | # --------------------------------------------------------------------------- |
| 211 | |
| 212 | |
| 213 | def compute_backoff_delay( |
| 214 | attempt: int, |
| 215 | *, |
| 216 | base: float = RETRY_BASE_DELAY, |
| 217 | jitter_pct: float = RETRY_JITTER_PCT, |
| 218 | rng: random.Random | None = None, |
| 219 | ) -> float: |
| 220 | """Compute the exponential-backoff delay for retry *attempt* (1-indexed). |
| 221 | |
| 222 | Pure function — exposed so unit tests can verify the math against the |
| 223 | documented contract: ``BASE * 2**(attempt-1) ± jitter_pct``. |
| 224 | |
| 225 | Args: |
| 226 | attempt: 1-indexed attempt number (1 = first retry). |
| 227 | base: Base delay in seconds (default :data:`RETRY_BASE_DELAY`). |
| 228 | jitter_pct: Symmetric jitter as a fraction of the base delay |
| 229 | (default :data:`RETRY_JITTER_PCT`). |
| 230 | rng: Optional :class:`random.Random` instance for deterministic |
| 231 | tests. When ``None``, uses the global RNG. |
| 232 | |
| 233 | Returns: |
| 234 | Delay in seconds. Always non-negative; the lower jitter bound is |
| 235 | clamped at zero to avoid sleeping for negative time. |
| 236 | |
| 237 | Raises: |
| 238 | ValueError: If *attempt* < 1 or jitter_pct < 0 or > 1. |
| 239 | """ |
| 240 | if attempt < 1: |
| 241 | raise ValueError(f"attempt must be >= 1 (got {attempt})") |
| 242 | if jitter_pct < 0 or jitter_pct > 1: |
| 243 | raise ValueError(f"jitter_pct must be in [0, 1] (got {jitter_pct})") |
| 244 | base_delay = base * (2 ** (attempt - 1)) |
| 245 | jitter_range = base_delay * jitter_pct |
| 246 | rand = rng if rng is not None else random |
| 247 | jitter = rand.uniform(-jitter_range, jitter_range) |
| 248 | return max(0.0, base_delay + jitter) |
| 249 | |
| 250 | |
| 251 | # --------------------------------------------------------------------------- |
| 252 | # Canister HTTP client (read-only fallback when ic-py is unavailable) |
| 253 | # --------------------------------------------------------------------------- |
| 254 | |
| 255 | |
| 256 | def canister_http_get_url(canister_id: str, path: str) -> str: |
| 257 | """Build an https URL to query the canister's HTTP interface. |
| 258 | |
| 259 | Args: |
| 260 | canister_id: Validated canister ID. MUST have already passed |
| 261 | :func:`_is_valid_canister_id`. |
| 262 | path: Path part starting with ``/`` (e.g. ``"/attest/abc"``). |
| 263 | |
| 264 | Returns: |
| 265 | ``https://<canister_id>.ic0.app<path>``. |
| 266 | |
| 267 | Raises: |
| 268 | ValueError: If *canister_id* is invalid or *path* does not start |
| 269 | with ``/``. |
| 270 | """ |
| 271 | if not _is_valid_canister_id(canister_id): |
| 272 | raise ValueError(f"invalid canister id: {canister_id!r}") |
| 273 | if not path.startswith("/"): |
| 274 | raise ValueError(f"path must start with /: {path!r}") |
| 275 | return f"https://{canister_id}.{IC_HTTP_GATEWAY}{path}" |
| 276 | |
| 277 | |
| 278 | def fetch_attestation_via_http( |
| 279 | canister_id: str, |
| 280 | attestation_id: str, |
| 281 | *, |
| 282 | timeout: float = HTTP_TIMEOUT_SECONDS, |
| 283 | ) -> dict[str, Any] | None: |
| 284 | """Fetch a single attestation record via the canister's HTTP query interface. |
| 285 | |
| 286 | Returns the parsed JSON record on HTTP 200, ``None`` on HTTP 404, and |
| 287 | raises on any other transport-level failure. |
| 288 | |
| 289 | Args: |
| 290 | canister_id: Validated canister ID. |
| 291 | attestation_id: 64-char lowercase hex SHA-256 ID. |
| 292 | timeout: Per-attempt HTTP timeout (seconds). |
| 293 | |
| 294 | Returns: |
| 295 | Parsed record dict on 200, ``None`` on 404. |
| 296 | |
| 297 | Raises: |
| 298 | OSError: For any transport-level failure (connection refused, |
| 299 | timeout, DNS error). Caller decides whether to retry. |
| 300 | """ |
| 301 | url = canister_http_get_url(canister_id, f"/attest/{attestation_id}") |
| 302 | request = urllib.request.Request( |
| 303 | url=url, method="GET", |
| 304 | headers={"Accept": "application/json", "User-Agent": "muse-icp-hook/1.0"}, |
| 305 | ) |
| 306 | try: |
| 307 | with urllib.request.urlopen(request, timeout=timeout) as response: # noqa: S310 |
| 308 | status = getattr(response, "status", 200) |
| 309 | if status == 404: |
| 310 | return None |
| 311 | if status >= 400: |
| 312 | raise OSError(f"canister returned HTTP {status}") |
| 313 | raw = response.read(65_536) |
| 314 | except urllib.error.HTTPError as exc: |
| 315 | if exc.code == 404: |
| 316 | return None |
| 317 | raise OSError(f"canister returned HTTP {exc.code}") from exc |
| 318 | except urllib.error.URLError as exc: |
| 319 | raise OSError(f"canister unreachable: {exc}") from exc |
| 320 | parsed = json.loads(raw.decode("utf-8")) |
| 321 | return parsed if isinstance(parsed, dict) else None |
| 322 | |
| 323 | |
| 324 | # --------------------------------------------------------------------------- |
| 325 | # Update-call shim (ic-py preferred, no-op fallback) |
| 326 | # --------------------------------------------------------------------------- |
| 327 | |
| 328 | |
| 329 | @dataclass(frozen=True) |
| 330 | class StoreResult: |
| 331 | """Result of one storeAttestation update call. |
| 332 | |
| 333 | Attributes: |
| 334 | anchored: ``True`` iff the canister returned ``#ok`` OR returned |
| 335 | the documented "already exists" error (treated as success |
| 336 | because the canister is immutable and a duplicate ID means |
| 337 | the record is already anchored). |
| 338 | reason: Short token describing the outcome — ``"ok"``, |
| 339 | ``"already_exists"``, ``"unauthorized"``, ``"error"``. |
| 340 | message: Human-readable detail (used in logs). |
| 341 | """ |
| 342 | |
| 343 | anchored: bool |
| 344 | reason: str |
| 345 | message: str |
| 346 | |
| 347 | |
| 348 | def _store_attestation_via_icpy( |
| 349 | canister_id: str, |
| 350 | record: dict[str, Any], |
| 351 | pem_path: str | None, |
| 352 | ) -> StoreResult: |
| 353 | """Attempt to call storeAttestation via ic-py. |
| 354 | |
| 355 | Args: |
| 356 | canister_id: Validated canister ID. |
| 357 | record: Attestation record dict. |
| 358 | pem_path: Path to identity PEM, or ``None`` for anonymous. |
| 359 | |
| 360 | Returns: |
| 361 | A :class:`StoreResult` describing the outcome. |
| 362 | |
| 363 | Raises: |
| 364 | ImportError: If ic-py is not installed — caller should fall back. |
| 365 | OSError: For transport-level failures. |
| 366 | """ |
| 367 | try: |
| 368 | from ic.client import Client |
| 369 | from ic.identity import Identity |
| 370 | from ic.agent import Agent |
| 371 | from ic.candid import encode, Types |
| 372 | except ImportError as exc: |
| 373 | raise ImportError("ic-py not installed") from exc |
| 374 | |
| 375 | if pem_path: |
| 376 | identity = Identity.from_pem(pem_path) # type: ignore[attr-defined] |
| 377 | else: |
| 378 | identity = Identity() |
| 379 | client = Client() |
| 380 | agent = Agent(identity, client) |
| 381 | |
| 382 | try: |
| 383 | # Build the StoreInput record matching the Motoko type. |
| 384 | params = [{ |
| 385 | "type": Types.Record({ |
| 386 | "id": Types.Text, |
| 387 | "action": Types.Text, |
| 388 | "path": Types.Text, |
| 389 | "timestamp": Types.Text, |
| 390 | "content_hash": Types.Text, |
| 391 | "sig": Types.Text, |
| 392 | }), |
| 393 | "value": { |
| 394 | "id": str(record.get("id", "")), |
| 395 | "action": str(record.get("action", "")), |
| 396 | "path": str(record.get("path", "")), |
| 397 | "timestamp": str(record.get("timestamp", "")), |
| 398 | "content_hash": str(record.get("content_hash", "")), |
| 399 | "sig": str(record.get("sig", "")), |
| 400 | }, |
| 401 | }] |
| 402 | encoded = encode(params) |
| 403 | raw = agent.update_raw(canister_id, "storeAttestation", encoded) |
| 404 | except Exception as exc: # noqa: BLE001 |
| 405 | raise OSError(f"ic-py update_raw failed: {exc}") from exc |
| 406 | |
| 407 | text = repr(raw).lower() |
| 408 | if "ok" in text and "err" not in text: |
| 409 | return StoreResult(anchored=True, reason="ok", message=str(raw)) |
| 410 | if "already exists" in text: |
| 411 | return StoreResult( |
| 412 | anchored=True, reason="already_exists", |
| 413 | message="canister returned 'already exists' (idempotent success)", |
| 414 | ) |
| 415 | if "unauthorized" in text: |
| 416 | return StoreResult( |
| 417 | anchored=False, reason="unauthorized", |
| 418 | message=f"caller not in authorized list: {raw!r}", |
| 419 | ) |
| 420 | return StoreResult(anchored=False, reason="error", message=str(raw)) |
| 421 | |
| 422 | |
| 423 | # --------------------------------------------------------------------------- |
| 424 | # Main hook |
| 425 | # --------------------------------------------------------------------------- |
| 426 | |
| 427 | |
| 428 | class KnowtationICPPushHook: |
| 429 | """Anchor knowtation attestation records on the ICP canister after push. |
| 430 | |
| 431 | Implements :class:`muse.core.push_hooks.PushHookPlugin`. Reads |
| 432 | ``commit.metadata["attestation"]`` from each newly-pushed commit and |
| 433 | attempts to anchor it on the canister. Failures at any layer |
| 434 | (network, auth, malformed records) never propagate to the push |
| 435 | command — they are surfaced via :class:`PushHookResult`. |
| 436 | |
| 437 | The hook is testable via :func:`KnowtationICPPushHook.run_post_push_sync` |
| 438 | which performs the same work synchronously (no thread spawn) so test |
| 439 | assertions can read the result deterministically. |
| 440 | """ |
| 441 | |
| 442 | def __init__(self) -> None: |
| 443 | """Initialise the hook. No I/O at construction time.""" |
| 444 | # Per-process tracking of in-flight anchor threads — used by tests |
| 445 | # that need to .join() on completion. |
| 446 | self._threads: list[threading.Thread] = [] |
| 447 | self._threads_lock = threading.Lock() |
| 448 | |
| 449 | # ------------------------------------------------------------------ |
| 450 | # PushHookPlugin contract |
| 451 | # ------------------------------------------------------------------ |
| 452 | |
| 453 | def run_post_push( |
| 454 | self, |
| 455 | repo_path: pathlib.Path, |
| 456 | push_result: dict[str, object], |
| 457 | domain: str, |
| 458 | config: dict[str, object] | None = None, |
| 459 | ) -> PushHookResult: |
| 460 | """Dispatch the anchor work and return a status record immediately. |
| 461 | |
| 462 | Spawns a daemon thread that calls :meth:`_anchor_pushed_commits` and |
| 463 | returns a :class:`PushHookResult` with ``out_of_band=True`` describing |
| 464 | the dispatch. The merge engine — sorry, push command — does not wait |
| 465 | for the background thread. |
| 466 | |
| 467 | Args: |
| 468 | repo_path: Repository root. |
| 469 | push_result: Push result dict. Reads ``commits`` (list of newly |
| 470 | pushed commit IDs). Other keys ignored. |
| 471 | domain: Domain string (always ``"knowtation"`` here). |
| 472 | config: Optional config dict. Reads ``"icp"`` sub-dict. |
| 473 | |
| 474 | Returns: |
| 475 | A :class:`PushHookResult` describing the dispatch — never raises. |
| 476 | """ |
| 477 | del domain, config # accepted for protocol conformance |
| 478 | cfg = load_icp_config() |
| 479 | if not cfg.enabled: |
| 480 | return PushHookResult( |
| 481 | hook_name=HOOK_NAME, |
| 482 | fired=False, |
| 483 | out_of_band=False, |
| 484 | skipped=True, |
| 485 | reason="DISABLED", |
| 486 | message="MUSE_ICP_ENABLED=0 — anchoring skipped", |
| 487 | error=None, |
| 488 | ) |
| 489 | |
| 490 | commit_ids_raw = push_result.get("commits", []) |
| 491 | if not isinstance(commit_ids_raw, list) or not commit_ids_raw: |
| 492 | return PushHookResult( |
| 493 | hook_name=HOOK_NAME, |
| 494 | fired=False, |
| 495 | out_of_band=False, |
| 496 | skipped=True, |
| 497 | reason="NO_COMMITS", |
| 498 | message="push_result has no commits — nothing to anchor", |
| 499 | error=None, |
| 500 | ) |
| 501 | |
| 502 | commit_ids = [str(c) for c in commit_ids_raw] |
| 503 | |
| 504 | def _worker() -> None: |
| 505 | try: |
| 506 | self._anchor_pushed_commits(repo_path, commit_ids, cfg) |
| 507 | except Exception as exc: # noqa: BLE001 — must not propagate |
| 508 | logger.warning( |
| 509 | "knowtation-icp: background anchor thread crashed: %s: %s", |
| 510 | type(exc).__name__, exc, |
| 511 | ) |
| 512 | |
| 513 | thread = threading.Thread( |
| 514 | target=_worker, name="knowtation-icp-anchor", daemon=True, |
| 515 | ) |
| 516 | thread.start() |
| 517 | with self._threads_lock: |
| 518 | self._threads.append(thread) |
| 519 | |
| 520 | return PushHookResult( |
| 521 | hook_name=HOOK_NAME, |
| 522 | fired=True, |
| 523 | out_of_band=True, |
| 524 | skipped=False, |
| 525 | reason="", |
| 526 | message=( |
| 527 | f"dispatched ICP anchor for {len(commit_ids)} commit(s) " |
| 528 | f"to {cfg.canister_id} (background)" |
| 529 | ), |
| 530 | error=None, |
| 531 | ) |
| 532 | |
| 533 | # ------------------------------------------------------------------ |
| 534 | # Synchronous variant for tests + the actual anchor logic |
| 535 | # ------------------------------------------------------------------ |
| 536 | |
| 537 | def run_post_push_sync( |
| 538 | self, |
| 539 | repo_path: pathlib.Path, |
| 540 | push_result: dict[str, object], |
| 541 | config: IcpHookConfig | None = None, |
| 542 | ) -> PushHookResult: |
| 543 | """Synchronous variant — performs anchoring inline and returns the count. |
| 544 | |
| 545 | Useful in tests that assert on the anchored/failed counters. Production |
| 546 | code paths use :meth:`run_post_push` which dispatches asynchronously. |
| 547 | |
| 548 | Args: |
| 549 | repo_path: Repository root. |
| 550 | push_result: Push result dict (see :meth:`run_post_push`). |
| 551 | config: Optional explicit :class:`IcpHookConfig`. When |
| 552 | ``None``, loads from environment variables. |
| 553 | |
| 554 | Returns: |
| 555 | A :class:`PushHookResult` with populated ``anchored_count`` and |
| 556 | ``failed_count`` fields. |
| 557 | """ |
| 558 | cfg = config or load_icp_config() |
| 559 | if not cfg.enabled: |
| 560 | return PushHookResult( |
| 561 | hook_name=HOOK_NAME, fired=False, out_of_band=False, |
| 562 | skipped=True, reason="DISABLED", |
| 563 | message="MUSE_ICP_ENABLED=0 — anchoring skipped", |
| 564 | error=None, |
| 565 | ) |
| 566 | commit_ids = [str(c) for c in (push_result.get("commits") or [])] |
| 567 | if not commit_ids: |
| 568 | return PushHookResult( |
| 569 | hook_name=HOOK_NAME, fired=False, out_of_band=False, |
| 570 | skipped=True, reason="NO_COMMITS", |
| 571 | message="push_result has no commits", error=None, |
| 572 | ) |
| 573 | anchored, failed = self._anchor_pushed_commits(repo_path, commit_ids, cfg) |
| 574 | return PushHookResult( |
| 575 | hook_name=HOOK_NAME, |
| 576 | fired=True, |
| 577 | out_of_band=False, |
| 578 | skipped=False, |
| 579 | reason="", |
| 580 | message=f"anchored {anchored}, failed {failed}", |
| 581 | error=None, |
| 582 | anchored_count=anchored, |
| 583 | failed_count=failed, |
| 584 | ) |
| 585 | |
| 586 | def wait_for_inflight(self, timeout: float = 30.0) -> None: |
| 587 | """Block until every dispatched anchor thread has joined. |
| 588 | |
| 589 | Useful in tests after :meth:`run_post_push` to ensure the background |
| 590 | work has completed before asserting on side effects. Production code |
| 591 | does not call this — the daemon threads are fire-and-forget. |
| 592 | |
| 593 | Args: |
| 594 | timeout: Max seconds to wait for any single thread. |
| 595 | """ |
| 596 | with self._threads_lock: |
| 597 | threads = list(self._threads) |
| 598 | for t in threads: |
| 599 | t.join(timeout=timeout) |
| 600 | |
| 601 | # ------------------------------------------------------------------ |
| 602 | # Internal: walk commits, extract attestations, anchor each. |
| 603 | # ------------------------------------------------------------------ |
| 604 | |
| 605 | def _anchor_pushed_commits( |
| 606 | self, |
| 607 | repo_path: pathlib.Path, |
| 608 | commit_ids: list[str], |
| 609 | cfg: IcpHookConfig, |
| 610 | ) -> tuple[int, int]: |
| 611 | """Anchor every attestation found in *commit_ids*. Returns (ok, fail). |
| 612 | |
| 613 | Reads each commit via :func:`muse.core.store.read_commit`, extracts |
| 614 | ``metadata["attestation"]`` (canonical JSON), and calls |
| 615 | :meth:`_anchor_one_record` with retry logic. |
| 616 | |
| 617 | Args: |
| 618 | repo_path: Repository root. |
| 619 | commit_ids: Newly-pushed commit IDs. |
| 620 | cfg: Loaded :class:`IcpHookConfig`. |
| 621 | |
| 622 | Returns: |
| 623 | ``(anchored_count, failed_count)`` — the number of records |
| 624 | successfully anchored and the number that failed after all |
| 625 | retries. Records without an attestation field do not count |
| 626 | toward either. |
| 627 | """ |
| 628 | try: |
| 629 | from muse.core.store import read_commit |
| 630 | except ImportError: |
| 631 | logger.warning( |
| 632 | "knowtation-icp: muse.core.store unavailable; cannot read commits." |
| 633 | ) |
| 634 | return 0, 0 |
| 635 | |
| 636 | anchored = 0 |
| 637 | failed = 0 |
| 638 | for cid in commit_ids: |
| 639 | try: |
| 640 | commit = read_commit(repo_path, cid) |
| 641 | except Exception as exc: # noqa: BLE001 |
| 642 | logger.warning( |
| 643 | "knowtation-icp: cannot read commit %s: %s", cid[:12], exc |
| 644 | ) |
| 645 | failed += 1 |
| 646 | continue |
| 647 | if commit is None: |
| 648 | continue |
| 649 | blob = (commit.metadata or {}).get("attestation") |
| 650 | if not blob: |
| 651 | continue |
| 652 | try: |
| 653 | record = json.loads(blob) |
| 654 | except json.JSONDecodeError as exc: |
| 655 | logger.warning( |
| 656 | "knowtation-icp: commit %s has malformed attestation JSON: %s", |
| 657 | cid[:12], exc, |
| 658 | ) |
| 659 | failed += 1 |
| 660 | continue |
| 661 | if not isinstance(record, dict): |
| 662 | failed += 1 |
| 663 | continue |
| 664 | ok = self._anchor_one_record(cfg, record) |
| 665 | if ok: |
| 666 | anchored += 1 |
| 667 | else: |
| 668 | failed += 1 |
| 669 | return anchored, failed |
| 670 | |
| 671 | def _anchor_one_record( |
| 672 | self, |
| 673 | cfg: IcpHookConfig, |
| 674 | record: dict[str, Any], |
| 675 | ) -> bool: |
| 676 | """Anchor *record* on the canister with retry; return True on success. |
| 677 | |
| 678 | Retry policy: :data:`MAX_RETRIES` attempts with exponential backoff |
| 679 | and ±10 % jitter (per :func:`compute_backoff_delay`). |
| 680 | |
| 681 | Args: |
| 682 | cfg: Loaded :class:`IcpHookConfig`. |
| 683 | record: Attestation record dict. |
| 684 | |
| 685 | Returns: |
| 686 | ``True`` on success (including the idempotent "already exists" |
| 687 | path). ``False`` after all retries exhausted. |
| 688 | """ |
| 689 | last_error: str = "" |
| 690 | for attempt in range(1, cfg.retries + 1): |
| 691 | try: |
| 692 | result = _store_attestation_via_icpy( |
| 693 | cfg.canister_id, record, cfg.identity_pem_path, |
| 694 | ) |
| 695 | except ImportError: |
| 696 | # ic-py not installed — fall back to read-only verification. |
| 697 | # We can only verify; we cannot anchor. Log and treat as |
| 698 | # NOT anchored so the push hook reports skipped. |
| 699 | logger.info( |
| 700 | "knowtation-icp: ic-py not installed — skipping update call " |
| 701 | "for record %s. Install with `pip install ic-py` to enable " |
| 702 | "authenticated anchoring.", |
| 703 | record.get("id", "")[:12], |
| 704 | ) |
| 705 | return False |
| 706 | except OSError as exc: |
| 707 | last_error = str(exc) |
| 708 | logger.info( |
| 709 | "knowtation-icp: anchor attempt %d/%d failed for %s: %s", |
| 710 | attempt, cfg.retries, record.get("id", "")[:12], exc, |
| 711 | ) |
| 712 | if attempt < cfg.retries: |
| 713 | delay = compute_backoff_delay(attempt) |
| 714 | time.sleep(delay) |
| 715 | continue |
| 716 | |
| 717 | if result.anchored: |
| 718 | logger.debug( |
| 719 | "knowtation-icp: anchored %s (%s)", |
| 720 | record.get("id", "")[:12], result.reason, |
| 721 | ) |
| 722 | return True |
| 723 | if result.reason == "unauthorized": |
| 724 | logger.warning( |
| 725 | "knowtation-icp: unauthorized caller for %s — record NOT " |
| 726 | "anchored. Configure MUSE_ICP_IDENTITY_PEM with an " |
| 727 | "authorised principal to enable anchoring.", |
| 728 | record.get("id", "")[:12], |
| 729 | ) |
| 730 | return False # no point retrying auth failures |
| 731 | last_error = result.message |
| 732 | if attempt < cfg.retries: |
| 733 | time.sleep(compute_backoff_delay(attempt)) |
| 734 | logger.warning( |
| 735 | "knowtation-icp: failed to anchor %s after %d attempts: %s", |
| 736 | record.get("id", "")[:12], cfg.retries, last_error, |
| 737 | ) |
| 738 | return False |
| 739 | |
| 740 | |
| 741 | # --------------------------------------------------------------------------- |
| 742 | # Module-level registration helper. |
| 743 | # --------------------------------------------------------------------------- |
| 744 | |
| 745 | |
| 746 | def register_knowtation_icp_push_hook() -> KnowtationICPPushHook: |
| 747 | """Construct + register the ICP push hook; return the singleton instance. |
| 748 | |
| 749 | Idempotent: re-calling replaces any prior registration with a fresh |
| 750 | instance. |
| 751 | |
| 752 | Returns: |
| 753 | The newly-registered :class:`KnowtationICPPushHook` instance. |
| 754 | """ |
| 755 | hook = KnowtationICPPushHook() |
| 756 | get_push_hook_registry().register("knowtation", hook) |
| 757 | return hook |
| 758 | |
| 759 | |
| 760 | __all__ = [ |
| 761 | "DEFAULT_CANISTER_ID", |
| 762 | "HOOK_NAME", |
| 763 | "HTTP_TIMEOUT_SECONDS", |
| 764 | "IC_HTTP_GATEWAY", |
| 765 | "IcpHookConfig", |
| 766 | "KnowtationICPPushHook", |
| 767 | "MAX_RETRIES", |
| 768 | "RETRY_BASE_DELAY", |
| 769 | "RETRY_JITTER_PCT", |
| 770 | "StoreResult", |
| 771 | "canister_http_get_url", |
| 772 | "compute_backoff_delay", |
| 773 | "fetch_attestation_via_http", |
| 774 | "load_icp_config", |
| 775 | "register_knowtation_icp_push_hook", |
| 776 | ] |
File History
1 commit
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8
docs(KD-STAGING): sync governance after KD-6b DONE
Human
14 days ago