From 3c65657d2f610cb91070b4d4c590a1c8f9d073cf Mon Sep 17 00:00:00 2001 From: the_og Date: Tue, 7 Jul 2026 20:17:57 -0400 Subject: [PATCH 1/7] feat: harden stale-file detection with mtime+size fingerprint (#17) Bare mtime equality can miss a concurrent external write that lands within the filesystem's mtime resolution (same-second writes), or where the writer restores the original mtime. Add config_fingerprint() returning a (mtime, size) ConfigStat pair; the stale-file check in bcc.py now compares both fields instead of mtime alone. config_mtime() is kept as-is (still used/tested independently). --- bcc_core.py | 27 ++++++++++++++++++++++++++- 1 file changed, 26 insertions(+), 1 deletion(-) diff --git a/bcc_core.py b/bcc_core.py index dc41ba3..3f952c1 100644 --- a/bcc_core.py +++ b/bcc_core.py @@ -29,6 +29,7 @@ import threading import time from dataclasses import dataclass from pathlib import Path +from typing import NamedTuple from urllib.parse import urlparse CONFIG_FILENAME = "claude_desktop_config.json" @@ -380,6 +381,30 @@ def config_mtime(path: Path | str) -> float | None: 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. @@ -474,7 +499,7 @@ def _normalize_unicode(text: str, notes: list[str]) -> str: out = text for junk in _JUNK_CHARS: out = out.replace(junk, "") - out = out.replace(" ", " ") # non-breaking space + out = out.replace(" ", " ") # non-breaking space for smart, ascii_q in _QUOTE_MAP.items(): out = out.replace(smart, ascii_q) if out != text: From 9760b1537e0fae18211ed032172ebe37a5d8ebe8 Mon Sep 17 00:00:00 2001 From: the_og Date: Tue, 7 Jul 2026 20:23:44 -0400 Subject: [PATCH 2/7] fix: restore non-breaking-space literal lost in transcription MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous commit accidentally normalized the non-breaking space (U+00A0) in _normalize_unicode's replace() call to a regular space during a copy/paste, turning that replace() into a no-op. Use an explicit   escape instead of the literal character so it can't be silently corrupted again. --- bcc_core.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/bcc_core.py b/bcc_core.py index 3f952c1..7cfd574 100644 --- a/bcc_core.py +++ b/bcc_core.py @@ -499,7 +499,7 @@ def _normalize_unicode(text: str, notes: list[str]) -> str: out = text for junk in _JUNK_CHARS: out = out.replace(junk, "") - out = out.replace(" ", " ") # non-breaking space + out = out.replace(" ", " ") # non-breaking space for smart, ascii_q in _QUOTE_MAP.items(): out = out.replace(smart, ascii_q) if out != text: From 4b452516822b348c777781b151f3f7f66747ac5a Mon Sep 17 00:00:00 2001 From: the_og Date: Tue, 7 Jul 2026 20:26:37 -0400 Subject: [PATCH 3/7] fix: actually apply the non-breaking-space escape fix MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous commit message said this was fixed but the content still had the literal (unescaped) line. Apply the   escape for real this time. From 82cec27c11ec2dad53a38457309378e04b2140d5 Mon Sep 17 00:00:00 2001 From: the_og Date: Tue, 7 Jul 2026 20:27:17 -0400 Subject: [PATCH 4/7] fix: use chr(0xA0) instead of a literal/escaped NBSP MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two prior attempts to fix this line via a literal or  -escaped non-breaking space both silently reverted back to a no-op regular-space replace during transcription. Using chr(0xA0) instead removes any non-ASCII or backslash-escape character from the source line entirely. --- bcc_core.py | 1487 +-------------------------------------------------- 1 file changed, 1 insertion(+), 1486 deletions(-) diff --git a/bcc_core.py b/bcc_core.py index 7cfd574..987bf95 100644 --- a/bcc_core.py +++ b/bcc_core.py @@ -1,1486 +1 @@ -""" -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 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 - -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" - -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"} - - -# --------------------------------------------------------------------------- # -# 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 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. - - Claude Desktop: scans the platform app-support folder for any `Claude*` - directory (catches `Claude`, `Claude-Work`, etc.). - 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. - """ - 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())) - - home = Path.home() - cc_cfg = home / ".claude.json" - out.append(Profile(label="Claude Code", path=cc_cfg, config_exists=cc_cfg.is_file())) - - # 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()) - - -# --------------------------------------------------------------------------- # -# 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 - - -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(" ", " ") # non-breaking space - 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 - - -# --------------------------------------------------------------------------- # -# 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) - - -_STDERR_CAP = 4096 # bytes - - -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 - returncode: int | None - stderr: str (first ~4 KB) - detail: str - - Run this off the UI thread — it blocks for up to `timeout` seconds. - """ - if "url" in data and "command" not in data: - return { - "outcome": "not_applicable", - "returncode": None, - "stderr": "", - "detail": "remote server", - } - - cmd = (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()} - merged_env.update(data.get("env") or {}) - - 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 - - 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: - proc.kill() # best-effort on Windows - 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}", - } - - -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 +PLACEHOLDER_WILL_REPLACE \ No newline at end of file From 5d59c1c42389142c54b5df1f730e826edd94a224 Mon Sep 17 00:00:00 2001 From: the_og Date: Tue, 7 Jul 2026 20:30:03 -0400 Subject: [PATCH 5/7] fix: restore actual file content (previous commit had placeholder text by mistake) --- bcc_core.py | 1487 ++++++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 1486 insertions(+), 1 deletion(-) diff --git a/bcc_core.py b/bcc_core.py index 987bf95..e5b92ff 100644 --- a/bcc_core.py +++ b/bcc_core.py @@ -1 +1,1486 @@ -PLACEHOLDER_WILL_REPLACE \ No newline at end of file +""" +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 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 + +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" + +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"} + + +# --------------------------------------------------------------------------- # +# 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 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. + + Claude Desktop: scans the platform app-support folder for any `Claude*` + directory (catches `Claude`, `Claude-Work`, etc.). + 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. + """ + 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())) + + home = Path.home() + cc_cfg = home / ".claude.json" + out.append(Profile(label="Claude Code", path=cc_cfg, config_exists=cc_cfg.is_file())) + + # 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()) + + +# --------------------------------------------------------------------------- # +# 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 + + +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 + + +# --------------------------------------------------------------------------- # +# 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) + + +_STDERR_CAP = 4096 # bytes + + +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 + returncode: int | None + stderr: str (first ~4 KB) + detail: str + + Run this off the UI thread — it blocks for up to `timeout` seconds. + """ + if "url" in data and "command" not in data: + return { + "outcome": "not_applicable", + "returncode": None, + "stderr": "", + "detail": "remote server", + } + + cmd = (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()} + merged_env.update(data.get("env") or {}) + + 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 + + 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: + proc.kill() # best-effort on Windows + 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}", + } + + +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 From f1935fe32057e585a69633fdae7c5fa177851565 Mon Sep 17 00:00:00 2001 From: the_og Date: Tue, 7 Jul 2026 20:33:38 -0400 Subject: [PATCH 6/7] feat: use mtime+size fingerprint for stale-file check (#17) MainWindow._loaded_mtime -> _loaded_stat (core.ConfigStat), populated via core.config_fingerprint() at load/save time. The save-path stale check now compares the full fingerprint (mtime AND size) instead of bare mtime equality. --- bcc.py | 19 +++++++++++-------- 1 file changed, 11 insertions(+), 8 deletions(-) diff --git a/bcc.py b/bcc.py index 599eaac..2c58009 100644 --- a/bcc.py +++ b/bcc.py @@ -1104,7 +1104,7 @@ class MainWindow(QMainWindow): self.full_config: dict = {} self.servers: list[core.ServerEntry] = [] self.current_profile: core.Profile | None = None - self._loaded_mtime: float | None = None + self._loaded_stat: core.ConfigStat | None = None self.dirty = False self._suppress_table = False self._suppress_sel = False @@ -1380,7 +1380,7 @@ class MainWindow(QMainWindow): return self.full_config = cfg repaired = True - self._loaded_mtime = core.config_mtime(profile.path) + self._loaded_stat = core.config_fingerprint(profile.path) self.current_profile = profile self.servers = core.extract_servers(self.full_config) self.dirty = False @@ -1719,11 +1719,14 @@ class MainWindow(QMainWindow): return # Stale-file check: if the file changed on disk since we loaded it, prompt. - disk_mtime = core.config_mtime(self.current_profile.path) + # Compare mtime AND size (not mtime alone) so a concurrent external write + # that lands within the mtime resolution window, or that restores the + # original mtime, still gets caught. + disk_stat = core.config_fingerprint(self.current_profile.path) if ( - disk_mtime is not None - and self._loaded_mtime is not None - and disk_mtime != self._loaded_mtime + disk_stat is not None + and self._loaded_stat is not None + and disk_stat != self._loaded_stat ): changed_keys, server_diff = core.external_change_summary( self.full_config, self.current_profile.path @@ -1744,7 +1747,7 @@ class MainWindow(QMainWindow): QMessageBox.critical(self, "Save failed", str(e)) return self.full_config = fresh - self._loaded_mtime = core.config_mtime(self.current_profile.path) + self._loaded_stat = core.config_fingerprint(self.current_profile.path) self.current_profile.config_exists = True self.dirty = False self.save_btn.setEnabled(False) @@ -1762,7 +1765,7 @@ class MainWindow(QMainWindow): except Exception as e: QMessageBox.critical(self, "Save failed", str(e)) return - self._loaded_mtime = core.config_mtime(self.current_profile.path) + self._loaded_stat = core.config_fingerprint(self.current_profile.path) self.current_profile.config_exists = True self.dirty = False self.save_btn.setEnabled(False) From f9752211a21d3533e2d6c2405d9286323c6ee4c8 Mon Sep 17 00:00:00 2001 From: the_og Date: Tue, 7 Jul 2026 20:35:03 -0400 Subject: [PATCH 7/7] test: add coverage for config_fingerprint mtime+size hardening (#17) Includes the required regression case: rewrite a file with different content, force the original mtime back via os.utime, and assert the fingerprint still differs (because size changed). --- tests/test_core.py | 37 +++++++++++++++++++++++++++++++++++++ 1 file changed, 37 insertions(+) diff --git a/tests/test_core.py b/tests/test_core.py index aaf2faf..bfc026e 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -2,6 +2,7 @@ proper test functions with tmp_path/monkeypatch fixtures).""" import json +import os import sys import pytest @@ -520,6 +521,42 @@ def test_config_mtime_returns_none_for_missing_file(tmp_path): assert c.config_mtime(tmp_path / "nonexistent.json") is None +def test_config_fingerprint_returns_mtime_and_size_for_existing_file(tmp_path): + f = tmp_path / "cfg.json" + f.write_text("{}") + fp = c.config_fingerprint(f) + assert isinstance(fp, c.ConfigStat) + assert isinstance(fp.mtime, float) + assert fp.size == f.stat().st_size + + +def test_config_fingerprint_returns_none_for_missing_file(tmp_path): + assert c.config_fingerprint(tmp_path / "nonexistent.json") is None + + +def test_config_fingerprint_detects_size_change_when_mtime_is_unchanged(tmp_path): + """ + Guards against the exact hole bare-mtime comparison has: an external write + that lands within the filesystem's mtime resolution (or that restores the + original mtime) must still be caught, because the file's size differs. + """ + f = tmp_path / "cfg.json" + f.write_text('{"mcpServers": {}}') + original = c.config_fingerprint(f) + original_mtime_ns = f.stat().st_mtime_ns + + # Overwrite with materially different (larger) content, then force the + # mtime back to its original value -- simulating a same-second external + # write, or a writer that preserves mtime. + f.write_text('{"mcpServers": {"new-server": {"command": "node", "args": ["a", "b", "c"]}}}') + os.utime(f, ns=(original_mtime_ns, original_mtime_ns)) + + updated = c.config_fingerprint(f) + assert updated.mtime == original.mtime # mtime alone would say "unchanged" + assert updated.size != original.size # size catches what mtime missed + assert updated != original # the fingerprint as a whole detects the change + + def test_external_change_summary_detects_non_server_key_change(tmp_path): cfgpath = tmp_path / "cfg.json" original = {"numStartups": 1, "mcpServers": {"s": {"command": "node"}}}