the_og/better-claude-config
1
0
Fork 0
Code Issues 8 Pull Requests 3 Actions Packages Projects Releases 4 Wiki Activity
Files
chore/68-key-rotation
better-claude-config/bcc_core.py
T
BCC Agent 4836c6cb48
CI / Lint (ruff) (pull_request) Successful in 9s
Details
CI / Tests (py3.10 / ubuntu-latest) (pull_request) Successful in 11s
Details
CI / Tests (py3.12 / ubuntu-latest) (pull_request) Successful in 10s
Details
CI / Tests (py3.12 / windows-latest) (pull_request) Successful in 35s
Details
CI / Tests (py3.13 / ubuntu-latest) (pull_request) Successful in 10s
Details
CI / Catalog signature (pull_request) Failing after 6s
Details
chore(#68): rotate signing keys, key-rotation re-attestation mode, harden show-seed-b64 
Wires the maintainer's freshly-rotated signing keys into BCC (issue #68
finding 5: the old key was shared between catalog+release and had been
exposed to CI), adds a key-rotation re-attestation mode to the Catalog
Console so the Console can actually re-sign under the new key, and
hardens the show-seed-b64 CLI prompt that led to a private key being
pasted into a chat.

## New keys (#68 finding 5)

- bcc_core.CATALOG_PUBKEYS -> new CATALOG pubkey
  (0s24PmkZcTT5yxNDdyTPHl5fyxArrHNPJKBjnXoQd8k=). Signs data/catalog.json,
  Console-only, offline.
- .github/workflows/ci.yml EXPECTED_CATALOG_PUBKEY_B64 -> same new
  catalog pubkey (the CI trust anchor added for #68 finding 4).
- scripts/sign_checksums.py RELEASE_PUBKEYS -> new RELEASE pubkey
  (6BnPgJEHJFyVltFoLTCNadIsehjy00iiW8IRlC1TfhA=). Signs SHA256SUMS only,
  CI-resident.
- README.md 'Verifying your download' / 'Signing keys' sections filled in
  with both pubkeys, explicit about which key is which.

The old key 082NOwVB7uURkvfyS3+knJ+40Fk6C9unsF47+2uPKo4= is retired (it
was shared and exposed to CI) and is deliberately NOT retained in either
trust list -- keeping a burned key in CATALOG_PUBKEYS would defeat the
point of rotating it.

## Key-rotation re-attestation mode (Catalog Console)

Problem: after the key swap, the existing data/catalog.json.sig (signed
with the OLD key) no longer verifies under the NEW CATALOG_PUBKEYS, but
catalog content is unchanged, so diff_catalogs(last_signed, current) is
empty -- and can_sign()'s empty-diff guard (load-bearing, #68 finding 1)
correctly refuses to sign an empty changeset. Without a rotation-aware
path, the Console could never re-sign and CI would stay red forever.

Fix: treat rotation as a full re-attestation, not a diff.

- catalog_review.py: ReviewSession/start_review gain reattest: bool =
  False. When set, changes is built via diff_catalogs(None, new_catalog)
  -- every entry presented as if newly added, requiring a fresh
  acknowledgement -- instead of diffing against old_catalog. can_sign()
  is UNCHANGED: it still refuses a genuinely-empty changeset and still
  enforces the TOCTOU blob-SHA pin and the blocking-risk check, because
  reattest sessions simply never produce an empty changeset (unless the
  catalog itself is empty).
- catalog_console.py: adds catalog_signature_valid_at(repo_dir, commit,
  raw), which calls bcc_core.verify_catalog_signature directly (never
  reimplemented) to detect whether the committed .sig verifies under the
  CURRENT CATALOG_PUBKEYS. ReviewWindow._on_load's source=main path uses
  this to decide reattest=True/False, and shows a loud, explicit red
  banner ("KEY ROTATION IN PROGRESS...") whenever reattest mode is
  entered -- never silent. Status text and Sign-button gating flow
  through the same can_sign()/all_entries_acknowledged() path as normal
  review.
- tests/test_catalog_review.py: 6 new tests covering re-attest mode
  (one change-entry per server, gating until all acknowledged, then
  permits), confirming normal mode still refuses an empty diff (rotation
  path is not a general bypass), and confirming reattest mode still
  enforces the blocking-risk check and the TOCTOU pin.

## show-seed-b64 hardening (Task 3)

The maintainer ran show-seed-b64 --release, saw an ambiguous prompt,
and pasted the printed PRIVATE seed into a chat believing it was public.

- cmd_show_seed_b64: passphrase prompt is now explicit ("Passphrase for
  the release signing key (the one YOU chose when generating it)"). A
  loud three-line warning banner prints to STDERR immediately before the
  seed ("!!! PRIVATE KEY BELOW..."); the seed itself stays alone on
  STDOUT so piping into pbcopy or a CI secret field still works cleanly.
- cmd_keygen: labels for both key kinds now say PUBLIC/PRIVATE explicitly
  and state safety properties inline (safe to commit vs. never commit),
  so the printed output can't be mistaken for the other key's.

## Verification

- ruff check . / ruff format --check .: clean.
- pytest: full suite green (360 passed, 1 skipped).
- Delete-the-check-and-watch-it-fail: each of can_sign()'s four gates
  (empty-diff, TOCTOU pin, blocking-risk, acknowledge-all) and the
  reattest branch itself were individually removed and confirmed to turn
  a test red, then restored -- see PR description for the exact
  failures.

Not touched: data/catalog.json (content-signed, maintainer's job via the
Console). No private key generated, requested, or committed. bcc.py
untouched (open PR #67).
2026-07-13 12:24:28 -04:00

2802 lines
105 KiB
Python
Raw Permalink Blame History

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