feat: MCP server catalog core -- signed, validated, resolvable (#10, #61)

Phase 1 of the MCP server catalog: pure, GUI-free core functions plus the seed data/catalog.json (20 servers). No GUI wiring in this PR -- bcc.py is untouched; a follow-up PR adds the picker dialog. - load_catalog(): strict json.loads ONLY. The lenient repair pipeline (repair_json_text / parse_pasted_json*) is never used on catalog bytes, by design and by comment, so a signature always authenticates exactly what gets parsed. - validate_catalog(): rejects the whole file (not per-entry) on: bad schema/version types, missing tier-appropriate fields (basic needs config.command+args, link-only needs docs_url and no config), a command allowlist (npx/uvx/docker/node/python/python3 only), -e/--eval/-c denial for node/python, --privileged and root/$HOME volume-mount denial for docker, non-empty env_required values (hard rejection -- secrets never ship in the catalog), secret-looking args (reuses _TOKEN_PREFIXES/_is_secret_value rather than reimplementing), non-https URL fields, and non-ASCII code points in id/command/args (homoglyph defence). - verify_catalog_signature(): Ed25519 via the cryptography package, domain-separated message (the literal prefix "bcc-catalog-v1|" + raw bytes), accepts a match against any key in CATALOG_PUBKEYS (rotation-ready), never raises. - resolve_catalog(): picks the highest version among bundled/cached/remote candidates that EACH independently pass verify + validate -- the bundled catalog gets no implicit trust, closing the hole where an unsigned payload merged to main would win on being local. Anti-rollback (never regress below the best verified candidate already in hand) and anti-freeze (reject a jump of more than 1000 versions) built in. - catalog_entry_to_paste_json() / config_has_unfilled_placeholders(): small pure helpers the future GUI dialog will use to feed a catalog pick into the existing paste-import path and to gate Save on unfilled placeholder tokens. data/catalog.json: the provided 20-server seed, with a signed_at field added at the top level (lives inside the signed payload once real signing lands in #62). Wired into bcc.spec's PyInstaller datas so it bundles into the frozen app. Security requirements from the issue, and where they landed: - Catalog bytes never touch the lenient JSON repair path -- enforced by load_catalog()'s strict json.loads and a comment warning against wiring it in later. - env_required values are a hard rejection when non-empty, not a warning. - Secret-looking args are rejected at validation time, reusing the existing secret-detection helpers instead of duplicating them. - Non-ASCII id/command/args rejected (typosquat/homoglyph defence). - URL fields restricted to https://. - Ed25519 signature verification is domain-separated and never raises. - The bundled catalog is verified at runtime exactly like remote/cached -- no implicit trust for being local. - Anti-rollback and anti-freeze bounds on resolve_catalog's version comparison. Tests: 42 new tests added to tests/test_core.py (full suite: 239 passed, 1 pre-existing unrelated skip). ruff check and ruff format --check both clean.
2026-07-12 17:32:20 -04:00
parent e6b60e94e7
commit 48904c7787
5 changed files with 1254 additions and 1 deletions
@@ -32,6 +32,9 @@ from pathlib import Path
 from typing import NamedTuple
 from urllib.parse import urlparse

+from cryptography.exceptions import InvalidSignature
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
+
 CONFIG_FILENAME = "claude_desktop_config.json"

 # Disabled servers are parked under this non-standard key. Claude Desktop only
@@ -2143,3 +2146,396 @@ def restart_claude_desktop() -> RestartResult:
    if sys.platform.startswith("win"):
        return _restart_claude_desktop_windows()
    return RestartResult(False, "Restarting Claude Desktop isn't supported on this platform.")
+
+
+# --------------------------------------------------------------------------- #
+# MCP server catalog (issue #10 / #61)
+#
+# A curated, SIGNED list of ready-to-use MCP server definitions (bundled with
+# the app and, later, fetchable/cacheable — see follow-up issues). Every
+# function here is pure and defensive: catalog bytes may come from a fetch
+# over the network, a disk cache, or the copy frozen into the binary, and
+# all three are treated as equally untrusted until their signature verifies.
+# --------------------------------------------------------------------------- #
+
+# Commands a catalog entry's config is allowed to launch. Anything else
+# (bash, sh, curl, a raw script interpreter that isn't on this list, ...)
+# is rejected by validate_catalog() regardless of how plausible it looks.
+CATALOG_ALLOWED_COMMANDS = frozenset({"npx", "uvx", "docker", "node", "python", "python3"})
+
+# Ed25519 public keys allowed to sign a catalog, raw 32-byte form. A LIST
+# (not a single key) so keys can be rotated without bricking installs that
+# still trust an older key: verify_catalog_signature() accepts a match
+# against ANY key in this list.
+CATALOG_PUBKEYS: list[bytes] = [
+    b"\x00" * 32,  # TODO: real key from Catalog Console (#62)
+]
+
+# Domain-separation prefix for the signed message. The signature covers
+# this prefix + the raw catalog bytes, never the raw bytes alone, so a
+# catalog signature can't be replayed against some other byte-for-byte-
+# identical payload that means something else in a different context.
+_CATALOG_SIG_DOMAIN = b"bcc-catalog-v1|"
+
+# Top-level fields that must be https:// URLs when present.
+_CATALOG_URL_FIELDS = ("homepage", "docs_url", "source")
+
+# Inline secret-flag=value forms. Distinct from _TOKEN_PREFIXES below --
+# this catches "--api-key=<real value>" even when the value itself doesn't
+# match a well-known token prefix.
+_CATALOG_SECRET_ARG_RE = re.compile(r"(?i)--api[-_]?key=|--token=|--password=")
+
+# <PLACEHOLDER>-style tokens the GUI must have the user fill in before Save.
+_PLACEHOLDER_RE = re.compile(r"<[^<>\s]+>")
+
+# How many versions a single accepted catalog jump may leap in one go. Bounds
+# a "freeze" attack: a compromised/leaked signing key claiming an absurd
+# future version would otherwise permanently outrank every legitimate
+# catalog release from then on, since the resolver always prefers the
+# highest verified version.
+_CATALOG_MAX_VERSION_JUMP = 1000
+
+
+def load_catalog(raw: bytes | str) -> dict:
+    """
+    Parse catalog bytes/text into a dict using STRICT json.loads ONLY.
+
+    🔴 CRITICAL: the lenient JSON repair pipeline (repair_json_text,
+    parse_pasted_json / parse_pasted_json_verbose) must NEVER be wired in
+    here, or anywhere near catalog handling. That pipeline exists to be
+    forgiving of hand-pasted snippets from docs and blog posts — smart
+    quotes, trailing commas, unquoted keys, whatever a human fat-fingered.
+    Forgiveness is exactly the property a signed payload cannot have:
+    verify_catalog_signature() authenticates the exact bytes that were
+    signed. If what gets displayed/executed is a "repaired" reinterpretation
+    of those bytes rather than the bytes themselves, the signature check
+    still passes while guaranteeing nothing about what actually runs. Always
+    verify raw bytes, then load_catalog() those SAME raw bytes.
+    """
+    return json.loads(raw)
+
+
+def catalog_version(data: dict) -> int:
+    """Extract the integer version from a parsed catalog dict (0 if absent/bad)."""
+    version = data.get("version") if isinstance(data, dict) else None
+    return version if isinstance(version, int) and not isinstance(version, bool) else 0
+
+
+def _secret_looking_arg(a: str) -> bool:
+    """
+    True when a catalog arg string looks like it embeds a real secret. Reuses
+    the existing token-prefix detector (_is_secret_value / _TOKEN_PREFIXES)
+    rather than reimplementing it — one definition of "looks like a secret"
+    for the whole app.
+    """
+    if _CATALOG_SECRET_ARG_RE.search(a):
+        return True
+    value = a.split("=", 1)[1] if "=" in a else a
+    return _is_secret_value(value) or _is_secret_value(a)
+
+
+def _docker_arg_violations(tag: str, args: list[str]) -> list[str]:
+    """--privileged and volume mounts rooted at / or $HOME are refused."""
+    problems: list[str] = []
+    if "--privileged" in args:
+        problems.append(f"{tag}: config.args uses --privileged, which is not allowed.")
+
+    i = 0
+    while i < len(args):
+        a = args[i]
+        mount = None
+        if a in ("-v", "--volume") and i + 1 < len(args):
+            mount = args[i + 1]
+            i += 1
+        elif a.startswith("--volume="):
+            mount = a.split("=", 1)[1]
+        elif a.startswith("-v") and a != "-v":
+            mount = a[2:]
+        if mount:
+            source = mount.split(":", 1)[0]
+            if source in ("/", "$HOME") or source.startswith("$HOME"):
+                problems.append(
+                    f"{tag}: config.args mounts {source!r}, which is not allowed "
+                    "(volume mounts of / or $HOME are refused)."
+                )
+        i += 1
+    return problems
+
+
+def _validate_catalog_config(tag: str, config) -> list[str]:
+    """Validate the `config` block of a basic-tier catalog entry."""
+    if not isinstance(config, dict):
+        return [f"{tag}: basic entries require a 'config' object with command+args."]
+
+    problems: list[str] = []
+
+    command = config.get("command")
+    if not isinstance(command, str) or not command:
+        problems.append(f"{tag}: config.command must be a non-empty string.")
+        command = ""
+    elif not command.isascii():
+        problems.append(f"{tag}: config.command must be ASCII (non-ASCII code points rejected).")
+
+    if command and command not in CATALOG_ALLOWED_COMMANDS:
+        problems.append(
+            f"{tag}: config.command {command!r} is not on the catalog allowlist "
+            f"({', '.join(sorted(CATALOG_ALLOWED_COMMANDS))})."
+        )
+
+    args = config.get("args")
+    if not isinstance(args, list) or not all(isinstance(a, str) for a in args):
+        problems.append(f"{tag}: config.args must be a list of strings.")
+        args = []
+
+    for a in args:
+        if not a.isascii():
+            problems.append(f"{tag}: config.args contains a non-ASCII value ({a!r}).")
+        if _secret_looking_arg(a):
+            problems.append(
+                f"{tag}: config.args contains a secret-looking value ({a!r}); "
+                "secrets belong in env, never args."
+            )
+
+    if command in ("node", "python", "python3") and any(a in ("-e", "--eval", "-c") for a in args):
+        problems.append(
+            f"{tag}: config.args uses -e/--eval/-c with {command!r}, which is not allowed."
+        )
+
+    if command == "docker":
+        problems.extend(_docker_arg_violations(tag, args))
+
+    env = config.get("env")
+    if env is not None and (
+        not isinstance(env, dict) or any(not isinstance(v, str) for v in env.values())
+    ):
+        problems.append(f"{tag}: config.env must be an object of string values.")
+
+    return problems
+
+
+def _validate_catalog_entry(idx: int, entry, seen_ids: set[str]) -> list[str]:
+    """Validate a single `servers[idx]` catalog entry."""
+    tag = f"servers[{idx}]"
+    if not isinstance(entry, dict):
+        return [f"{tag}: must be an object."]
+
+    problems: list[str] = []
+
+    entry_id = entry.get("id")
+    if not isinstance(entry_id, str) or not entry_id.strip():
+        problems.append(f"{tag}: 'id' must be a non-empty string.")
+    else:
+        tag = f"servers[{idx}] ({entry_id!r})"
+        if not entry_id.isascii():
+            problems.append(f"{tag}: 'id' must be ASCII (non-ASCII code points rejected).")
+        if entry_id in seen_ids:
+            problems.append(f"{tag}: duplicate id.")
+        seen_ids.add(entry_id)
+
+    for field in ("display", "description", "category"):
+        if not isinstance(entry.get(field), str) or not entry[field].strip():
+            problems.append(f"{tag}: '{field}' must be a non-empty string.")
+
+    if not isinstance(entry.get("official"), bool):
+        problems.append(f"{tag}: 'official' must be a boolean.")
+
+    setup = entry.get("setup")
+    if setup not in ("basic", "link-only"):
+        problems.append(f"{tag}: 'setup' must be 'basic' or 'link-only'.")
+
+    env_required = entry.get("env_required")
+    if not isinstance(env_required, dict):
+        problems.append(f"{tag}: 'env_required' must be an object.")
+    else:
+        for k, v in env_required.items():
+            if not isinstance(k, str):
+                problems.append(f"{tag}: 'env_required' keys must be strings.")
+            if v != "":
+                problems.append(
+                    f"{tag}: env_required[{k!r}] must be an empty string — "
+                    "catalog entries never ship secret values, only the names "
+                    "of env vars the user must fill in."
+                )
+
+    for field in _CATALOG_URL_FIELDS:
+        if field in entry and entry[field] is not None:
+            url = entry[field]
+            if not isinstance(url, str) or not url.startswith("https://"):
+                problems.append(f"{tag}: '{field}' must be an https:// URL.")
+
+    if setup == "link-only":
+        if entry.get("config") is not None:
+            problems.append(f"{tag}: link-only entries must not have a 'config'.")
+        docs_url = entry.get("docs_url")
+        if not isinstance(docs_url, str) or not docs_url.startswith("https://"):
+            problems.append(f"{tag}: link-only entries require an https:// 'docs_url'.")
+    elif setup == "basic":
+        problems.extend(_validate_catalog_config(tag, entry.get("config")))
+
+    return problems
+
+
+def validate_catalog(data) -> list[str]:
+    """
+    Validate a parsed catalog dict. Returns a list of human-readable
+    problems; an EMPTY list means the catalog is valid.
+
+    A non-empty list means REJECT THE WHOLE FILE, not just the offending
+    entry. There is no per-entry salvage here: a catalog that is invalid in
+    one place is untrusted everywhere, because a caller that tried to keep
+    "the other 19 entries that looked fine" would need its own judgment call
+    about which parts of a failed-validation file to trust — exactly the
+    judgment call this function exists to make once, centrally.
+    """
+    if not isinstance(data, dict):
+        return ["Catalog root must be a JSON object."]
+
+    problems: list[str] = []
+
+    schema = data.get("schema")
+    if not isinstance(schema, int) or isinstance(schema, bool) or schema < 1:
+        problems.append("'schema' must be a positive integer.")
+
+    version = data.get("version")
+    if not isinstance(version, int) or isinstance(version, bool) or version < 1:
+        problems.append("'version' must be a positive integer.")
+
+    servers = data.get("servers")
+    if not isinstance(servers, list):
+        problems.append("'servers' must be a list.")
+        return problems  # nothing else to check without a server list
+
+    seen_ids: set[str] = set()
+    for idx, entry in enumerate(servers):
+        problems.extend(_validate_catalog_entry(idx, entry, seen_ids))
+
+    return problems
+
+
+def verify_catalog_signature(raw: bytes, sig: bytes, pubkeys: list[bytes]) -> bool:
+    """
+    Verify an Ed25519 signature over `raw` catalog bytes.
+
+    The signed message is domain-separated: b"bcc-catalog-v1|" + raw, not
+    raw alone (see _CATALOG_SIG_DOMAIN).
+
+    Returns True if ANY key in `pubkeys` verifies — this is what lets keys
+    rotate without bricking installs still trusting an older key.
+
+    Never raises. An invalid signature, a garbage/wrong-length key, a
+    non-bytes argument, an empty signature — all of it just returns False.
+    Signature verification is exactly the wrong place for an exception to
+    accidentally propagate into a code path that fails open.
+    """
+    if not isinstance(raw, bytes) or not isinstance(sig, (bytes, bytearray)):
+        return False
+    if not sig:
+        return False
+    message = _CATALOG_SIG_DOMAIN + raw
+    for pk in pubkeys or []:
+        try:
+            Ed25519PublicKey.from_public_bytes(bytes(pk)).verify(bytes(sig), message)
+            return True
+        except (InvalidSignature, ValueError, TypeError):
+            continue
+    return False
+
+
+def resolve_catalog(
+    bundled: tuple[bytes, bytes] | None,
+    cached: tuple[bytes, bytes] | None,
+    remote: tuple[bytes, bytes] | None,
+) -> dict:
+    """
+    Pick the highest-version catalog among bundled/cached/remote. Each
+    argument is either None (unavailable) or an (raw_bytes, signature_bytes)
+    pair.
+
+    🔴 SECURITY: every candidate — including `bundled`, the copy frozen into
+    this binary — is verified against CATALOG_PUBKEYS and re-validated from
+    scratch right here. The bundled catalog gets NO implicit trust. This was
+    a hole in the original design: bundling data/catalog.json as a plain
+    asset would let an unsigned/malformed payload that somehow merged to
+    main ship inside the next release and win the version comparison simply
+    by virtue of being local. Signing (and checking the signature at
+    runtime, every time) closes that.
+
+    Anti-rollback: a candidate's version is never accepted if it's lower
+    than the best verified candidate already found in this same resolution
+    pass — an attacker replaying an old, since-superseded signed catalog
+    can't downgrade you.
+
+    Anti-freeze: a candidate whose version leaps more than
+    _CATALOG_MAX_VERSION_JUMP past the current best is also rejected. A
+    compromised/leaked signing key claiming an absurd future version would
+    otherwise permanently outrank every legitimate release from then on,
+    since the resolver always prefers the highest verified version — this
+    caps how far a single accepted jump can go.
+
+    Returns the winning catalog dict, or {} if nothing verified and
+    validated.
+    """
+    best: dict = {}
+    best_version = -1
+
+    for candidate in (bundled, cached, remote):
+        if not candidate:
+            continue
+        raw, sig = candidate
+        if not verify_catalog_signature(raw, sig, CATALOG_PUBKEYS):
+            continue
+        try:
+            data = load_catalog(raw)
+        except (ValueError, TypeError):
+            continue
+        if validate_catalog(data):
+            continue
+
+        version = catalog_version(data)
+        if best_version >= 0:
+            if version < best_version:
+                continue  # anti-rollback
+            if version > best_version + _CATALOG_MAX_VERSION_JUMP:
+                continue  # anti-freeze
+
+        best = data
+        best_version = version
+
+    return best
+
+
+def catalog_entry_to_paste_json(entry: dict) -> dict:
+    """
+    Convert a basic-tier catalog entry into the {name: {command, args, env}}
+    shape parse_pasted_json()/_import_server() already understand, so the
+    (future) catalog picker dialog can feed a selection straight into the
+    existing paste-import path instead of growing a parallel one.
+    """
+    config = entry.get("config") or {}
+    name = entry.get("id") or entry.get("display") or "server"
+    data: dict = {
+        "command": config.get("command", ""),
+        "args": list(config.get("args") or []),
+    }
+    env = config.get("env")
+    if env:
+        data["env"] = dict(env)
+    return {str(name): data}
+
+
+def config_has_unfilled_placeholders(cfg: dict) -> bool:
+    """
+    True if any <PLACEHOLDER>-style token remains anywhere in a server
+    config's command/args/env (the shape produced by
+    catalog_entry_to_paste_json). The GUI uses this to refuse Save until
+    every <ALLOWED_DIR>-style token has been filled in with a real value.
+    """
+    values: list[str] = []
+    cmd = cfg.get("command")
+    if isinstance(cmd, str):
+        values.append(cmd)
+    values.extend(a for a in (cfg.get("args") or []) if isinstance(a, str))
+    env = cfg.get("env") or {}
+    if isinstance(env, dict):
+        values.extend(v for v in env.values() if isinstance(v, str))
+    return any(_PLACEHOLDER_RE.search(v) for v in values)