""" mcp_core.py -- Pure logic for the Claude MCP config manager. No GUI imports live in here on purpose: every function below is unit-testable and could just as easily back a CLI. The GUI (mcp_manager.py) is a thin shell over these functions. The cardinal rule of this module: when writing a config back to disk we ONLY ever touch the `mcpServers` block (and our own `_disabledMcpServers` parking key). Every other top-level key in the user's config is preserved verbatim and in its original position. """ from __future__ import annotations import base64 import contextlib import difflib import functools import glob import json import os import re import shutil import signal as _signal import subprocess import sys import tempfile import threading import time from dataclasses import dataclass 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 # reads `mcpServers`, so anything here is ignored by the app but kept on disk so # we can toggle it back on without losing the definition. DISABLED_KEY = "_disabledMcpServers" # Named server sets: {set_name: [enabled server names]}. Same pattern as # DISABLED_KEY — a bcc-owned key Claude ignores, stored in the config file so # sets travel with it. Applying a set enables exactly the listed servers and # parks the rest under DISABLED_KEY. SETS_KEY = "_bccServerSets" BACKUP_DIRNAME = ".bcc_backups" MAX_BACKUPS = 15 # Fields the editor knows how to render. Anything else on a server object is # considered "extra" and is preserved untouched on save. KNOWN_FIELDS = {"command", "args", "env", "url", "type", "headers"} # --------------------------------------------------------------------------- # # Version / update checking # # __version__ is the single source of truth for the app version (must match # pyproject.toml's [project] version). The About dialog and the update # checker both read this constant instead of hard-coding a version string. # # The update checker is notify-only: it reads release metadata from the # repo's Gitea releases API and NEVER downloads or replaces the running # binary. All network I/O here is fail-quiet (returns None on any problem) # so it's safe to run unattended, off the UI thread, at startup. # --------------------------------------------------------------------------- # __version__ = "1.3.0" REPO_URL = "https://git.avezzano.io/the_og/better-claude-config" ISSUES_URL = f"{REPO_URL}/issues" RELEASES_URL = f"{REPO_URL}/releases" LICENSE_URL = f"{REPO_URL}/raw/branch/main/LICENSE" # Public repo -> anonymously reachable, no auth/token needed or embedded. _RELEASES_API_URL = ( "https://git.avezzano.io/api/v1/repos/the_og/better-claude-config/releases/latest" ) def parse_version(v: str) -> tuple[int, ...]: """ Parse a version string into a tuple of ints for numeric comparison. Strips a leading 'v' ("v1.2.3" -> "1.2.3") and any pre-release/build metadata after a '-' or '+' ("1.2.3-beta.1" -> "1.2.3"). Stops at the first non-numeric dotted component. Empty or entirely non-numeric input returns an empty tuple rather than raising, so a malformed tag from a flaky API response degrades gracefully instead of crashing the caller. """ s = (v or "").strip() if s[:1].lower() == "v": s = s[1:] s = re.split(r"[-+]", s, maxsplit=1)[0] parts: list[int] = [] for chunk in s.split("."): m = re.match(r"\d+", chunk) if not m: break parts.append(int(m.group())) return tuple(parts) def is_newer_version(current: str, candidate: str) -> bool: """ True if `candidate` is a strictly newer version than `current`. Comparison is purely numeric (major.minor.patch, ...) — NEVER a lexical string compare, so "v2.0.0" vs "v10.0.0" resolves correctly instead of sorting "2" after "10". Tuples of differing length are zero-padded before comparing, so "1.2" and "1.2.0" are correctly treated as equal. An unparseable `candidate` always yields False (nothing to report). An unparseable `current` is treated as "0" for comparison purposes — a malformed local version shouldn't silently suppress a real update. """ cur = parse_version(current) new = parse_version(candidate) if not new: return False width = max(len(cur), len(new), 1) cur = cur + (0,) * (width - len(cur)) new = new + (0,) * (width - len(new)) return new > cur def fetch_latest_release(timeout: float = 4.0) -> dict | None: """ Query the repo's (public, anonymous) Gitea releases API for the latest release. Returns {"version": "", "url": ""} on success, or None on ANY failure: network error, timeout, bad status, malformed JSON, or a response missing tag_name. Fail-quiet by design — this is meant to be called off the UI thread (see UpdateCheckWorker in bcc.py) for both the About dialog's "Check for updates" button and an optional silent startup check. Never downloads or touches any binary; this only ever reads release metadata. """ import urllib.error import urllib.request req = urllib.request.Request( _RELEASES_API_URL, headers={ "Accept": "application/json", "User-Agent": f"BetterClaudeConfig/{__version__}", }, ) try: with urllib.request.urlopen(req, timeout=timeout) as r: payload = json.loads(r.read().decode("utf-8")) except (urllib.error.URLError, TimeoutError, ValueError, OSError): return None if not isinstance(payload, dict): return None tag = payload.get("tag_name") if not tag or not isinstance(tag, str): return None return {"version": tag, "url": payload.get("html_url") or RELEASES_URL} def update_notice( current: str, release: dict | None, url_fallback: str = RELEASES_URL ) -> dict | None: """Decide whether to tell the user about a release, and what to say. Returns {"version", "text", "url"} when `release` is newer than `current`, or None when it isn't, when the check failed, or when the payload is malformed. Kept here rather than in the GUI so the wording and the should-we-notify decision are testable -- bcc.py can't be imported by the test suite, which has no PySide6. The text deliberately names no menu path. The old status-line notice read "Help > About to view it", which is wrong on macOS: Qt relocates the About action into the application menu (#79). A notice that carries its own action can't drift out of sync with the platform. """ if not isinstance(release, dict): return None version = release.get("version") if not version or not isinstance(version, str): return None if not is_newer_version(current, version): return None # Tags carry a "v" prefix and __version__ doesn't; render both the same way # so the notice doesn't read "Version v1.3.0 ... you're running 1.2.0". shown = version.lstrip("vV") return { "version": version, "text": f"Version {shown} is available. You're running {current.lstrip('vV')}.", "url": release.get("url") or url_fallback, } # --------------------------------------------------------------------------- # # Data model # --------------------------------------------------------------------------- # @dataclass class Profile: """A discovered (or manually added) Claude install location.""" label: str path: Path config_exists: bool def __post_init__(self): self.path = Path(self.path) @dataclass class ServerEntry: name: str data: dict enabled: bool = True @property def kind(self) -> str: return "remote" if "url" in self.data and "command" not in self.data else "stdio" # --------------------------------------------------------------------------- # # Discovery # --------------------------------------------------------------------------- # def app_support_base() -> Path: """The per-platform directory that holds the `Claude*` data folders.""" if sys.platform == "darwin": return Path.home() / "Library" / "Application Support" if os.name == "nt": return Path(os.environ.get("APPDATA", Path.home() / "AppData" / "Roaming")) return Path(os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config")) def msix_config_paths(localappdata: str | os.PathLike | None = None) -> list[Path]: """ Find MSIX/Store-packaged Claude Desktop configs. When Claude Desktop is installed from the Microsoft Store (MSIX), Windows virtualizes its filesystem writes to a per-package folder under `%LOCALAPPDATA%\\Packages\\\\LocalCache\\Roaming\\Claude\\` instead of the normal `%APPDATA%\\Claude\\`. A user (or BCC) editing the plain %APPDATA% path can end up changing a file the running app never reads -- see anthropics/claude-code issues #26073, #29100, #38830. Globs `/Packages/*Claude*/LocalCache/Roaming/Claude/ claude_desktop_config.json` and returns every match that actually exists, sorted for determinism. `localappdata` defaults to the %LOCALAPPDATA% env var (falling back to the usual Windows path) but is accepted as a parameter so this is unit-testable with tmp_path on any platform. This function itself is platform-independent (it just globs whatever directory it's given); callers that care about the *current* machine should gate on sys.platform -- see `detect_msix_claude`. """ base = ( Path(localappdata) if localappdata is not None else Path(os.environ.get("LOCALAPPDATA", str(Path.home() / "AppData" / "Local"))) ) packages = base / "Packages" if not packages.is_dir(): return [] out: list[Path] = [] for pkg_dir in sorted(packages.glob("*Claude*")): cfg = pkg_dir / "LocalCache" / "Roaming" / "Claude" / CONFIG_FILENAME if cfg.is_file(): out.append(cfg) return out def detect_msix_claude( appdata: str | os.PathLike | None = None, localappdata: str | os.PathLike | None = None, ) -> Path | None: """ Best-effort detection of an MSIX-virtualized Claude Desktop install. Returns the first virtualized `claude_desktop_config.json` found (see `msix_config_paths`), or None when not running on Windows, no matching package folder exists, or a package folder exists but has no config file written yet. The sys.platform gate makes this a safe no-op to call unconditionally from discovery/diagnostics code on macOS/Linux. `appdata`/`localappdata` are threaded through (rather than read straight from os.environ) purely so the whole detection path is unit-testable via tmp_path + monkeypatch without mutating real env vars. """ if not sys.platform.startswith("win"): return None hits = msix_config_paths(localappdata) return hits[0] if hits else None def msix_warning_text( appdata: str | os.PathLike | None = None, localappdata: str | os.PathLike | None = None, ) -> str | None: """ A one-line, paste-safe warning for the diagnostics/status surface when Claude Desktop looks like an MSIX/Store install whose real config lives somewhere other than the plain %APPDATA%\\Claude\\ path. Returns None when nothing was detected (including on non-Windows platforms) or when the virtualized path and the plain path happen to coincide -- i.e. there is nothing surprising to warn about. Contains only filesystem paths, no env values or secrets. """ real = detect_msix_claude(appdata, localappdata) if real is None: return None plain_base = ( Path(appdata) if appdata is not None else Path(os.environ.get("APPDATA", str(Path.home() / "AppData" / "Roaming"))) ) plain_cfg = plain_base / "Claude" / CONFIG_FILENAME if plain_cfg == real: return None return ( "Claude Desktop looks like it's installed from the Microsoft Store (MSIX). " f"Windows virtualizes its config, so edits to {plain_cfg} may be silently " f"ignored by the running app. The real config is at: {real}" ) def discover_project_configs(claude_json_path: str | os.PathLike) -> list[Profile]: """ Find project-scope `.mcp.json` configs known to Claude Code. `~/.claude.json` keeps a `projects` map keyed by absolute project directory path (that's what the CLI writes as it's used in each repo). Any project whose directory has a `.mcp.json` file next to it -- a standalone file with a top-level `mcpServers` object, same shape BCC already edits -- is surfaced here as its own profile so it can be opened via 'Add config...' without hunting for the path by hand. Fails quiet: a missing/unreadable/malformed `claude_json_path`, or a `projects` value that isn't a dict, just yields an empty list rather than raising -- this is best-effort discovery, not a required config load. """ try: cfg = load_config(claude_json_path) except Exception: return [] projects = cfg.get("projects") if not isinstance(projects, dict): return [] out: list[Profile] = [] for key in sorted(k for k in projects if isinstance(k, str)): mcp_path = Path(key) / ".mcp.json" if mcp_path.is_file(): out.append( Profile(label=f"Project: {Path(key).name}", path=mcp_path, config_exists=True) ) return out def discover_profiles() -> list[Profile]: """ Find every `Claude*` data directory in the platform's app-support base (Claude Desktop installs), then also check for a Claude Code global config and any project-scope `.mcp.json` configs it knows about. Claude Desktop: scans the platform app-support folder for any `Claude*` directory (catches `Claude`, `Claude-Work`, etc.). Windows/MSIX: if Claude Desktop was installed from the Microsoft Store, its real config lives in a virtualized per-package folder rather than the plain %APPDATA%\\Claude\\ path above (see `detect_msix_claude`); when that's detected, it's surfaced here as its own profile so the user can edit the file the app actually reads. Claude Code: user-scope MCP servers live in ~/.claude.json (that's what `claude mcp add` writes; project scope is a per-repo .mcp.json, which can be opened via 'Add config…'). NOT ~/.claude/settings.json — that file is for permissions/hooks and rejects an mcpServers key with a schema error. Project scope: ~/.claude.json also tracks a `projects` map, one entry per directory Claude Code has been run in; any of those with a `.mcp.json` file are surfaced as their own profiles too (see `discover_project_configs`). """ base = app_support_base() out: list[Profile] = [] if base.is_dir(): seen = set() for d in sorted(base.glob("Claude*")): if d.is_dir() and d.name not in seen: seen.add(d.name) cfg = d / CONFIG_FILENAME out.append(Profile(label=d.name, path=cfg, config_exists=cfg.is_file())) msix_cfg = detect_msix_claude() if msix_cfg is not None and str(msix_cfg) not in {str(p.path) for p in out}: out.append( Profile(label="Claude (Microsoft Store / MSIX)", path=msix_cfg, config_exists=True) ) home = Path.home() cc_cfg = home / ".claude.json" out.append(Profile(label="Claude Code", path=cc_cfg, config_exists=cc_cfg.is_file())) existing_paths = {str(p.path) for p in out} for proj in discover_project_configs(cc_cfg): if str(proj.path) not in existing_paths: existing_paths.add(str(proj.path)) out.append(proj) # Legacy: earlier BCC versions (and hand-edits) may have parked servers in # ~/.claude/settings.json, where Claude Code ignores them. Surface that # file only when it actually contains an mcpServers block, so the user can # Copy to ▸ the entries into the real config. legacy = home / ".claude" / "settings.json" if legacy.is_file(): with contextlib.suppress(Exception): if "mcpServers" in load_config(legacy): out.append( Profile( label="Claude Code (legacy settings.json)", path=legacy, config_exists=True ) ) return out def profile_from_path(path: str | os.PathLike) -> Profile: """Build a Profile from a user-supplied config path (the 'override' case).""" p = Path(path) # Prefer the parent folder name as the label (e.g. .../Claude-Work/...json -> Claude-Work) label = p.parent.name or p.name return Profile(label=label, path=p, config_exists=p.is_file()) def profile_targets_claude_desktop(profile: Profile) -> bool: """ True when `profile` points at a Claude Desktop config (claude_desktop_config.json), as opposed to Claude Code (~/.claude.json or the legacy ~/.claude/settings.json). Used to gate Desktop-only actions like "Restart Claude Desktop" so they never show up for a Claude Code profile -- restarting the CLI makes no sense. """ return Path(profile.path).name == CONFIG_FILENAME # --------------------------------------------------------------------------- # # Load / extract / apply # --------------------------------------------------------------------------- # def load_config(path: str | os.PathLike) -> dict: """Read the full config dict. Missing/empty file -> {}. Bad JSON -> raises.""" p = Path(path) if not p.is_file(): return {} text = p.read_text(encoding="utf-8") if not text.strip(): return {} obj = json.loads(text) # JSONDecodeError bubbles up for the GUI to display if not isinstance(obj, dict): raise ValueError("Top-level JSON in the config file is not an object.") return obj def repair_config_file(path: str | os.PathLike) -> tuple[dict, list[str], str]: """ Attempt to repair a config file that failed strict parsing, WITHOUT writing anything to disk. The caller decides what to do with the result (BCC shows the user the issues and asks before saving). Returns (cfg, notes, repaired_text): cfg - the parsed dict from the repaired JSON notes - human-readable list of fixes that were applied repaired_text - pretty-printed JSON the config would become Raises ValueError if the file can't be salvaged automatically. """ text = Path(path).read_text(encoding="utf-8") candidate, notes = repair_json_text(text) try: obj = json.loads(candidate) except json.JSONDecodeError as e: raise ValueError( f"Couldn't repair the file automatically (line {e.lineno}, column {e.colno}: {e.msg})." ) from e if not isinstance(obj, dict): raise ValueError("Even after repair, the top level isn't a JSON object.") pretty = json.dumps(obj, indent=2, ensure_ascii=False) + "\n" return obj, notes, pretty def extract_servers(cfg: dict) -> list[ServerEntry]: """Pull enabled (`mcpServers`) and disabled (`_disabledMcpServers`) servers.""" out: list[ServerEntry] = [] for name, data in (cfg.get("mcpServers") or {}).items(): out.append(ServerEntry(name=name, data=dict(data), enabled=True)) for name, data in (cfg.get(DISABLED_KEY) or {}).items(): out.append(ServerEntry(name=name, data=dict(data), enabled=False)) return out # --------------------------------------------------------------------------- # # Named server sets (issue #52) # --------------------------------------------------------------------------- # def list_server_sets(cfg: dict) -> dict[str, list[str]]: """ Return {set_name: [enabled server names]} from cfg's SETS_KEY. Fail-soft: entries whose value isn't a list of strings (hand-edited or corrupted) are skipped rather than raising, so one bad set never hides the rest. """ raw = cfg.get(SETS_KEY) if not isinstance(raw, dict): return {} out: dict[str, list[str]] = {} for name, members in raw.items(): if isinstance(members, list) and all(isinstance(m, str) for m in members): out[str(name)] = list(members) return out def save_server_set(cfg: dict, name: str, servers: list[ServerEntry]) -> list[str]: """ Snapshot the current enabled-server names into cfg under SETS_KEY as `name` (overwriting an existing set of that name). Returns the saved member list. The caller decides when cfg reaches disk (normal Save flow). """ members = [s.name for s in servers if s.enabled] sets = cfg.get(SETS_KEY) if not isinstance(sets, dict): sets = {} cfg[SETS_KEY] = sets sets[name] = members return members def delete_server_set(cfg: dict, name: str) -> bool: """Remove set `name` from cfg. Drops SETS_KEY entirely when the last set goes, so untouched configs don't grow an empty bcc key. Returns True if something was deleted.""" sets = cfg.get(SETS_KEY) if not isinstance(sets, dict) or name not in sets: return False del sets[name] if not sets: cfg.pop(SETS_KEY, None) return True def apply_server_set(servers: list[ServerEntry], enabled_names: list[str]) -> list[str]: """ Enable exactly the servers named in `enabled_names`; disable every other entry (in place). Returns the set members that no longer exist in `servers` — the caller surfaces those as a warning, and the rest of the set still applies. """ wanted = set(enabled_names) present: set[str] = set() for s in servers: s.enabled = s.name in wanted present.add(s.name) return sorted(wanted - present) def resolve_name_collision(name: str, existing: set[str]) -> str: """ Return a name guaranteed not to collide with `existing`. If `name` isn't already taken it's returned unchanged. Otherwise a `-2`, `-3`, ... suffix is appended until the result is unique — this is the "keep both (renamed)" branch used by paste/import when the user doesn't want to overwrite an existing server of the same name. """ if name not in existing: return name n = 2 candidate = f"{name}-{n}" while candidate in existing: n += 1 candidate = f"{name}-{n}" return candidate def apply_servers(cfg: dict, servers: list[ServerEntry]) -> dict: """ Write the server list back into `cfg` in place, preserving every other key and the position of `mcpServers`. Returns the same dict for convenience. """ enabled = {s.name: s.data for s in servers if s.enabled} disabled = {s.name: s.data for s in servers if not s.enabled} cfg["mcpServers"] = enabled # replaces value if key existed; appends otherwise if disabled: cfg[DISABLED_KEY] = disabled else: cfg.pop(DISABLED_KEY, None) return cfg # --------------------------------------------------------------------------- # # Write (atomic, with rotating backups) # --------------------------------------------------------------------------- # def _make_backup(path: Path) -> Path: bdir = path.parent / BACKUP_DIRNAME bdir.mkdir(exist_ok=True) stamp = time.strftime("%Y%m%d-%H%M%S") dest = bdir / f"{path.stem}.{stamp}.json" # Avoid clobbering a same-second backup n = 1 while dest.exists(): dest = bdir / f"{path.stem}.{stamp}.{n}.json" n += 1 shutil.copy2(path, dest) # Prune oldest beyond MAX_BACKUPS backups = sorted(bdir.glob(f"{path.stem}.*.json")) for old in backups[:-MAX_BACKUPS]: with contextlib.suppress(OSError): old.unlink() return dest def write_config(path: str | os.PathLike, cfg: dict) -> Path | None: """ Atomically write `cfg` to `path` (2-space pretty JSON). Backs up any existing file first. Returns the backup path (or None if there was nothing to back up). """ p = Path(path) p.parent.mkdir(parents=True, exist_ok=True) backup = _make_backup(p) if p.is_file() else None payload = json.dumps(cfg, indent=2, ensure_ascii=False) + "\n" fd, tmp = tempfile.mkstemp(dir=str(p.parent), prefix=".tmp_mcp_", suffix=".json") try: with os.fdopen(fd, "w", encoding="utf-8") as f: f.write(payload) os.replace(tmp, p) # atomic on the same filesystem finally: if os.path.exists(tmp): os.remove(tmp) return backup # --------------------------------------------------------------------------- # # Backup utility functions # --------------------------------------------------------------------------- # _BACKUP_TS_RE = re.compile(r"(\d{8}-\d{6})") def list_backups(config_path: Path | str) -> list[Path]: """Return backup files for config_path, newest-first. Returns [] if none exist.""" p = Path(config_path) bdir = p.parent / BACKUP_DIRNAME if not bdir.is_dir(): return [] return sorted(bdir.glob(f"{p.stem}.*.json"), reverse=True) def backup_label(backup_path: Path | str) -> str: """Human-readable label derived from the backup filename timestamp.""" m = _BACKUP_TS_RE.search(Path(backup_path).name) if not m: return Path(backup_path).name ts = m.group(1) # e.g. "20260702-004124" return f"{ts[:4]}-{ts[4:6]}-{ts[6:8]} {ts[9:11]}:{ts[11:13]}:{ts[13:]}" def _redact_server_data(data: dict) -> dict: """Return a copy of a server definition with secrets masked for display.""" out = dict(data) if "args" in out: out["args"] = redact_args(list(out["args"] or [])) if "env" in out: out["env"] = {k: (MASK if is_secret_key(k) else v) for k, v in (out["env"] or {}).items()} return out def _redact_servers_block(block: dict | None) -> dict: """Return a sanitized copy of a mcpServers / _disabledMcpServers block.""" if not block: return {} return {name: _redact_server_data(data) for name, data in block.items()} def _server_sections(cfg: dict) -> dict: """Return the masked server sections of a config dict, safe for diff display.""" out: dict = {"mcpServers": _redact_servers_block(cfg.get("mcpServers"))} if DISABLED_KEY in cfg: out[DISABLED_KEY] = _redact_servers_block(cfg.get(DISABLED_KEY)) return out def backup_diff( config_path: Path | str, backup_path: Path | str, current_cfg: dict | None = None, ) -> str: """ Unified diff showing what the server sections would look like after restoring the backup. Only mcpServers and _disabledMcpServers are compared; non-server keys are excluded entirely (they are preserved by restore, not changed). Secret values in args and env are masked so the diff is safe to share. fromfile = current state, tofile = what will be written after restore. Pass current_cfg to diff against an already-loaded in-memory config instead of re-reading from disk. """ p = Path(config_path) bp = Path(backup_path) if current_cfg is None: try: current_cfg = load_config(p) except Exception: current_cfg = {} backup_servers = extract_servers(load_config(bp)) after_cfg = dict(current_cfg) apply_servers(after_cfg, backup_servers) before_lines = ( json.dumps(_server_sections(current_cfg), indent=2, ensure_ascii=False) + "\n" ).splitlines(keepends=True) after_lines = ( json.dumps(_server_sections(after_cfg), indent=2, ensure_ascii=False) + "\n" ).splitlines(keepends=True) result = "".join( difflib.unified_diff( before_lines, after_lines, fromfile=f"current: {p.name}", tofile=f"restore: {bp.name}", ) ) return result or "(no differences — backup matches current servers)" def restore_backup( config_path: Path | str, backup_path: Path | str, current_cfg: dict | None = None, ) -> Path | None: """ Restore server entries from backup_path into config_path. Only mcpServers and _disabledMcpServers are replaced; all other keys in the current config (conversation history, project state, etc.) are preserved verbatim and in their original order. Goes through write_config() so a pre-restore backup of the current file is created automatically. Returns that backup path (or None if no prior file). """ p = Path(config_path) if current_cfg is None: try: current_cfg = load_config(p) except Exception: current_cfg = {} backup_servers = extract_servers(load_config(Path(backup_path))) cfg = dict(current_cfg) apply_servers(cfg, backup_servers) return write_config(p, cfg) def config_mtime(path: Path | str) -> float | None: """Return the file's mtime, or None if the file does not exist.""" try: return Path(path).stat().st_mtime except OSError: return None class ConfigStat(NamedTuple): """A snapshot of a config file's mtime + size. Pairing size with mtime hardens stale-file detection beyond bare mtime equality: a concurrent external write can land within the filesystem's mtime resolution (e.g. same-second writes on ext4/HFS+) or have its mtime restored by the writing process, in which case mtime alone would miss the change. Comparing both fields catches those cases without the cost of a full content hash. """ mtime: float size: int def config_fingerprint(path: Path | str) -> ConfigStat | None: """Return the file's (mtime, size) snapshot, or None if it does not exist.""" try: st = Path(path).stat() except OSError: return None return ConfigStat(st.st_mtime, st.st_size) def external_change_summary(original_cfg: dict, path: Path | str) -> tuple[list[str], str]: """ Compare original_cfg (what BCC loaded) with the current on-disk state. Returns: changed_keys — sorted list of top-level keys whose values differ. server_diff — masked unified diff of server sections (empty if unchanged or the file cannot be read). """ try: disk_cfg = load_config(Path(path)) except Exception: return [], "" all_keys = set(original_cfg) | set(disk_cfg) changed_keys = sorted(k for k in all_keys if original_cfg.get(k) != disk_cfg.get(k)) server_diff = "" if any(k in {"mcpServers", DISABLED_KEY} for k in changed_keys): before_lines = ( json.dumps(_server_sections(original_cfg), indent=2, ensure_ascii=False) + "\n" ).splitlines(keepends=True) after_lines = ( json.dumps(_server_sections(disk_cfg), indent=2, ensure_ascii=False) + "\n" ).splitlines(keepends=True) server_diff = "".join( difflib.unified_diff(before_lines, after_lines, fromfile="loaded", tofile="on disk now") ) return changed_keys, server_diff # --------------------------------------------------------------------------- # # Paste / import parsing # --------------------------------------------------------------------------- # def looks_like_server(data) -> bool: return isinstance(data, dict) and ("command" in data or "url" in data) def suggest_name(data: dict) -> str: """Derive a friendly default name from a server definition.""" if "url" in data and "command" not in data: host = urlparse(data["url"]).hostname or "remote" return host.split(".")[0] or "remote" for a in data.get("args", []) or []: if isinstance(a, str) and ("/" in a or a.startswith("@")): base = a.rstrip("/").split("/")[-1] base = base.replace("server-", "").replace("mcp-server-", "").replace("mcp-", "") if base: return base cmd = data.get("command", "server") return Path(str(cmd)).stem or "server" # --------------------------------------------------------------------------- # # Lenient JSON repair # # Snippets pasted from MCP docs, blog posts, and chat windows are frequently # not valid JSON: markdown fences, surrounding prose, // comments, trailing # commas, smart quotes, single quotes, unquoted keys, missing braces. The # whole point of BCC is that nobody should have to hand-fix JSON, so the # paste pipeline repairs what it can and reports what it changed. # --------------------------------------------------------------------------- # # Word-processor / web artifacts mapped back to ASCII. _QUOTE_MAP = { "“": '"', "”": '"', "„": '"', "«": '"', "»": '"', "‘": "'", "’": "'", "‚": "'", } _JUNK_CHARS = "​‌‍" # zero-width chars / BOM def _strip_fences(text: str, notes: list[str]) -> str: """Pull the contents out of a ```json ... ``` fence (or drop stray fence lines).""" m = re.search(r"```(?:json[c5]?|javascript|js)?\s*\n(.*?)```", text, re.DOTALL | re.IGNORECASE) if m: notes.append("extracted code from markdown fence") return m.group(1) if "```" in text: notes.append("removed markdown fence markers") return text.replace("```json", "").replace("```", "") return text def _normalize_unicode(text: str, notes: list[str]) -> str: out = text for junk in _JUNK_CHARS: out = out.replace(junk, "") out = out.replace(chr(0xA0), " ") # non-breaking space (defensive: avoid a literal char here) for smart, ascii_q in _QUOTE_MAP.items(): out = out.replace(smart, ascii_q) if out != text: notes.append("normalized smart quotes / invisible characters") return out def _tokenize(text: str, notes: list[str]) -> list[tuple[bool, str]]: """ One careful pass over the text producing (is_string, segment) pairs. While walking it also: converts single-quoted strings to double-quoted, and drops //, /* */ and # comments (never inside strings). """ segs: list[tuple[bool, str]] = [] buf: list[str] = [] i, n = 0, len(text) def flush(): if buf: segs.append((False, "".join(buf))) buf.clear() while i < n: ch = text[i] if ch == '"': # proper double-quoted string: copy verbatim flush() j = i + 1 s = ['"'] while j < n: c = text[j] s.append(c) if c == "\\" and j + 1 < n: s.append(text[j + 1]) j += 2 continue if c == '"': j += 1 break j += 1 segs.append((True, "".join(s))) i = j elif ch == "'": # single-quoted string: convert to double-quoted flush() j = i + 1 inner: list[str] = [] while j < n: c = text[j] if c == "\\" and j + 1 < n: nxt = text[j + 1] if nxt == "'": inner.append("'") # \' has no meaning in JSON else: inner.append(c) inner.append(nxt) j += 2 continue if c == "'": j += 1 break if c == '"': inner.append('\\"') j += 1 continue inner.append(c) j += 1 segs.append((True, '"' + "".join(inner) + '"')) notes.append("converted single-quoted strings") i = j elif ch == "/" and i + 1 < n and text[i + 1] == "/": notes.append("removed // comments") while i < n and text[i] != "\n": i += 1 elif ch == "/" and i + 1 < n and text[i + 1] == "*": notes.append("removed /* */ comments") i += 2 while i + 1 < n and not (text[i] == "*" and text[i + 1] == "/"): i += 1 i = min(i + 2, n) elif ch == "#": # Only treat as a comment at line start (after whitespace) — never # mid-value, where # could be part of an unquoted token. line_start = text.rfind("\n", 0, i) + 1 if text[line_start:i].strip() == "": notes.append("removed # comments") while i < n and text[i] != "\n": i += 1 else: buf.append(ch) i += 1 else: buf.append(ch) i += 1 flush() return segs def _repair_segments(segs: list[tuple[bool, str]], notes: list[str]) -> str: """Structural fixes that only apply outside strings.""" fixed: list[str] = [] for idx, (is_str, seg) in enumerate(segs): if is_str: fixed.append(seg) continue s = seg # Python / JS literals -> JSON s2 = re.sub(r"\bTrue\b", "true", s) s2 = re.sub(r"\bFalse\b", "false", s2) s2 = re.sub(r"\bNone\b|\bundefined\b", "null", s2) if s2 != s: notes.append("converted Python/JS literals (True/False/None)") s = s2 # Unquoted keys: { key: or , key: -> "key": s2 = re.sub(r"([{,]\s*)([A-Za-z_$][\w$.-]*)(\s*:)", r'\1"\2"\3', s) if s2 != s: notes.append("quoted bare object keys") s = s2 # Missing comma before the next string on a new line, e.g. # "args": ["x"]\n"env": {...} or {...}\n"name2": {...} # A string directly after a value-ish ending across a newline can never # be valid JSON without a comma, so inserting one is always safe. if idx + 1 < len(segs) and segs[idx + 1][0]: body = s.rstrip() tail_ws = s[len(body) :] prev = body[-1:] if body else (fixed[-1].rstrip()[-1:] if fixed else "") value_ending = prev in ('"', "}", "]") or prev.isdigit() if "\n" in tail_ws and value_ending: notes.append("inserted missing commas") s = body + "," + tail_ws fixed.append(s) out = "".join(fixed) # Trailing commas (safe now: strings are intact, commas here are structural) segs2 = _tokenize(out, []) parts: list[str] = [] changed = False for is_str, seg in segs2: if is_str: parts.append(seg) else: new = re.sub(r",(\s*[}\]])", r"\1", seg) changed = changed or new != seg parts.append(new) if changed: notes.append("removed trailing commas") return "".join(parts) def _extract_and_balance(text: str, notes: list[str]) -> str: """Slice out the JSON object (dropping surrounding prose) and close any unclosed braces/brackets.""" start = text.find("{") if start == -1: return text if text[:start].strip(): notes.append("ignored text before the JSON block") stack: list[str] = [] i, n = start, len(text) end = -1 while i < n: ch = text[i] if ch == '"': # skip strings i += 1 while i < n: if text[i] == "\\": i += 2 continue if text[i] == '"': break i += 1 elif ch in "{[": stack.append("}" if ch == "{" else "]") elif ch in "}]": if stack and stack[-1] == ch: stack.pop() if not stack: end = i + 1 break i += 1 if end != -1: if text[end:].strip(): notes.append("ignored text after the JSON block") return text[start:end] # Ran out of input with open scopes: close them. if stack: notes.append("closed unclosed braces/brackets") return text[start:] + "".join(reversed(stack)) return text[start:] def repair_json_text(text: str) -> tuple[str, list[str]]: """ Best-effort repair of an almost-JSON snippet. Returns (candidate, notes) where notes is a human-readable list of the fixes applied (deduplicated, in order). Does NOT guarantee the result parses — callers still try json.loads and surface its error if repair wasn't enough. """ notes: list[str] = [] t = _strip_fences(text, notes) t = _normalize_unicode(t, notes) t = t.strip() # Brace-less inner fragment: "name": { ... } (with no outer braces). # A quoted key at the start is a strong signal; a bare word only counts # when followed by '{' so prose like "Note: ..." isn't swallowed. if re.match(r'^\s*"[^"\n]+"\s*:\s*[{["]', t) or re.match(r"^\s*[A-Za-z_$][\w$.-]*\s*:\s*\{", t): notes.append("wrapped fragment in braces") t = "{" + t + "}" t = _extract_and_balance(t, notes) segs = _tokenize(t, notes) t = _repair_segments(segs, notes) # Re-balance in case comment/quote fixes exposed structure. t = _extract_and_balance(t, []) seen: set[str] = set() unique = [x for x in notes if not (x in seen or seen.add(x))] return t, unique def parse_pasted_json_verbose(text: str) -> tuple[dict[str, dict], list[str]]: """ Like parse_pasted_json, but forgiving. Tries strict JSON first; if that fails, runs repair_json_text() and retries. Returns (servers, repair_notes). repair_notes is empty when the input was already valid. """ raw = (text or "").strip() if not raw: raise ValueError("Nothing to parse.") try: return _shape_servers(json.loads(raw)), [] except json.JSONDecodeError as strict_err: candidate, notes = repair_json_text(raw) try: obj = json.loads(candidate) except json.JSONDecodeError: # Repair wasn't enough — report the original, more meaningful error. raise ValueError( f"Couldn't parse that as JSON even after auto-repair " f"(line {strict_err.lineno}, column {strict_err.colno}: {strict_err.msg})." ) from strict_err return _shape_servers(obj), notes def _shape_servers(obj) -> dict[str, dict]: """Shared shape-detection for parsed paste content.""" if not isinstance(obj, dict): raise ValueError("Top-level JSON must be an object ({ ... }).") if isinstance(obj.get("mcpServers"), dict): servers = obj["mcpServers"] elif looks_like_server(obj): servers = {suggest_name(obj): obj} elif obj and all(isinstance(v, dict) for v in obj.values()): servers = obj else: raise ValueError("Couldn't find any MCP server definitions in that JSON.") clean: dict[str, dict] = {} for name, data in servers.items(): if not looks_like_server(data): raise ValueError( f"'{name}' doesn't look like an MCP server (it needs a 'command' or a 'url')." ) clean[str(name)] = data return clean def parse_pasted_json(text: str) -> dict[str, dict]: """ Accept any of the shapes MCP docs hand out and return {name: server_dict}: 1. Full config: {"mcpServers": {"name": {...}}} 2. Inner block only: {"name": {"command": ...}} 3. A bare server obj: {"command": ..., "args": [...]} (name is suggested) Input doesn't have to be valid JSON — markdown fences, comments, trailing commas, smart/single quotes, unquoted keys, surrounding prose, and missing braces are repaired automatically (see repair_json_text). Raises ValueError with a human message on anything unrecognizable. """ servers, _notes = parse_pasted_json_verbose(text) return servers # --------------------------------------------------------------------------- # # Secrets: detection + redaction # --------------------------------------------------------------------------- # _SECRET_KEY_RE = re.compile( r"(?i)(token|secret|passw|api[-_]?key|apikey|auth|credential|bearer|private[-_]?key|access[-_]?key)" ) # Well-known token prefixes that identify a bare value as a secret even # without a telling key/flag name next to it. _TOKEN_PREFIXES = ( "ghp_", "gho_", "ghu_", "ghs_", "github_pat_", # GitHub "sk-", "sk_live_", "sk_test_", "rk_live_", # OpenAI / Stripe "xoxb-", "xoxp-", "xoxc-", "xoxs-", "xapp-", # Slack "glpat-", "gldt-", # GitLab "AKIA", "ASIA", # AWS access key IDs "ya29.", "AIza", # Google "pypi-", "npm_", "dop_v1_", "figd_", ) MASK = "••••••••" # Matches userinfo credentials embedded in a URL: scheme://user:pass@host # Fires on postgres://user:pass@host but NOT on https://host/path or ssh://user@host. _EMBEDDED_CRED_RE = re.compile(r"://[^:@/\s]+:[^:@/\s]+@") def is_secret_key(name: str) -> bool: """Does this env-var / header / flag name look like it holds a secret?""" return bool(_SECRET_KEY_RE.search(name or "")) def _is_secret_value(value: str) -> bool: return isinstance(value, str) and value.startswith(_TOKEN_PREFIXES) def redact_args(args: list[str]) -> list[str]: """ Mask secret values in an args list for display/diagnostics: --token abc123 -> --token •••••••• (value after a secret flag) --api-key=abc123 -> --api-key=•••••••• (inline flag=value) ghp_abc123 -> •••••••• (well-known token prefix) Everything else passes through untouched. """ out: list[str] = [] mask_next = False for a in args: s = str(a) if mask_next: out.append(MASK) mask_next = False continue if s.startswith("-") and "=" in s and is_secret_key(s.split("=", 1)[0]): out.append(s.split("=", 1)[0] + "=" + MASK) continue if s.startswith("-") and is_secret_key(s): out.append(s) mask_next = True continue if _is_secret_value(s): out.append(MASK) continue out.append(s) return out def args_secret_warning(data: dict) -> str | None: """ Return a warning string when any arg looks like a raw secret that would be better placed in `env`. Returns None when no concern is found. Skips --flag=value inline pairs (already partially self-documenting). Fires on: - positional values that start with a well-known token prefix (ghp_, sk-, …) - values that follow a secret-named flag (--token abc, --api-key abc) - URLs with embedded user:pass credentials (postgres://user:pass@host) """ args = [str(a) for a in (data.get("args") or [])] mask_next = False for a in args: if mask_next: mask_next = False if not a.startswith("-"): return ( "An arg value following a secret-named flag looks like a credential. " "Where the server supports it, prefer Environment variables — " "args are visible in process listings." ) continue # --flag=value inline: skip (the flag name already labels it) if a.startswith("-") and "=" in a: continue # --secretflag (no inline value): flag the next positional arg if a.startswith("-") and is_secret_key(a): mask_next = True continue if a.startswith("-"): continue # Positional value: check for token prefix or embedded URL credentials if _is_secret_value(a) or _EMBEDDED_CRED_RE.search(a): return ( "An arg value looks like a credential. " "Where the server supports it, prefer Environment variables — " "args are visible in process listings." ) return None def split_suspicious_args(args: list[str]) -> tuple[list[str], list[str]]: """ Detect the classic argument-entry mistake: several argv tokens typed on one line ("--directory /path/to/server"). An entry is only flagged when it contains whitespace AND at least one whitespace-separated token starts with "-" — so legitimate single arguments with spaces ("My Project Notes", "/Users/me/My Documents") are never touched. Returns (fixed_args, notes). notes is empty when nothing was suspicious; otherwise it describes each split so a UI can show the proposed fix. Quotes are respected when splitting ('--name "My Server"' becomes ['--name', 'My Server']). """ import shlex out: list[str] = [] notes: list[str] = [] for a in args: if isinstance(a, str) and _looks_like_multiple_args(a): try: parts = shlex.split(a) except ValueError: # unbalanced quotes — fall back to plain split parts = a.split() if len(parts) > 1: out.extend(parts) notes.append(f"“{a}” looks like {len(parts)} arguments — split onto separate lines") continue out.append(a) return out, notes def _looks_like_multiple_args(a: str) -> bool: toks = a.split() return len(toks) > 1 and any(t.startswith("-") for t in toks) # --------------------------------------------------------------------------- # # Validation # --------------------------------------------------------------------------- # def validate_servers(servers: list[ServerEntry]) -> list[str]: """Return a list of human-readable problems. Empty list == all good.""" problems: list[str] = [] seen: dict[str, int] = {} for s in servers: nm = s.name.strip() if not nm: problems.append("A server has an empty name.") seen[nm] = seen.get(nm, 0) + 1 if s.kind == "stdio": if not str(s.data.get("command", "")).strip(): problems.append(f"'{nm or '(unnamed)'}' is a local server but has no command.") else: url = str(s.data.get("url", "")).strip() if not url: problems.append(f"'{nm or '(unnamed)'}' is a remote server but has no URL.") elif not (url.startswith("http://") or url.startswith("https://")): problems.append(f"'{nm}' has a URL that isn't http(s).") for nm, count in seen.items(): if nm and count > 1: problems.append(f"Duplicate server name: '{nm}' ({count}x).") return problems def lint_server(name: str, data: dict) -> list[str]: """Return non-blocking structural warnings for a single server definition. Unlike validate_servers, nothing here blocks Save -- these are advisory notes about shapes that will round-trip through JSON fine but are probably not what the user intended (args given as a plain string instead of a list, an env value that isn't a string, an unrecognized `type`, unknown top-level fields, etc.). """ nm = name.strip() or "(unnamed)" warnings: list[str] = [] if "command" in data and not isinstance(data["command"], str): warnings.append(f"'{nm}': 'command' should be a string") if "args" in data: args = data["args"] if not isinstance(args, list): warnings.append(f"'{nm}': 'args' should be a list (one argument per item)") elif any(not isinstance(a, str) for a in args): warnings.append( f"'{nm}': 'args' contains non-string values " "(they will be saved as-is; Claude expects strings)" ) for field in ("env", "headers"): if field not in data: continue val = data[field] if not isinstance(val, dict): warnings.append(f"'{nm}': '{field}' should be an object of string key/value pairs") elif any(not isinstance(v, str) for v in val.values()): warnings.append( f"'{nm}': '{field}' contains non-string values " "(they will be saved as-is; Claude expects strings)" ) if "type" in data: t = data["type"] if t not in ("http", "sse", "stdio"): warnings.append(f"'{nm}': 'type' should be one of http, sse, stdio (found {t!r})") extra = sorted(k for k in data if k not in KNOWN_FIELDS) if extra: warnings.append(f"'{nm}': extra fields preserved as-is: {', '.join(extra)}") return warnings def lint_servers(servers: list[ServerEntry]) -> list[str]: """Concatenate lint_server warnings across every entry, in order.""" out: list[str] = [] for s in servers: out.extend(lint_server(s.name, s.data)) return out # --------------------------------------------------------------------------- # # Search / filter # --------------------------------------------------------------------------- # def server_matches_filter(entry: ServerEntry, query: str) -> bool: """ Case-insensitive substring match against a server's name, and its command (stdio) or url (remote). An empty/whitespace-only query matches everything -- that's what lets the search box double as "no filter". """ q = (query or "").strip().lower() if not q: return True if q in entry.name.lower(): return True if entry.kind == "remote": haystack = str(entry.data.get("url", "")) else: haystack = str(entry.data.get("command", "")) return q in haystack.lower() def filter_servers(entries: list[ServerEntry], query: str) -> list[ServerEntry]: """Return only the entries that match `query` (see server_matches_filter).""" return [e for e in entries if server_matches_filter(e, query)] # --------------------------------------------------------------------------- # # Dependency / PATH checking # --------------------------------------------------------------------------- # @functools.lru_cache(maxsize=1) def system_path() -> str: """ The PATH that a GUI app (e.g., Claude Desktop launched from Finder/dock) reliably inherits, independent of how BCC itself was started. This is used to determine the 'ok' vs 'warn' distinction: ok = found in system_path() → works however Claude is launched warn = found only in augmented_path() → works from a terminal but may not work when Claude is opened from the desktop macOS : /etc/paths + /etc/paths.d/* (what launchd provides) plus well-known package-manager prefixes (/opt/homebrew, /opt/local). Windows: system + user PATH from the registry. Linux : /etc/environment + standard FHS dirs + /snap/bin. """ dirs: list[str] = [] if sys.platform == "darwin": for p in ["/etc/paths", *sorted(glob.glob("/etc/paths.d/*"))]: with contextlib.suppress(OSError): dirs.extend(ln.strip() for ln in Path(p).read_text().splitlines() if ln.strip()) # Package-manager install prefixes: present on disk once installed, # accessible to all processes regardless of shell configuration. for d in ( "/opt/homebrew/bin", "/opt/homebrew/sbin", # Homebrew (Apple Silicon) "/usr/local/bin", "/usr/local/sbin", # Homebrew (Intel) / manual installs "/opt/local/bin", "/opt/local/sbin", # MacPorts ): dirs.append(d) elif os.name == "nt": try: import winreg for hive, sub in [ ( winreg.HKEY_LOCAL_MACHINE, r"SYSTEM\CurrentControlSet\Control\Session Manager\Environment", ), (winreg.HKEY_CURRENT_USER, r"Environment"), ]: try: with winreg.OpenKey(hive, sub) as k: val, _ = winreg.QueryValueEx(k, "PATH") dirs.extend(os.path.expandvars(val).split(os.pathsep)) except OSError: pass except ImportError: pass dirs.extend( [ "C:\\Windows\\System32", "C:\\Windows", "C:\\Windows\\System32\\Wbem", os.path.expandvars(r"%ProgramFiles%\nodejs"), ] ) else: # Linux / other try: for line in Path("/etc/environment").read_text().splitlines(): if line.upper().startswith("PATH="): dirs.extend(line[5:].strip("\"'").split(os.pathsep)) except OSError: pass for d in ("/usr/local/bin", "/usr/bin", "/bin", "/usr/sbin", "/sbin", "/snap/bin"): dirs.append(d) seen, out = set(), [] for d in dirs: d = d.strip() if d and d not in seen: seen.add(d) out.append(d) return os.pathsep.join(out) @functools.lru_cache(maxsize=1) def augmented_path() -> str: """ The widest PATH BCC searches when looking for a command — system_path() plus user-specific runtime locations (nvm, cargo, volta, bun, etc.) that require shell configuration to be active. Memoized for the session. Call refresh_path_cache() to re-scan, e.g. after the user installs a new runtime. """ base = system_path().split(os.pathsep) # Also include the current process PATH (catches unusual CI / container setups) env_parts = os.environ.get("PATH", "").split(os.pathsep) home = Path.home() user_dirs = [ str(home / ".local" / "bin"), str(home / ".cargo" / "bin"), str(home / ".deno" / "bin"), str(home / ".bun" / "bin"), str(home / ".volta" / "bin"), ] user_dirs += glob.glob(str(home / ".nvm" / "versions" / "node" / "*" / "bin")) user_dirs += glob.glob( str(home / ".local" / "share" / "fnm" / "node-versions" / "*" / "installation" / "bin") ) if os.name == "nt": appdata = os.environ.get("APPDATA", "") if appdata: user_dirs.append(str(Path(appdata) / "npm")) seen, ordered = set(), [] for p in base + env_parts + user_dirs: if p and p not in seen and Path(p).is_dir(): seen.add(p) ordered.append(p) return os.pathsep.join(ordered) def refresh_path_cache(): """Forget memoized PATHs so the next check re-scans (e.g. after an install).""" system_path.cache_clear() augmented_path.cache_clear() # Tailored install advice keyed by the runner's basename. RUNNER_HINTS = { "npx": "Node.js / npx not found. Install Node from nodejs.org or `brew install node`. " "With nvm, the binary lives at ~/.nvm/versions/node//bin.", "node": "Node.js not found. Install from nodejs.org or `brew install node`.", "uvx": "uv not found. Install with `curl -LsSf https://astral.sh/uv/install.sh | sh` " "(or `brew install uv`). uvx ships inside uv.", "uv": "uv not found. Install with `curl -LsSf https://astral.sh/uv/install.sh | sh` " "(or `brew install uv`).", "python": "Python not found on this PATH.", "python3": "Python 3 not found on this PATH.", "docker": "Docker not found. Install Docker Desktop and make sure it's running.", "bun": "Bun not found. Install with `curl -fsSL https://bun.sh/install | bash`.", "deno": "Deno not found. Install with `curl -fsSL https://deno.land/install.sh | sh`.", } def runner_hint(cmd: str) -> str: base = Path(cmd).name.lower() if base.endswith(".exe"): base = base[:-4] if base in RUNNER_HINTS: return RUNNER_HINTS[base] if ("/" in cmd) or ("\\" in cmd): return ( f"'{cmd}' looks like a file path, but it doesn't exist or isn't executable. " f"Check the path and that it's marked executable (chmod +x)." ) return f"'{cmd}' was not found on PATH. Install it, or put the full path to the executable in 'command'." SHELL_WRAPPERS = {"cmd", "cmd.exe", "sh", "bash", "zsh", "powershell", "powershell.exe", "pwsh"} def _commands_to_check(data: dict) -> list[str]: cmd = data.get("command") if not cmd: return [] cmds = [str(cmd)] base = Path(str(cmd)).name.lower() if base in SHELL_WRAPPERS: for a in data.get("args", []) or []: if isinstance(a, str) and not a.startswith(("/", "-")) and " " not in a: cmds.append(a) break return cmds def check_dependency(data: dict, path: str | None = None) -> dict: """ Decide whether a server's command can actually be found, and gather enough context to troubleshoot when it can't. status is one of: 'ok' - resolved on the normal (inherited) PATH; will work anywhere. 'warn' - resolved, but ONLY via an augmented location. Works in a terminal launch; a bundled .app launch of Claude may not see it. 'missing' - not found anywhere. 'remote' - a url-based server (no local command to check). 'unknown' - no command set. Returns: {status, label, resolved, in_base, searched_path, hints, commands, detail} """ if "url" in data and "command" not in data: return { "status": "remote", "label": "remote endpoint", "detail": data.get("url", ""), "resolved": {}, "in_base": {}, "searched_path": "", "hints": [], "commands": [], } cmds = _commands_to_check(data) if not cmds: return { "status": "unknown", "label": "no command set", "detail": "", "resolved": {}, "in_base": {}, "searched_path": "", "hints": ["This server has no 'command' to run."], "commands": [], } aug = path or augmented_path() resolved: dict[str, str | None] = {} in_base: dict[str, bool] = {} for c in cmds: if (os.sep in c) or ("/" in c): # explicit path, independent of PATH cp = Path(c) ok = cp.exists() and os.access(cp, os.X_OK) resolved[c] = str(cp) if ok else None in_base[c] = ok else: resolved[c] = shutil.which(c, path=aug) # 'in_base' means: found in system_path() — the PATH that Claude # Desktop (or any GUI app) reliably inherits regardless of shell config. # This check is consistent whether BCC runs from a terminal or as a .app. in_base[c] = bool(shutil.which(c, path=system_path())) missing = [c for c, r in resolved.items() if not r] hints: list[str] = [] if missing: for c in missing: hints.append(runner_hint(c)) status = "missing" label = "missing: " + ", ".join(missing) else: nonbase = [c for c in resolved if not in_base[c]] if nonbase: status = "warn" label = "found, but only on a non-standard PATH" for c in nonbase: hints.append( f"'{c}' was found at {resolved[c]}, but that location is not in the standard " f"system PATH. Claude Desktop launched from Finder or the dock may not see it. " f"Clicking 'Use full path ↳' rewrites the command to its absolute path, which " f"works regardless of how Claude is launched." ) else: status = "ok" label = "found: " + ", ".join(Path(p).name for p in resolved.values()) return { "status": status, "label": label, "resolved": resolved, "in_base": in_base, "searched_path": aug, "hints": hints, "commands": cmds, "detail": "", } def diagnostics_text(name: str, data: dict) -> str: """A copy-pasteable plain-text troubleshooting report for one server.""" res = check_dependency(data) L: list[str] = [] if res["status"] == "remote": L.append(f"Server: {name or '(unnamed)'} [remote]") L.append(f"URL: {data.get('url', '')}") if data.get("type"): L.append(f"Transport: {data['type']}") if data.get("headers"): L.append("Headers: " + ", ".join(data["headers"].keys())) L.append("Status: REMOTE (network reachability is not checked)") return "\n".join(L) L.append(f"Server: {name or '(unnamed)'} [local / stdio]") L.append(f"Command: {data.get('command', '')}") if data.get("args"): # Redacted: this report is designed to be pasted into bug reports. L.append("Args: " + " ".join(redact_args(list(data["args"])))) if data.get("env"): L.append("Env keys: " + ", ".join(data["env"].keys())) L.append(f"Status: {res['status'].upper()}") L.append("") L.append("Command resolution:") for c, r in res["resolved"].items(): if r: tag = "" if res["in_base"].get(c) else " <-- non-standard PATH; app may not see it" L.append(f" {c} -> {r}{tag}") else: L.append(f" {c} -> NOT FOUND") if res["hints"]: L.append("") L.append("Hints:") for h in res["hints"]: L.append(f" - {h}") sys_dirs = set(system_path().split(os.pathsep)) aug_dirs = res["searched_path"].split(os.pathsep) if res["searched_path"] else [] extra = [d for d in aug_dirs if d not in sys_dirs] L.append("") L.append(f"PATH searched ({len(aug_dirs)} dirs):") for d in aug_dirs: L.append((" + " if d in extra else " ") + d) if extra: L.append("") L.append( "(+ = user-specific dirs not in the standard system PATH. " "Claude Desktop launched from Finder may NOT have these.)" ) return "\n".join(L) def server_log_path(name: str) -> Path | None: """ The platform-specific Claude Desktop MCP server log file for `name`, or None if it doesn't exist yet (nothing has been logged for this server). macOS : ~/Library/Logs/Claude/mcp-server-.log (one file per server) Windows: %APPDATA%\\Claude\\logs\\mcp.log (one shared file) Other platforms: Claude Desktop doesn't ship a log in a known location -> None. """ if sys.platform == "darwin": p = Path.home() / "Library" / "Logs" / "Claude" / f"mcp-server-{name}.log" elif sys.platform.startswith("win"): appdata = Path(os.environ.get("APPDATA", Path.home() / "AppData" / "Roaming")) p = appdata / "Claude" / "logs" / "mcp.log" else: return None return p if p.is_file() else None _STDERR_CAP = 4096 # bytes def _kill_process_tree_windows(pid: int) -> None: """ Kill `pid` and its whole descendant tree via `taskkill /T /F` (issue #13). Popen.kill() only terminates the direct child; runner-style commands (npx → node → server, cmd → real process) leave the actual server alive, leaking a process on every Windows spawn test. taskkill walks the tree. """ flags = getattr(subprocess, "CREATE_NO_WINDOW", 0) # no console flash from the GUI exe with contextlib.suppress(OSError): subprocess.run( ["taskkill", "/PID", str(pid), "/T", "/F"], capture_output=True, creationflags=flags, ) def spawn_test(data: dict, timeout: float = 3.0) -> dict: """ Attempt to start a stdio server and observe it for `timeout` seconds. Returns a dict: outcome: "ok" — still running after timeout; server started correctly "exited" — exited with code 0 before timeout; unusual for a server "crashed" — exited with a non-zero code before timeout "not_found" — command could not be resolved to an executable "not_applicable" — remote server or no command; nothing to spawn "error" — unexpected internal failure while spawning/observing returncode: int | None stderr: str (first ~4 KB) detail: str Never raises: the GUI threads (Test launch / Test all) re-enable their buttons only when a result arrives, so an escaping exception would leave the UI stuck. Anything unexpected comes back as outcome "error". Run this off the UI thread — it blocks for up to `timeout` seconds. """ try: return _spawn_test_impl(data, timeout) except Exception as e: return { "outcome": "error", "returncode": None, "stderr": "", "detail": f"unexpected error: {e!r}", } def _spawn_test_impl(data: dict, timeout: float) -> dict: if "url" in data and "command" not in data: return { "outcome": "not_applicable", "returncode": None, "stderr": "", "detail": "remote server", } # str() first: pasted JSON can legally carry a non-string here and the # value never round-trips through the editor before a Test all run. cmd = str(data.get("command") or "").strip() if not cmd: return { "outcome": "not_applicable", "returncode": None, "stderr": "", "detail": "no command set", } # Resolve to absolute path so subprocess doesn't fight with PATH in env. resolved_cmd = shutil.which(cmd, path=augmented_path()) if resolved_cmd is None: return { "outcome": "not_found", "returncode": None, "stderr": "", "detail": f"command not found: {cmd}", } args_list = [resolved_cmd] + [str(a) for a in (data.get("args") or [])] merged_env = {**os.environ, "PATH": augmented_path()} # Popen rejects non-string env values; pasted JSON may carry numbers. merged_env.update({str(k): str(v) for k, v in (data.get("env") or {}).items()}) stderr_chunks: list[bytes] = [] def _drain(pipe) -> None: # Read until EOF so the pipe buffer never fills and blocks the subprocess. # Only keep the first _STDERR_CAP bytes; the rest is discarded. captured = 0 try: while True: chunk = pipe.read(1024) if not chunk: break if captured < _STDERR_CAP: keep = min(len(chunk), _STDERR_CAP - captured) stderr_chunks.append(chunk[:keep]) captured += keep except Exception: pass popen_kwargs: dict = dict( stdin=subprocess.PIPE, # keep open so servers block on read rather than seeing EOF stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, env=merged_env, ) if os.name != "nt": popen_kwargs["start_new_session"] = True # own process group → clean kill else: # The packaged app is windowed (console=False); without this every # spawn test of a console server flashes a console window. popen_kwargs["creationflags"] = getattr(subprocess, "CREATE_NO_WINDOW", 0) try: proc = subprocess.Popen(args_list, **popen_kwargs) except (FileNotFoundError, OSError) as e: return { "outcome": "not_found", "returncode": None, "stderr": "", "detail": str(e), } drain_thread = threading.Thread(target=_drain, args=(proc.stderr,), daemon=True) drain_thread.start() try: proc.wait(timeout=timeout) except subprocess.TimeoutExpired: try: if os.name != "nt": os.killpg(os.getpgid(proc.pid), _signal.SIGKILL) else: _kill_process_tree_windows(proc.pid) except OSError: pass with contextlib.suppress(subprocess.TimeoutExpired): proc.wait(timeout=1.0) drain_thread.join(0.5) stderr = b"".join(stderr_chunks).decode(errors="replace") return { "outcome": "ok", "returncode": None, "stderr": stderr, "detail": f"still running after {timeout:.0f}s — server started successfully", } drain_thread.join(0.5) stderr = b"".join(stderr_chunks).decode(errors="replace") rc = proc.returncode if rc == 0: return { "outcome": "exited", "returncode": rc, "stderr": stderr, "detail": "process exited cleanly (code 0) — unusual; a healthy server should keep running", } return { "outcome": "crashed", "returncode": rc, "stderr": stderr, "detail": f"process exited with code {rc}", } # --------------------------------------------------------------------------- # # Health status (maps a spawn_test() result to a simple tri-state for the # server-list UI's per-row status dot; see "Test all" in bcc.py) # --------------------------------------------------------------------------- # class HealthStatus: """ Tri-state health for the server-list status dot. A plain class of string constants -- not an Enum -- to match the plain-string status values used elsewhere in this module (see check_dependency's 'status'). """ UNTESTED = "untested" OK = "ok" FAILED = "failed" def health_from_spawn_result(result: dict) -> tuple[str, str]: """ Map a spawn_test() result dict to (HealthStatus, short_summary) for the server-list status column. Reuses spawn_test's own outcome classification rather than re-deriving pass/fail from returncode/stderr: outcome "ok" -> OK (server started and kept running) outcome "not_applicable" -> UNTESTED (remote server, or no command set) anything else -> FAILED (exited, crashed, or not found) The summary is short enough for a table cell/tooltip; when the process wrote to stderr before dying, its first line is appended for context. """ outcome = result.get("outcome", "") detail = result.get("detail", "") or "" stderr = (result.get("stderr") or "").strip() if outcome == "ok": return HealthStatus.OK, detail or "started" if outcome == "not_applicable": return HealthStatus.UNTESTED, detail or "not applicable" # exited / crashed / not_found: the server didn't come up cleanly. summary = detail or outcome if stderr: first_line = stderr.splitlines()[0].strip() if first_line: summary = f"{summary} — {first_line}" return HealthStatus.FAILED, summary def test_remote(url: str, timeout: float = 5.0) -> tuple[bool, str]: """ Reachability check for a url-based MCP server. ANY HTTP response (even 4xx/5xx) counts as reachable; only connection/DNS/timeout failures count as unreachable. Returns (ok, detail). Network call — run this off the UI thread. """ import socket import urllib.error import urllib.request url = (url or "").strip() if not (url.startswith("http://") or url.startswith("https://")): return False, "not an http(s) URL" req = urllib.request.Request( url, method="GET", headers={ "Accept": "text/event-stream, application/json, */*", "User-Agent": "BetterClaudeConfig/1.0", }, ) try: with urllib.request.urlopen(req, timeout=timeout) as r: return True, f"HTTP {r.status}" except urllib.error.HTTPError as e: return True, f"HTTP {e.code} (reachable)" except urllib.error.URLError as e: reason = getattr(e, "reason", e) if isinstance(reason, socket.timeout): return False, "timed out" return False, str(reason) except TimeoutError: return False, "timed out" except Exception as e: return False, str(e) def pin_command_path(data: dict, path: str | None = None) -> tuple[dict, str | None]: """ Fix a 'PATH-risk' (warn) server by rewriting the bare runner name to the absolute path it resolved to, so any process (incl. a bundled Claude.app) can find it. Returns (new_data, note). note is None if nothing was pinned. """ res = check_dependency(data, path) if res["status"] != "warn": return data, None out = dict(data) for c, resolved in res["resolved"].items(): if resolved and not res["in_base"].get(c, True) and ("/" not in c and "\\" not in c): if str(out.get("command", "")) == c: out["command"] = resolved return out, f"command → {resolved}" args = list(out.get("args", []) or []) for i, a in enumerate(args): if a == c: args[i] = resolved out["args"] = args return out, f"'{c}' → {resolved}" return data, None # --------------------------------------------------------------------------- # # Restart Claude Desktop (issue #9) # # Scoped strictly to Claude DESKTOP, the GUI app -- never Claude Code (the # CLI), which has no long-running process to bounce. Callers should gate this # behind profile_targets_claude_desktop() before offering it in the UI. # --------------------------------------------------------------------------- # class RestartResult(NamedTuple): """Outcome of a restart_claude_desktop() attempt.""" success: bool detail: str def _run_quiet(cmd: list[str]) -> None: """Best-effort fire-and-forget command. Never raises: a nonzero exit (e.g. pkill finding nothing to kill) is expected and not an error.""" with contextlib.suppress(OSError): subprocess.run(cmd, capture_output=True) def restart_supported() -> bool: """ True only where restarting Claude Desktop makes sense (macOS, Windows). There is no official Claude Desktop for Linux, and the obvious binary name there ("claude") is the Claude Code CLI — killing or spawning it would be actively harmful. GUI callers gate the Restart button on this. """ return sys.platform == "darwin" or sys.platform.startswith("win") _MACOS_QUIT_WAIT_S = 5.0 def _macos_claude_running() -> bool: try: return subprocess.run(["pgrep", "-x", "Claude"], capture_output=True).returncode == 0 except OSError: return False def _restart_claude_desktop_macos() -> RestartResult: _run_quiet(["pkill", "-x", "Claude"]) # Wait for the old instance to actually exit: `open -a` against a dying # process can merely re-activate it, and the config is only re-read on a # true relaunch. Blocks up to _MACOS_QUIT_WAIT_S — callers run this off # the UI thread (see RestartWorker in bcc.py). deadline = time.monotonic() + _MACOS_QUIT_WAIT_S while _macos_claude_running(): if time.monotonic() > deadline: return RestartResult( False, f"Claude Desktop didn't quit within {_MACOS_QUIT_WAIT_S:.0f}s — " "quit it manually, then reopen it.", ) time.sleep(0.15) try: result = subprocess.run(["open", "-a", "Claude"], capture_output=True, text=True) except OSError as e: return RestartResult(False, f"Couldn't launch Claude Desktop: {e}") if result.returncode != 0: detail = (result.stderr or result.stdout or "").strip() or "'open -a Claude' failed" return RestartResult(False, detail) return RestartResult(True, "Claude Desktop restarted.") def _claude_windows_start_menu_shortcut() -> Path: appdata = os.environ.get("APPDATA", str(Path.home())) return Path(appdata) / "Microsoft" / "Windows" / "Start Menu" / "Programs" / "Claude.lnk" def _restart_claude_desktop_windows() -> RestartResult: shortcut = _claude_windows_start_menu_shortcut() if not shortcut.is_file(): # Checked BEFORE killing: an MSIX/Store install has no Start-menu .lnk # at this path, and killing without a relaunch path would leave the # user with no running Claude at all. return RestartResult( False, f"Claude's Start-menu shortcut wasn't found ({shortcut}). " "If Claude Desktop is installed from the Microsoft Store, " "quit and reopen it manually.", ) _run_quiet(["taskkill", "/IM", "Claude.exe", "/F"]) try: # `cmd /c start "" ` launches detached, the same as double-clicking # the Start-menu shortcut, and returns immediately. result = subprocess.run( ["cmd", "/c", "start", "", str(shortcut)], capture_output=True, text=True ) except OSError as e: return RestartResult(False, f"Couldn't launch Claude Desktop: {e}") if result.returncode != 0: detail = ( result.stderr or result.stdout or "" ).strip() or "failed to relaunch Claude Desktop" return RestartResult(False, detail) return RestartResult(True, "Claude Desktop restarted.") def restart_claude_desktop() -> RestartResult: """ Kill and relaunch the Claude Desktop app so a freshly saved config takes effect. The app not currently running is NOT a failure -- pkill/taskkill exiting non-zero just means "nothing to kill", and we go straight to relaunching. Only a failed relaunch is reported as success=False. macOS blocks for up to _MACOS_QUIT_WAIT_S while the old instance exits — run off the UI thread. Unsupported platforms (see restart_supported()) refuse without touching any process. """ if sys.platform == "darwin": return _restart_claude_desktop_macos() 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"}) # Env var keys a catalog entry's config.env must never set. Every one of # these is a loader/interpreter override that lets a value walk straight # past CATALOG_ALLOWED_COMMANDS and the -e/--eval/-c deny-rule below: e.g. # NODE_OPTIONS="--require /tmp/x.js" turns an allowlisted `npx` entry into # arbitrary code execution without ever touching config.args, which is the # only field the allowlist/deny-rules/ASCII/secret checks used to cover. # Matched case-insensitively -- env keys are case-sensitive on POSIX, but a # `node_options` lookalike is exactly the kind of thing this must catch. CATALOG_DENIED_ENV_KEYS = frozenset( { "NODE_OPTIONS", "PYTHONSTARTUP", "PYTHONPATH", "PYTHONHOME", "LD_PRELOAD", "LD_LIBRARY_PATH", "DYLD_INSERT_LIBRARIES", "DYLD_LIBRARY_PATH", "BROWSER", "PATH", "NODE_REPL_EXTERNAL_MODULE", } ) # 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] = [ base64.b64decode("082NOwVB7uURkvfyS3+knJ+40Fk6C9unsF47+2uPKo4="), ] # 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=" 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=") # -style tokens the GUI must have the user fill in before Save. _PLACEHOLDER_RE = re.compile(r"<[^<>\s]+>") # A catalog entry's id becomes an mcpServers JSON key AND is interpolated # into Qt.AutoText widgets (status bar, QMessageBox) -- an id like # "Verified" renders as markup there. Not RCE, but UI spoofing, so # ids are constrained to a plain lowercase slug. _CATALOG_ID_RE = re.compile(r"^[a-z0-9][a-z0-9._-]{0,63}$") # Docker flags that consume the next arg as a value (so that value must not # be mistaken for the image reference when locating it in config.args). _DOCKER_VALUE_FLAGS = frozenset( { "-e", "--env", "-v", "--volume", "-p", "--publish", "--name", "-w", "--workdir", "-u", "--user", "--entrypoint", "--network", "--platform", "--add-host", "-l", "--label", } ) # 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 _catalog_package_spec_version(spec: str) -> str | None: """ Extract the version pin from an npm-style package spec, or None if the spec carries no pin. Handles unscoped "name@version" and scoped "@scope/name@version" -- scoped names have a leading "@" that is NOT the version separator, so a naive split on the first/only "@" misparses "@scope/pkg" (no version) as pinned to "scope/pkg". Splitting from the right side instead is safe for both forms because a package name may contain "@" only as the scope's leading character. """ if spec.startswith("@"): rest = spec[1:] if "@" not in rest: return None _, _, version = rest.rpartition("@") return version or None if "@" not in spec: return None _, _, version = spec.rpartition("@") return version or None def _catalog_package_spec_pinned(spec: str) -> bool: """ True if `spec` carries an exact version pin. Covers npm's "name@version" / "@scope/name@version" and uv's documented PyPI pin forms "name@version" and "name==version". """ if "==" in spec: _, _, version = spec.partition("==") return bool(version) return bool(_catalog_package_spec_version(spec)) def _first_catalog_package_spec(args: list[str]) -> str | None: """ The first arg that could plausibly BE a package spec: skip flags (leading "-") and tokens (which can't be validated and are filled in by the user later, never shipped by the catalog as the package name itself). Everything after the first hit is ignored -- trailing flags, paths, and placeholders are not package specs. """ for a in args: if a.startswith("-"): continue if _PLACEHOLDER_RE.fullmatch(a): continue return a return None def _catalog_pin_violations(tag: str, command: str, args: list[str]) -> list[str]: """ Version-pinning enforcement (finding #3): a catalog PR can otherwise ship `npx -y @scope/pkg` or `docker run img:latest` and the *next* resolve of that package/image is whatever the registry serves that day -- outside review, outside the signature's meaning. This is the only place that enforces pinning at runtime; catalog_review.py's risk_unpinned_package() is a maintainer-facing hint, not a gate. """ if command in ("npx", "uvx"): spec = _first_catalog_package_spec(args) if spec is None: return [f"{tag}: config.args must include a package spec to pin (e.g. name@1.2.3)."] if not _catalog_package_spec_pinned(spec): return [ f"{tag}: config.args package {spec!r} is not version-pinned; use " "name@version, @scope/name@version, or name==version." ] return [] if command == "docker": image = _docker_image_ref(args) if image is None: return [f"{tag}: config.args docker command has no image reference to pin."] _, sep, image_tag = image.rpartition(":") if not sep or "/" in image_tag: return [ f"{tag}: config.args docker image {image!r} has no explicit tag; " "pin an exact version (not 'latest', not untagged)." ] if image_tag == "latest": return [ f"{tag}: config.args docker image {image!r} uses the 'latest' tag, " "which is not allowed; pin an exact version." ] return [] return [] def _docker_image_ref(args: list[str]) -> str | None: """ Locate the image reference in a `docker run ...` args list: skip the "run" subcommand and any flags, including ones that consume the next token as a value (-e, -v, --name, ...) so that value isn't mistaken for the image. The first remaining positional token is the image. """ i = 0 if i < len(args) and args[i] == "run": i += 1 while i < len(args): a = args[i] if a.startswith("-"): if a in _DOCKER_VALUE_FLAGS and "=" not in a: i += 2 else: i += 1 continue return a return None 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))})." ) raw_args = config.get("args") args_ok = isinstance(raw_args, list) and all(isinstance(a, str) for a in raw_args) args = raw_args if args_ok else [] if not args_ok: problems.append(f"{tag}: config.args must be a list of strings.") 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)) # Version pinning (finding #3) -- only meaningful once command/args are # actually well-formed; a malformed args list already got its own # problem above and has nothing left to pin-check. if args_ok and command in ("npx", "uvx", "docker"): problems.extend(_catalog_pin_violations(tag, command, args)) env = config.get("env") if env is not None: env_ok = isinstance(env, dict) and all( isinstance(k, str) and isinstance(v, str) for k, v in env.items() ) if not env_ok: problems.append(f"{tag}: config.env must be an object of string values.") else: # config.env (finding #2): unlike args, env was previously # type-checked ONLY -- no allowlist, no deny-rule, no ASCII # check, no secret check. That made it the single easiest way # to smuggle a payload past every other guard in this # function: an allowlisted `command: npx` plus # NODE_OPTIONS=--require /tmp/x.js in env walks straight past # the command allowlist AND the -e/--eval/-c deny-rule above, # because neither of those ever looks at env. for key, value in env.items(): if not key.isascii(): problems.append( f"{tag}: config.env key {key!r} must be ASCII " "(non-ASCII code points rejected)." ) if key.upper() in CATALOG_DENIED_ENV_KEYS: problems.append( f"{tag}: config.env key {key!r} is on the catalog deny-list " "(interpreter/loader override) and is not allowed." ) if not value.isascii(): problems.append( f"{tag}: config.env value for {key!r} must be ASCII " "(non-ASCII code points rejected)." ) if _is_secret_value(value): problems.append( f"{tag}: config.env[{key!r}] looks like a real secret value; " "catalog entries must never ship secret values." ) if value != "" and not _PLACEHOLDER_RE.fullmatch(value): problems.append( f"{tag}: config.env[{key!r}] must be an empty string or a " "single token -- the catalog declares which env " "vars a server needs, it never supplies their 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).") elif not _CATALOG_ID_RE.match(entry_id): problems.append( f"{tag}: 'id' must match ^[a-z0-9][a-z0-9._-]{{0,63}}$ " "(it becomes an mcpServers JSON key and is interpolated into " "Qt.AutoText widgets)." ) 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 _verify_catalog_candidate(candidate: tuple[bytes, bytes] | None) -> tuple[dict | None, int]: """Verify+load+validate one (raw, sig) candidate. Returns (None, -1) on any failure.""" if not candidate: return None, -1 raw, sig = candidate if not verify_catalog_signature(raw, sig, CATALOG_PUBKEYS): return None, -1 try: data = load_catalog(raw) except (ValueError, TypeError): return None, -1 if validate_catalog(data): return None, -1 return data, catalog_version(data) def resolve_catalog( bundled: tuple[bytes, bytes] | None, cached: tuple[bytes, bytes] | None, remote: tuple[bytes, bytes] | None, floor: int = 0, ) -> dict: """ Pick the highest-version catalog among bundled/cached/remote. Each of bundled/cached/remote 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. `floor` is a pure, caller-supplied lower bound (e.g. a persisted "last accepted version" the GUI can load from disk and pass in) — this function does no storage of its own. Anti-rollback / anti-freeze, and WHY they apply to every candidate including the first one evaluated: the previous version of this function only ran these checks `if best_version >= 0`, i.e. once a candidate had already been accepted in this pass. That let the FIRST verified candidate through unconditionally — a signed catalog claiming version=999999999 sailed straight past both guards if it happened to be evaluated first, and rollback protection reset on every call anyway (nothing persisted across restarts). Now both guards are anchored to something that doesn't depend on iteration order: - The anti-freeze cap is measured against the BUNDLED catalog's version (verified independently, once), not against "whatever was accepted so far in this loop." Bundled ships inside the binary, so it's the one candidate that isn't attacker-supplied at resolve time — the natural trust anchor. If bundled itself doesn't verify, `floor` is the anchor instead. - The anti-rollback floor is max(floor, bundled's version), so a caller that persists `floor` across restarts gets real rollback protection; a caller that doesn't still gets "never below bundled." On a version TIE, the bundled candidate wins over cached/remote (it previously lost ties to whichever candidate happened to be evaluated last, silently preferring remote over bundled at equal version). Returns the winning catalog dict, or {} if nothing verified and validated. """ bundled_data, bundled_version = _verify_catalog_candidate(bundled) anchor = bundled_version if bundled_version >= 0 else floor min_accepted = max(floor, bundled_version if bundled_version >= 0 else 0) candidates = ( ("bundled", bundled_data, bundled_version), ("cached", *_verify_catalog_candidate(cached)), ("remote", *_verify_catalog_candidate(remote)), ) best: dict = {} best_version = -1 best_is_bundled = False for source, data, version in candidates: if data is None: continue if version < min_accepted: continue # anti-rollback / below the persisted floor if version > anchor + _CATALOG_MAX_VERSION_JUMP: continue # anti-freeze, capped against the bundled trust anchor is_bundled = source == "bundled" better = version > best_version or ( version == best_version and is_bundled and not best_is_bundled ) if better: best = data best_version = version best_is_bundled = is_bundled 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 -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 -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)