"""Knowtation identity / org-quorum provider — Phase 7.7. Identity-record resolver and helpers for org-quorum signing in the Knowtation domain. Implements :class:`muse.core.attestation.MuseIdentityProvider` backed by a local JSON cache under ``/.muse/identity/.json`` (falling back to ``$MUSE_IDENTITY_DIR``). In production deployments the cache is hydrated by the MuseHub identity API; in tests the cache is populated directly. Identity record schema ---------------------- Identity records are stored as JSON files keyed by handle:: { "handle": "@my-org", "kind": "org", "public_key": "", (orgs have no key) "members": ["alice", "bob", "carol"], "quorum_threshold": 2, "spawned_by": "@root-org", (optional) "revoked_at": "" (empty = active) } For human/agent records the ``public_key`` is a base64url-encoded raw Ed25519 public key (32 bytes → 43 chars no padding). ``members`` and ``quorum_threshold`` MUST be absent or empty for non-org records. Threat model ------------ The cycle-detection step in :func:`muse.core.verify.verify_identity_chain` is **defensive in depth**. The identity domain at staging.musehub.ai already rejects cycles at write time, but a stale or tampered local cache could re-introduce one. The verifier walks with a depth-bounded BFS and a seen-set so any cycle is reported as ``IdentityChainResult.cycle_detected=True`` and the chain is marked invalid — never silently terminates the walk. """ from __future__ import annotations import json import logging import os import pathlib import re from typing import Any from muse.core.attestation import ( AttestationVerifyError, IdentityRecord, MuseIdentityProvider, register_identity_provider, ) logger = logging.getLogger(__name__) PROVIDER_DOMAIN: str = "knowtation" """Domain string this provider registers under. Same as the HMAC attestation provider so a single domain owns both identity and attestation concerns; the registries are separate so this is purely conventional.""" #: Maximum identity-record file size (bytes). 16 KiB is generous for a #: handful of fields with member lists; anything larger is treated as #: corrupt. _MAX_RECORD_BYTES: int = 16_384 #: Maximum number of members allowed in a single org record. Bounds the #: cost of a quorum walk and prevents DoS via a record with millions of #: members. Keep loose enough to support real-world orgs (≤ 1024). _MAX_MEMBERS: int = 1024 #: Identity handle character set. Allows alphanumerics, dots, dashes, #: underscores; org handles also allow a leading ``@``. This excludes #: filesystem path-traversal characters (``/``, ``\\``, ``..``, NUL). _HANDLE_RE: re.Pattern[str] = re.compile(r"\A@?[a-zA-Z0-9_.\-]{1,128}\Z") def is_valid_handle(handle: object) -> bool: """Return True iff *handle* is a syntactically valid identity handle. Validates against :data:`_HANDLE_RE` so handles cannot contain path separators, embedded NULs, or other characters that would let a malicious handle escape the identity directory at lookup time. Additional defensive rules layered on top of the regex: * The body (after a leading ``@``) MUST NOT be ``.`` or ``..``. * The body MUST NOT start with ``.`` (would render hidden files / special FS entries). * The body MUST NOT contain ``..`` as a substring (path-segment traversal even when ``/`` is whitelisted out). Args: handle: Candidate handle (any type — non-strings always return ``False``). Returns: ``True`` when the handle matches the strict character whitelist; ``False`` otherwise. """ if not isinstance(handle, str) or not _HANDLE_RE.match(handle): return False body = handle[1:] if handle.startswith("@") else handle if body in (".", "..") or body.startswith(".") or ".." in body: return False return True def _resolve_identity_dir(repo_root: pathlib.Path | None) -> pathlib.Path: """Resolve the directory holding identity-record JSON files. Resolution order: 1. ``$MUSE_IDENTITY_DIR`` environment variable (absolute path). 2. ``/.muse/identity/`` when ``repo_root`` is supplied. 3. ``~/.muse/identity/`` as the user-global default. Args: repo_root: Optional repo root for the per-repo cache. Returns: :class:`pathlib.Path` to the directory. Existence is NOT enforced — a missing directory simply means "no identities cached". """ env_dir = os.environ.get("MUSE_IDENTITY_DIR", "").strip() if env_dir: return pathlib.Path(env_dir) if repo_root is not None: return repo_root / ".muse" / "identity" return pathlib.Path.home() / ".muse" / "identity" def _handle_to_filename(handle: str) -> str: """Encode *handle* into a safe filename. Replaces the leading ``@`` with ``_at_`` so org handles round-trip on case-insensitive filesystems (macOS HFS+ default) without colliding with single-actor handles. Args: handle: Validated identity handle. Returns: Filename string (no extension); callers append ``.json``. """ return ("_at_" + handle[1:]) if handle.startswith("@") else handle def _normalise_record(raw: Any, handle: str) -> IdentityRecord: """Coerce a parsed JSON object into an :class:`IdentityRecord`. Performs strict shape validation: * Top-level must be a dict. * ``handle`` (when present) must equal the lookup handle exactly — mismatches raise to surface tampering of the JSON filename ↔ content relationship. * ``kind`` must be one of ``human|agent|org``. * ``members`` must be a list of valid handles, length ≤ ``_MAX_MEMBERS``. * ``quorum_threshold`` must be an int in ``[1, len(members)]`` for org records; absent/zero for non-org. Args: raw: Parsed JSON value from the identity file. handle: The handle that was looked up (used to cross-check the ``handle`` field on the parsed object). Returns: A typed :class:`IdentityRecord` with all consumed fields populated. Raises: AttestationVerifyError: For any shape violation. Distinct from "record not found" (which returns ``None`` to the caller of :meth:`KnowtationFsIdentityProvider.get_identity`). """ if not isinstance(raw, dict): raise AttestationVerifyError( f"identity record for {handle!r} is not a JSON object", code="MALFORMED_IDENTITY", ) rec_handle = raw.get("handle", handle) if rec_handle != handle: raise AttestationVerifyError( f"identity file content handle {rec_handle!r} does not match " f"lookup handle {handle!r} — possible tampering", code="HANDLE_MISMATCH", ) kind = raw.get("kind", "") if kind not in ("human", "agent", "org"): raise AttestationVerifyError( f"identity {handle!r}: kind must be 'human'|'agent'|'org' " f"(got {kind!r})", code="INVALID_KIND", ) record: IdentityRecord = { "handle": handle, "kind": kind, "public_key": str(raw.get("public_key", "") or ""), "members": [], "quorum_threshold": 0, "spawned_by": str(raw.get("spawned_by", "") or ""), "revoked_at": str(raw.get("revoked_at", "") or ""), } if kind == "org": members = raw.get("members", []) if not isinstance(members, list): raise AttestationVerifyError( f"org {handle!r}: members must be a list", code="MALFORMED_IDENTITY", ) if len(members) > _MAX_MEMBERS: raise AttestationVerifyError( f"org {handle!r}: members list exceeds {_MAX_MEMBERS} entries", code="MEMBERS_TOO_LARGE", ) for m in members: if not is_valid_handle(m): raise AttestationVerifyError( f"org {handle!r}: member {m!r} is not a valid handle", code="MALFORMED_IDENTITY", ) record["members"] = list(members) threshold = raw.get("quorum_threshold", 0) if not isinstance(threshold, int) or threshold < 1: raise AttestationVerifyError( f"org {handle!r}: quorum_threshold must be an int >= 1 " f"(got {threshold!r})", code="INVALID_THRESHOLD", ) if threshold > len(members): raise AttestationVerifyError( f"org {handle!r}: quorum_threshold {threshold} exceeds " f"member count {len(members)}", code="INVALID_THRESHOLD", ) record["quorum_threshold"] = int(threshold) return record class KnowtationFsIdentityProvider: """Filesystem-backed identity provider for the Knowtation domain. Reads identity records from JSON files under :func:`_resolve_identity_dir`. Suitable for both local development (records hand-written for tests) and production deployments where a sync daemon hydrates the cache from MuseHub's identity API. Thread-safety: stateless — each :meth:`get_identity` call performs an independent file-system read. Multiple threads can call concurrently without coordination. Attributes: repo_root: Optional repo root used by :func:`_resolve_identity_dir` when ``$MUSE_IDENTITY_DIR`` is unset. ``None`` falls back to the user-global directory. """ repo_root: pathlib.Path | None def __init__(self, repo_root: pathlib.Path | None = None) -> None: """Construct a provider rooted at *repo_root* (or user-global). Args: repo_root: Optional path to the repo whose ``.muse/identity/`` directory should be searched first. When ``None``, only ``$MUSE_IDENTITY_DIR`` and ``~/.muse/identity/`` are consulted. """ self.repo_root = repo_root def get_identity(self, handle: str) -> IdentityRecord | None: """Read the identity record for *handle*, or return ``None``. Walks ``/.muse/identity/.json`` first then falls back to ``$MUSE_IDENTITY_DIR``/``~/.muse/identity/.json``. Args: handle: Identity handle to resolve. Returns: :class:`IdentityRecord` when the file exists and parses; ``None`` when no record was found in any of the searched directories. Raises: AttestationVerifyError: For malformed JSON, oversized files, or shape violations. Distinct from "not found" (returns ``None``). Callers must surface as ``partial=True``. """ if not is_valid_handle(handle): raise AttestationVerifyError( f"identity handle {handle!r} fails strict validation; " f"rejecting to prevent path traversal", code="INVALID_HANDLE", ) filename = _handle_to_filename(handle) + ".json" candidates: list[pathlib.Path] = [] if self.repo_root is not None: candidates.append(self.repo_root / ".muse" / "identity" / filename) candidates.append(_resolve_identity_dir(self.repo_root) / filename) for path in candidates: try: if not path.is_file(): continue size = path.stat().st_size if size > _MAX_RECORD_BYTES: raise AttestationVerifyError( f"identity record {path} exceeds {_MAX_RECORD_BYTES}" f" bytes ({size}); refusing to read", code="RECORD_TOO_LARGE", ) raw_bytes = path.read_bytes() except OSError as exc: raise AttestationVerifyError( f"cannot read identity record {path}: {exc}", code="IO_ERROR", ) from exc try: parsed = json.loads(raw_bytes.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise AttestationVerifyError( f"identity record {path} is not valid JSON: {exc}", code="MALFORMED_IDENTITY", ) from exc return _normalise_record(parsed, handle) return None def register_knowtation_identity_provider( repo_root: pathlib.Path | None = None, ) -> KnowtationFsIdentityProvider: """Construct + register a :class:`KnowtationFsIdentityProvider`. Idempotent — a subsequent call replaces the prior registration with a new instance scoped to *repo_root*. Args: repo_root: Optional repo root for per-repo identity caches. Returns: The newly-registered provider instance. """ provider = KnowtationFsIdentityProvider(repo_root=repo_root) register_identity_provider(PROVIDER_DOMAIN, provider) return provider # --------------------------------------------------------------------------- # Quorum metadata helpers — used by both the commit and verify code paths # --------------------------------------------------------------------------- def build_quorum_metadata( org_handle: str, threshold: int, signers: list[dict[str, str]], ) -> dict[str, object]: """Build the canonical ``commit.metadata['quorum']`` value. The shape is fixed across producers and verifiers so that the JSON serialisation is deterministic — important because the canonical metadata bytes are hashed into the parent commit graph downstream. Args: org_handle: Handle of the org commit was made on behalf of. MUST start with ``@`` (validated). threshold: Quorum threshold copied from the org's identity record at sign time — surfaced for audit trails (the verifier re-fetches the live threshold so a stripped-down threshold here cannot bypass quorum). signers: List of signer dicts. Each dict MUST have ``handle``, ``public_key`` (base64url no padding), and ``signature`` (base64url no padding). Returns: A dict ready to JSON-serialise into ``commit.metadata["quorum"]``. Raises: ValueError: For invalid org handle, non-positive threshold, or malformed signer entries. """ if not is_valid_handle(org_handle) or not org_handle.startswith("@"): raise ValueError(f"org_handle must start with '@': {org_handle!r}") if not isinstance(threshold, int) or threshold < 1: raise ValueError(f"threshold must be a positive int (got {threshold!r})") if not isinstance(signers, list) or not signers: raise ValueError("signers must be a non-empty list") cleaned: list[dict[str, str]] = [] seen: set[str] = set() for s in signers: if not isinstance(s, dict): raise ValueError("each signer entry must be a dict") h = s.get("handle", "") pk = s.get("public_key", "") sig = s.get("signature", "") if not is_valid_handle(h): raise ValueError(f"signer handle {h!r} is not valid") if h in seen: raise ValueError( f"signer handle {h!r} appears more than once — " "duplicate signatures cannot count toward quorum" ) seen.add(h) if not isinstance(pk, str) or not pk: raise ValueError(f"signer {h!r}: public_key must be a non-empty string") if not isinstance(sig, str) or not sig: raise ValueError(f"signer {h!r}: signature must be a non-empty string") cleaned.append({"handle": h, "public_key": pk, "signature": sig}) return { "org_handle": org_handle, "threshold": int(threshold), "signers": cleaned, } __all__ = [ "KnowtationFsIdentityProvider", "PROVIDER_DOMAIN", "build_quorum_metadata", "is_valid_handle", "register_knowtation_identity_provider", ]