Compare commits
11
Commits
cdda60da1b
...
v1.4.0
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f73b49f17c | ||
|
|
451e04cd96 | ||
|
|
675490e3c3 | ||
|
|
3768113938 | ||
|
|
583e4e4af0 | ||
|
|
38e9cf2b26 | ||
|
|
d4ce2647ae | ||
|
|
ece8c99f79 | ||
|
|
66b0101dea | ||
|
|
c5a6bdd1d1 | ||
|
|
ded1eef2dd |
@@ -3,13 +3,16 @@ name: CI
|
||||
on:
|
||||
push:
|
||||
branches: [main]
|
||||
paths-ignore: ["**/*.md"]
|
||||
pull_request:
|
||||
paths-ignore: ["**/*.md"]
|
||||
workflow_dispatch:
|
||||
|
||||
jobs:
|
||||
lint:
|
||||
runs-on: ubuntu-latest
|
||||
name: Lint (ruff)
|
||||
timeout-minutes: 10
|
||||
steps:
|
||||
- name: Checkout
|
||||
uses: actions/checkout@v4
|
||||
@@ -31,6 +34,11 @@ jobs:
|
||||
test:
|
||||
runs-on: ${{ matrix.os }}
|
||||
name: Tests (py${{ matrix.python }} / ${{ matrix.os }})
|
||||
# Guards against a job that hangs mid-run (e.g. a wedged test). Note: this
|
||||
# counts from when a runner PICKS UP the job, so it does not rescue a job
|
||||
# stuck "Waiting to run" because the self-hosted Windows runner is offline —
|
||||
# for that, bring the runner back or skip via paths-ignore (docs).
|
||||
timeout-minutes: 15
|
||||
strategy:
|
||||
fail-fast: false
|
||||
matrix:
|
||||
@@ -85,6 +93,7 @@ jobs:
|
||||
catalog-signature:
|
||||
name: Catalog signature
|
||||
runs-on: ubuntu-latest
|
||||
timeout-minutes: 10
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
|
||||
|
||||
@@ -0,0 +1,91 @@
|
||||
# Changelog
|
||||
|
||||
All notable changes to **BetterClaudeConfig** are recorded here. The format is
|
||||
based on [Keep a Changelog](https://keepachangelog.com/), and this project
|
||||
follows [Semantic Versioning](https://semver.org/): breaking changes bump the
|
||||
major, new features bump the minor, fixes bump the patch.
|
||||
|
||||
**When you open a PR, add a line under `[Unreleased]`.** At release time, that
|
||||
section is renamed to the new version + date and a fresh `[Unreleased]` is
|
||||
started.
|
||||
|
||||
## [Unreleased]
|
||||
|
||||
## [1.4.0] — 2026-08-13
|
||||
|
||||
> ⚠️ **Using the `ssh-mcp` server?** Upstream `ssh-mcp` shipped a **breaking
|
||||
> v2**: it removed the `--password`, `--sudoPassword`, `--suPassword` and
|
||||
> `--disableSudo` command-line flags, and it now reads a `config.toml` sidecar
|
||||
> file that overrides your command-line arguments. A config written for v1
|
||||
> crashes on v2's startup. This release adds tools to detect and fix that.
|
||||
> **BetterClaudeConfig itself has no breaking changes** — your existing configs
|
||||
> keep working; nothing is changed without your click.
|
||||
|
||||
### Added
|
||||
- **ssh-mcp v2 flag migration.** Detects the credential flags v2 removed and
|
||||
one-click-moves each into the environment variable ssh-mcp now reads
|
||||
(`--password` → `SSH_MCP_PASSWORD`; `--sudoPassword` / `--suPassword` →
|
||||
`SSH_MCP_SUDO_PASSWORD`). Also flags `--maxChars=none`, whose meaning changed
|
||||
between versions.
|
||||
- **Sidecar-config awareness.** When a server reads a separate config file
|
||||
(ssh-mcp's `config.toml`), BCC shows where that file actually lives on your
|
||||
platform and warns when your command-line args are **inert** because the file
|
||||
takes precedence — including when a file sits at the wrong, documented-but-
|
||||
unused path.
|
||||
- **In-app sidecar editor.** Edit that external config from inside BCC —
|
||||
pick-lists for known fields, a raw-text fallback — written atomically with a
|
||||
timestamped backup and locked to `0600`. No dropping to a terminal.
|
||||
- **Live hot-reload.** External changes to a sidecar (or to your Claude config)
|
||||
now surface without restarting BCC.
|
||||
- **Version pinning for `npx` servers.** Spots servers launched with `-y` /
|
||||
`@latest` that resolve a fresh version every run, shows the version currently
|
||||
resolved, and offers a one-click **Pin to this version** — with a drift note
|
||||
when a pin has fallen behind.
|
||||
- **Permission pre-flight.** Warns when a credential-bearing config is readable
|
||||
by other users on the machine and offers a one-click fix to `0600`/`0700`.
|
||||
- **Light theme + system-following** (the dark theme is preserved exactly).
|
||||
- **"Move to environment variable."** Convert a plaintext secret in a config
|
||||
into a `${VAR}` reference in place — offered only on clients that actually
|
||||
expand references, so it can't silently break a Claude Desktop config.
|
||||
- **Cross-client foundation.** Claude Desktop and Claude Code now flow through a
|
||||
single adapter — groundwork for supporting more clients.
|
||||
|
||||
### Changed
|
||||
- The update checker is now **visible** — a persistent banner plus a Help-menu
|
||||
item — instead of being buried in the About dialog.
|
||||
- Config writes share one atomic-write path; the temp file is created `0600`, so
|
||||
a secret is never briefly world-readable mid-write.
|
||||
|
||||
### Fixed
|
||||
- Loading a config with a non-object server value no longer crashes, and named
|
||||
server sets survive an external-change merge.
|
||||
- Project profiles that share a directory basename are disambiguated, so you
|
||||
can't accidentally edit the wrong `.mcp.json`.
|
||||
|
||||
### Internal / maintainer
|
||||
- Signed-catalog core and a maintainer-only **Catalog Console** (review + sign),
|
||||
with hardening of the review gate and a split of the signing keys. No
|
||||
user-facing catalog browser ships yet.
|
||||
|
||||
## [1.3.0] — 2026-07-12
|
||||
Named server sets, Claude Code project `.mcp.json` discovery, structural schema
|
||||
lint, and UX polish (Ctrl+S to save, enable-all / disable-all).
|
||||
|
||||
## [1.2.1] — 2026-07-12
|
||||
First shipped binaries: app-icon fix, a batch of audit fixes, and Windows
|
||||
process-tree cleanup on spawn-tests.
|
||||
|
||||
## [1.2.0] — 2026-07-08
|
||||
Server log viewer, duplicate-name conflict handling, a Restart-Claude button,
|
||||
stale-file protection, an About dialog, and a notify-only update checker.
|
||||
|
||||
## [1.1.0] — 2026-07-04
|
||||
Backup / restore UI and secret masking.
|
||||
|
||||
## [1.0.1] — 2026-06-29
|
||||
## [1.0.0] — 2026-06-29
|
||||
Initial releases: the core `mcpServers` editor with the lenient paste/repair
|
||||
pipeline that tolerates malformed JSON.
|
||||
|
||||
<!-- Backfill for 1.0.0–1.3.0 is summarised from release notes; the Unreleased
|
||||
section onward is maintained per-PR. -->
|
||||
@@ -16,7 +16,18 @@ import time
|
||||
from pathlib import Path
|
||||
from typing import ClassVar
|
||||
|
||||
from PySide6.QtCore import QRect, QSettings, QSize, Qt, QThread, QTimer, QUrl, Signal
|
||||
from PySide6.QtCore import (
|
||||
QEvent,
|
||||
QFileSystemWatcher,
|
||||
QRect,
|
||||
QSettings,
|
||||
QSize,
|
||||
Qt,
|
||||
QThread,
|
||||
QTimer,
|
||||
QUrl,
|
||||
Signal,
|
||||
)
|
||||
from PySide6.QtGui import (
|
||||
QAction,
|
||||
QActionGroup,
|
||||
@@ -38,8 +49,10 @@ from PySide6.QtWidgets import (
|
||||
QDialog,
|
||||
QDialogButtonBox,
|
||||
QFileDialog,
|
||||
QFormLayout,
|
||||
QFrame,
|
||||
QGridLayout,
|
||||
QGroupBox,
|
||||
QHBoxLayout,
|
||||
QHeaderView,
|
||||
QInputDialog,
|
||||
@@ -52,6 +65,7 @@ from PySide6.QtWidgets import (
|
||||
QMessageBox,
|
||||
QPlainTextEdit,
|
||||
QPushButton,
|
||||
QSpinBox,
|
||||
QSplitter,
|
||||
QStackedWidget,
|
||||
QStyledItemDelegate,
|
||||
@@ -922,13 +936,23 @@ class ServerEditor(QFrame):
|
||||
v.addLayout(rf_row)
|
||||
# Shown when a server reads a config sidecar (ssh-mcp's TOML): the args
|
||||
# may be inert, or a file may sit at the README path the server never
|
||||
# reads. Read-only advisory (#91) — no auto-fix; editing the sidecar is
|
||||
# a separate, deliberate action.
|
||||
# reads. Advisory (#91); #102 adds an "Edit config…" button that opens
|
||||
# the in-app sidecar editor whenever the server has a resolvable sidecar.
|
||||
self.sidecar_warn = QLabel("")
|
||||
self.sidecar_warn.setStyleSheet(f"color: {WARN};")
|
||||
self.sidecar_warn.setWordWrap(True)
|
||||
self.sidecar_warn.hide()
|
||||
v.addWidget(self.sidecar_warn)
|
||||
self.sidecar_edit_btn = QPushButton("Edit config…")
|
||||
self.sidecar_edit_btn.setToolTip(
|
||||
"Open this server's external config file in BCC (edit → atomic save "
|
||||
"with backup → the file is tightened to 0600)"
|
||||
)
|
||||
self.sidecar_edit_btn.clicked.connect(self._edit_sidecar)
|
||||
self.sidecar_edit_btn.hide()
|
||||
sc_row = QHBoxLayout()
|
||||
sc_row.addWidget(self.sidecar_warn, 1)
|
||||
sc_row.addWidget(self.sidecar_edit_btn, 0, Qt.AlignmentFlag.AlignTop)
|
||||
v.addLayout(sc_row)
|
||||
# Shown when the sidecar config is group/other-accessible (#93): ssh-mcp
|
||||
# refuses to start unless it's 0600 / its dir 0700. One-click chmod fix.
|
||||
# POSIX only — hidden on Windows where modes don't apply.
|
||||
@@ -1125,6 +1149,17 @@ class ServerEditor(QFrame):
|
||||
self.refresh_dependency(auto_open=False)
|
||||
self._check_args()
|
||||
|
||||
def recheck_advisories(self):
|
||||
"""Re-run the read-only sidecar / permission advisories (#101 hot-reload).
|
||||
|
||||
Public entry point for the file-watcher and focus-in fallback: it only
|
||||
recomputes the warning labels from the current form + the filesystem — it
|
||||
never touches the form's field values, so it is safe to call regardless of
|
||||
unsaved edits (it can't clobber them). Delegates to the same recompute the
|
||||
editor runs on every field change.
|
||||
"""
|
||||
self._check_args()
|
||||
|
||||
# --- args sanity check ------------------------------------------------ #
|
||||
def _current_arg_lines(self) -> list[str]:
|
||||
return [ln for ln in self.args.toPlainText().splitlines() if ln.strip() != ""]
|
||||
@@ -1138,6 +1173,7 @@ class ServerEditor(QFrame):
|
||||
self.removed_flag_warn.hide()
|
||||
self.removed_flag_fix_btn.hide()
|
||||
self.sidecar_warn.hide()
|
||||
self.sidecar_edit_btn.hide()
|
||||
self.perm_warn.hide()
|
||||
self.perm_fix_btn.hide()
|
||||
return
|
||||
@@ -1180,6 +1216,9 @@ class ServerEditor(QFrame):
|
||||
self.sidecar_warn.show()
|
||||
else:
|
||||
self.sidecar_warn.hide()
|
||||
# Offer the in-app editor (#102) whenever this server has a resolvable
|
||||
# sidecar — even before the file exists, so it can be created from BCC.
|
||||
self.sidecar_edit_btn.setVisible(core.sidecar_status(stdio) is not None)
|
||||
# Sidecar filesystem permissions (#93). Real platform/fs; no-op on Windows.
|
||||
perm_warnings = core.sidecar_permission_warnings(stdio)
|
||||
if perm_warnings:
|
||||
@@ -1226,6 +1265,35 @@ class ServerEditor(QFrame):
|
||||
return
|
||||
self._check_args() # re-check; the warning clears when perms are now tight
|
||||
|
||||
def _edit_sidecar(self):
|
||||
"""Open the in-app sidecar editor for this server's external config (#102)."""
|
||||
stdio = {
|
||||
"command": self.command.text().strip(),
|
||||
"args": self._current_arg_lines(),
|
||||
"env": self.env.dump(),
|
||||
}
|
||||
status = core.sidecar_status(stdio)
|
||||
spec = core.resolve_server_spec(stdio)
|
||||
if status is None or spec is None:
|
||||
return
|
||||
path = status["path"]
|
||||
try:
|
||||
text = path.read_text(encoding="utf-8") if status["exists"] else ""
|
||||
except OSError as e:
|
||||
QMessageBox.warning(self.window(), "Can't open config", f"{path}\n\n{e}")
|
||||
return
|
||||
dlg = SidecarEditorDialog(self.window(), spec.package, path, spec.schema, text)
|
||||
if dlg.exec() and dlg.saved:
|
||||
# Re-run detection through the same read-only path the #101 watcher
|
||||
# uses, so "args inert" / permission advisories update live.
|
||||
self._check_args()
|
||||
bnote = f" · backup: {dlg.backup_path.name}" if dlg.backup_path else " · (new file)"
|
||||
QMessageBox.information(
|
||||
self.window(),
|
||||
"Config saved",
|
||||
f"Saved {path}{bnote}\nThe file was tightened to 0600.",
|
||||
)
|
||||
|
||||
# --- dependency ------------------------------------------------------ #
|
||||
def refresh_dependency(self, auto_open=False):
|
||||
if not self.isEnabled():
|
||||
@@ -2140,6 +2208,176 @@ class NoticeBanner(QFrame):
|
||||
self.show()
|
||||
|
||||
|
||||
_UNSET_CHOICE = "— (unset) —" # sentinel entry for an enum pick-list
|
||||
|
||||
|
||||
class SidecarEditorDialog(QDialog):
|
||||
"""Edit an external sidecar config (ssh-mcp's config.toml) from inside BCC (#102).
|
||||
|
||||
A minimal, reviewable first pass: pick-lists for the schema enum fields and a
|
||||
numeric spinner for the port, plus a raw-text editor that is the full,
|
||||
always-available fallback. On Save the changed managed fields are applied
|
||||
surgically on top of the raw text (so comments/unknown keys survive), the
|
||||
managed values are validated against the ServerSpec schema, and the file is
|
||||
written through core.write_sidecar (atomic + timestamped backup + chmod
|
||||
0600). All decision logic lives in bcc_core; this class is wiring.
|
||||
"""
|
||||
|
||||
def __init__(self, parent, package: str, path: Path, schema: dict, initial_text: str):
|
||||
super().__init__(parent)
|
||||
self.setWindowTitle(f"Edit {package} config")
|
||||
self.resize(620, 620)
|
||||
self._path = Path(path)
|
||||
self._schema = schema or {}
|
||||
self.backup_path: Path | None = None # set on a successful save
|
||||
self.saved = False
|
||||
|
||||
v = QVBoxLayout(self)
|
||||
info = QLabel(
|
||||
f"Editing <code>{path}</code><br>"
|
||||
"Pick-lists cover the known settings; the raw editor below is the full "
|
||||
"file. Saving writes atomically, keeps a timestamped backup, and tightens "
|
||||
"the file to <code>0600</code>."
|
||||
)
|
||||
info.setObjectName("muted")
|
||||
info.setWordWrap(True)
|
||||
info.setTextFormat(Qt.TextFormat.RichText)
|
||||
v.addWidget(info)
|
||||
|
||||
# Which section the pick-lists target. ssh-mcp configs are section-based
|
||||
# ([server] / per-profile); default to the first section, else top-level.
|
||||
sec_row = QHBoxLayout()
|
||||
sec_row.addWidget(QLabel("Section:"))
|
||||
self.section_combo = QComboBox()
|
||||
sections = core.toml_sections(initial_text) or [""]
|
||||
for s in sections:
|
||||
self.section_combo.addItem("(top level)" if s == "" else s, s)
|
||||
# Prefer a section literally called "server" if present.
|
||||
if "server" in sections:
|
||||
self.section_combo.setCurrentIndex(sections.index("server"))
|
||||
self.section_combo.currentIndexChanged.connect(self._reload_fields_from_raw)
|
||||
sec_row.addWidget(self.section_combo, 1)
|
||||
v.addLayout(sec_row)
|
||||
|
||||
# Managed schema fields.
|
||||
self._fields_box = QGroupBox("Known settings")
|
||||
self._form = QFormLayout(self._fields_box)
|
||||
self._enum_widgets: dict[str, QComboBox] = {}
|
||||
self._port_widget: QSpinBox | None = None
|
||||
self._port_present = QCheckBox("set") # gate for whether port is written
|
||||
self._build_fields()
|
||||
v.addWidget(self._fields_box)
|
||||
|
||||
v.addWidget(QLabel("Raw TOML (full file — edit anything here):"))
|
||||
self.raw = QPlainTextEdit()
|
||||
self.raw.setPlainText(initial_text)
|
||||
mono = self.raw.font()
|
||||
mono.setFamily("Menlo, Consolas, monospace")
|
||||
self.raw.setFont(mono)
|
||||
v.addWidget(self.raw, 1)
|
||||
|
||||
self.err = QLabel("")
|
||||
self.err.setStyleSheet(f"color: {BAD};")
|
||||
self.err.setWordWrap(True)
|
||||
self.err.hide()
|
||||
v.addWidget(self.err)
|
||||
|
||||
btns = QDialogButtonBox()
|
||||
self.save_btn = btns.addButton("Save", QDialogButtonBox.ButtonRole.AcceptRole)
|
||||
self.save_btn.setObjectName("primary")
|
||||
btns.addButton(QDialogButtonBox.StandardButton.Cancel)
|
||||
btns.accepted.connect(self._save)
|
||||
btns.rejected.connect(self.reject)
|
||||
v.addWidget(btns)
|
||||
|
||||
self._reload_fields_from_raw()
|
||||
|
||||
def _current_section(self) -> str | None:
|
||||
s = self.section_combo.currentData()
|
||||
return None if s == "" else s
|
||||
|
||||
def _build_fields(self):
|
||||
"""Create a widget per schema key (enums -> combo, port range -> spin)."""
|
||||
for key, rule in self._schema.items():
|
||||
if isinstance(rule, list): # enum
|
||||
combo = QComboBox()
|
||||
combo.addItem(_UNSET_CHOICE, None)
|
||||
for opt in rule:
|
||||
combo.addItem(str(opt), opt)
|
||||
self._enum_widgets[key] = combo
|
||||
self._form.addRow(f"{key}:", combo)
|
||||
elif isinstance(rule, dict) and "min" in rule and "max" in rule: # numeric range
|
||||
row = QHBoxLayout()
|
||||
spin = QSpinBox()
|
||||
spin.setRange(int(rule["min"]), int(rule["max"]))
|
||||
spin.setEnabled(False)
|
||||
self._port_present.toggled.connect(spin.setEnabled)
|
||||
row.addWidget(self._port_present)
|
||||
row.addWidget(spin, 1)
|
||||
holder = QWidget()
|
||||
holder.setLayout(row)
|
||||
self._port_widget = spin
|
||||
self._form.addRow(f"{key}:", holder)
|
||||
|
||||
def _reload_fields_from_raw(self):
|
||||
"""Populate the pick-lists from the raw text for the chosen section, and
|
||||
remember the loaded state so Save only applies fields the user changed."""
|
||||
values = core.read_toml_section(self.raw.toPlainText(), self._current_section())
|
||||
self._loaded: dict = {}
|
||||
for key, combo in self._enum_widgets.items():
|
||||
val = values.get(key)
|
||||
idx = combo.findData(val) if val is not None else 0
|
||||
combo.setCurrentIndex(idx if idx >= 0 else 0)
|
||||
# An out-of-enum current value can't be shown; leave it at (unset) but
|
||||
# DON'T record it as loaded so we never silently overwrite it on save.
|
||||
self._loaded[key] = combo.currentData()
|
||||
if self._port_widget is not None:
|
||||
pv = values.get("port")
|
||||
has = isinstance(pv, int) and not isinstance(pv, bool)
|
||||
self._port_present.setChecked(has)
|
||||
if has:
|
||||
self._port_widget.setValue(pv)
|
||||
self._loaded["port"] = pv if has else None
|
||||
|
||||
def _managed_updates(self) -> dict:
|
||||
"""The managed keys the user actually changed -> new value (None = delete).
|
||||
|
||||
Only changed fields are applied, so untouched keys (including any the
|
||||
pick-list can't represent) are left exactly as the raw text has them.
|
||||
"""
|
||||
updates: dict = {}
|
||||
for key, combo in self._enum_widgets.items():
|
||||
cur = combo.currentData()
|
||||
if cur != self._loaded.get(key):
|
||||
updates[key] = cur # None here means "delete the key"
|
||||
if self._port_widget is not None:
|
||||
cur = self._port_widget.value() if self._port_present.isChecked() else None
|
||||
if cur != self._loaded.get("port"):
|
||||
updates["port"] = cur
|
||||
return updates
|
||||
|
||||
def _save(self):
|
||||
section = self._current_section()
|
||||
updates = self._managed_updates()
|
||||
# Validate only the managed values the user is actually setting (a None =
|
||||
# delete needs no enum check).
|
||||
to_check = {k: v for k, v in updates.items() if v is not None}
|
||||
problems = core.validate_sidecar_values(to_check, self._schema)
|
||||
if problems:
|
||||
self.err.setText("⚠ " + "\n".join(problems))
|
||||
self.err.show()
|
||||
return
|
||||
try:
|
||||
new_text = core.update_toml(self.raw.toPlainText(), updates, section=section)
|
||||
self.backup_path = core.write_sidecar(self._path, new_text)
|
||||
except Exception as e: # surface any write/permission failure in-place
|
||||
self.err.setText(f"⚠ Could not save: {e}")
|
||||
self.err.show()
|
||||
return
|
||||
self.saved = True
|
||||
self.accept()
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Restart worker: core.restart_claude_desktop() blocks up to ~5 s on macOS
|
||||
# waiting for the old instance to exit, so it must run off the UI thread.
|
||||
@@ -2180,6 +2418,18 @@ class MainWindow(QMainWindow):
|
||||
self._test_all_done = 0
|
||||
self._health_tester: SpawnTester | None = None
|
||||
|
||||
# Hot-reload (#101): watch the selected server's external sidecar path(s)
|
||||
# and BCC's own loaded config, so changes made outside BCC surface without
|
||||
# a restart. The watcher fires the debounce timer; the timer re-checks.
|
||||
self._fs_watcher = QFileSystemWatcher(self)
|
||||
self._fs_watcher.fileChanged.connect(self._on_fs_signal)
|
||||
self._fs_watcher.directoryChanged.connect(self._on_fs_signal)
|
||||
self._fs_debounce = QTimer(self)
|
||||
self._fs_debounce.setSingleShot(True)
|
||||
self._fs_debounce.setInterval(300) # coalesce a burst of writes
|
||||
self._fs_debounce.timeout.connect(self._recheck_external_state)
|
||||
self._sidecar_fp: tuple | None = None # last observed sidecar state
|
||||
|
||||
central = QWidget()
|
||||
self.setCentralWidget(central)
|
||||
root = QVBoxLayout(central)
|
||||
@@ -2202,6 +2452,12 @@ class MainWindow(QMainWindow):
|
||||
self.update_banner = NoticeBanner(self)
|
||||
root.addWidget(self.update_banner)
|
||||
|
||||
# External-change notice (#101): when BCC's own loaded config is edited
|
||||
# outside BCC, surface it here with a one-click Reload rather than
|
||||
# silently overwriting — a live reload must never clobber unsaved edits.
|
||||
self.reload_banner = NoticeBanner(self)
|
||||
root.addWidget(self.reload_banner)
|
||||
|
||||
# User-draggable divider between the server list and the editor.
|
||||
split = QSplitter(Qt.Orientation.Horizontal)
|
||||
split.setChildrenCollapsible(False)
|
||||
@@ -2703,6 +2959,8 @@ class MainWindow(QMainWindow):
|
||||
self._health.clear() # health results are per-profile; a fresh load invalidates them
|
||||
self._refresh_sets_combo() # sets are per-config; repopulate from the loaded file
|
||||
self._refresh_tables(select_index=0 if self.servers else -1)
|
||||
self.reload_banner.hide() # loaded fresh: any external-change notice is now stale
|
||||
self._rewatch_paths() # watch this profile's config + the selection's sidecar (#101)
|
||||
self._update_status(saved=False)
|
||||
if repaired:
|
||||
self._mark_dirty()
|
||||
@@ -2993,6 +3251,120 @@ class MainWindow(QMainWindow):
|
||||
self.copy_btn.setEnabled(entry is not None and len(self.profiles) > 1)
|
||||
self.dup_btn.setEnabled(entry is not None)
|
||||
self.del_btn.setEnabled(entry is not None)
|
||||
self._rewatch_paths()
|
||||
|
||||
# --- hot-reload (#101) ------------------------------------------------- #
|
||||
def _watch_targets(self) -> list[Path]:
|
||||
"""Every external path worth watching for the current view.
|
||||
|
||||
BCC's own loaded config (+ its directory), plus the selected server's
|
||||
sidecar path(s) resolved from its ServerSpec. Directories are watched too
|
||||
so a file appearing/disappearing — which a watch on a not-yet-existent
|
||||
file would miss, and which an atomic-rename replace looks like — is still
|
||||
observed. All entries must exist for QFileSystemWatcher to accept them.
|
||||
"""
|
||||
targets: list[Path] = []
|
||||
if self.current_profile is not None:
|
||||
cfg = Path(self.current_profile.path)
|
||||
targets.append(cfg)
|
||||
targets.append(cfg.parent)
|
||||
if self.editor.isEnabled():
|
||||
data = self.editor.dump_data()
|
||||
targets.extend(core.sidecar_watch_paths(data))
|
||||
# De-dup, keep only paths that currently exist (the watcher rejects the
|
||||
# rest; the parent dir covers a not-yet-created file).
|
||||
seen: set[str] = set()
|
||||
out: list[Path] = []
|
||||
for p in targets:
|
||||
key = str(p)
|
||||
if key in seen:
|
||||
continue
|
||||
seen.add(key)
|
||||
if p.exists():
|
||||
out.append(p)
|
||||
return out
|
||||
|
||||
def _rewatch_paths(self):
|
||||
"""Point the watcher at the current targets and snapshot sidecar state.
|
||||
|
||||
Re-applied whenever the selection or the edited command/args change (which
|
||||
server is selected decides which sidecar to watch) and after a reload.
|
||||
"""
|
||||
watcher = self._fs_watcher
|
||||
existing = watcher.files() + watcher.directories()
|
||||
if existing:
|
||||
watcher.removePaths(existing)
|
||||
wanted = [str(p) for p in self._watch_targets()]
|
||||
if wanted:
|
||||
watcher.addPaths(wanted)
|
||||
self._sidecar_fp = self._current_sidecar_fp()
|
||||
|
||||
def _current_sidecar_fp(self) -> tuple | None:
|
||||
if not self.editor.isEnabled():
|
||||
return None
|
||||
return core.sidecar_state_fingerprint(self.editor.dump_data())
|
||||
|
||||
def _on_fs_signal(self, _path=None):
|
||||
"""A watched path changed — coalesce a burst of writes via the debounce."""
|
||||
self._fs_debounce.start()
|
||||
|
||||
def _recheck_external_state(self):
|
||||
"""Debounced re-check: re-arm the watcher, refresh advisories if state
|
||||
actually changed, and surface an external edit to BCC's own config.
|
||||
|
||||
Runs from both the file-watcher and the focus-in fallback. Never mutates
|
||||
the editor form or the loaded config — the only content reload is the
|
||||
user pressing Reload on the banner, so unsaved edits are safe.
|
||||
"""
|
||||
# Re-arm: an atomic-rename replace drops the old inode from the watcher,
|
||||
# so paths must be re-added or the next change goes unseen.
|
||||
self._rewatch_after_event()
|
||||
|
||||
# Sidecar/permission advisories for the selected server (#91/#93). Only
|
||||
# refresh when the observable state changed, so an unrelated write in the
|
||||
# watched directory doesn't thrash the panel.
|
||||
new_fp = self._current_sidecar_fp()
|
||||
if core.sidecar_state_changed(self._sidecar_fp, new_fp):
|
||||
self._sidecar_fp = new_fp
|
||||
self.editor.recheck_advisories()
|
||||
|
||||
# BCC's own loaded config edited outside BCC: surface it (non-destructive).
|
||||
self._check_config_changed_on_disk()
|
||||
|
||||
def _rewatch_after_event(self):
|
||||
"""Re-add any wanted paths the watcher dropped, without disturbing the
|
||||
sidecar fingerprint (which _recheck_external_state compares itself)."""
|
||||
watcher = self._fs_watcher
|
||||
current = set(watcher.files()) | set(watcher.directories())
|
||||
readd = [str(p) for p in self._watch_targets() if str(p) not in current]
|
||||
if readd:
|
||||
watcher.addPaths(readd)
|
||||
|
||||
def _check_config_changed_on_disk(self):
|
||||
if self.current_profile is None:
|
||||
return
|
||||
disk = core.config_fingerprint(self.current_profile.path)
|
||||
if disk is not None and self._loaded_stat is not None and disk != self._loaded_stat:
|
||||
self.reload_banner.show_notice(
|
||||
f"{self.current_profile.path} changed on disk (edited outside BCC).",
|
||||
action_label="Reload from disk",
|
||||
on_action=lambda: self.load_profile(self.current_profile, confirm=True),
|
||||
)
|
||||
else:
|
||||
# Back in sync (e.g. the user reloaded, or the change was reverted).
|
||||
self.reload_banner.hide()
|
||||
|
||||
def changeEvent(self, event):
|
||||
"""Re-check external state when the window regains focus (#101).
|
||||
|
||||
A cheap, always-works fallback: QFileSystemWatcher can miss changes
|
||||
(notably atomic-rename replaces, and files that didn't exist when the
|
||||
watch was set), so a re-check on activation covers the gap.
|
||||
"""
|
||||
if event.type() == QEvent.Type.ActivationChange and self.isActiveWindow():
|
||||
# Go through the debounce so activation + a watcher signal coalesce.
|
||||
self._fs_debounce.start()
|
||||
super().changeEvent(event)
|
||||
|
||||
def _table_item_changed(self, item: QTableWidgetItem):
|
||||
if self._suppress_table or item.column() != 0:
|
||||
@@ -3047,6 +3419,9 @@ class MainWindow(QMainWindow):
|
||||
health_item.setToolTip("Not tested since last edit.")
|
||||
self._suppress_table = False
|
||||
self._refresh_badges()
|
||||
# The edited command/args may change which package (and thus which
|
||||
# sidecar) this server resolves to — re-point the watcher (#101).
|
||||
self._rewatch_paths()
|
||||
self._mark_dirty()
|
||||
|
||||
# --- server actions -------------------------------------------------- #
|
||||
|
||||
+466
-13
@@ -77,7 +77,7 @@ KNOWN_FIELDS = {"command", "args", "env", "url", "type", "headers"}
|
||||
# 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"
|
||||
__version__ = "1.4.0"
|
||||
|
||||
REPO_URL = "https://git.avezzano.io/the_og/better-claude-config"
|
||||
ISSUES_URL = f"{REPO_URL}/issues"
|
||||
@@ -1049,21 +1049,40 @@ 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"
|
||||
# Back up under the source file's own extension so a JSON config and a TOML
|
||||
# sidecar in the same directory keep separate backup pools. For a .json
|
||||
# config this is byte-identical to the historical ".json" naming.
|
||||
suffix = path.suffix or ".json"
|
||||
dest = bdir / f"{path.stem}.{stamp}{suffix}"
|
||||
# Avoid clobbering a same-second backup
|
||||
n = 1
|
||||
while dest.exists():
|
||||
dest = bdir / f"{path.stem}.{stamp}.{n}.json"
|
||||
dest = bdir / f"{path.stem}.{stamp}.{n}{suffix}"
|
||||
n += 1
|
||||
shutil.copy2(path, dest)
|
||||
# Prune oldest beyond MAX_BACKUPS
|
||||
backups = sorted(bdir.glob(f"{path.stem}.*.json"))
|
||||
# Prune oldest beyond MAX_BACKUPS (scoped to this file's extension)
|
||||
backups = sorted(bdir.glob(f"{path.stem}.*{suffix}"))
|
||||
for old in backups[:-MAX_BACKUPS]:
|
||||
with contextlib.suppress(OSError):
|
||||
old.unlink()
|
||||
return dest
|
||||
|
||||
|
||||
def _atomic_write_text(p: Path, payload: str, *, prefix: str) -> None:
|
||||
"""Write `payload` to `p` via a temp file + atomic os.replace on the same
|
||||
filesystem. The shared mechanism behind both write_config (JSON) and
|
||||
write_sidecar (TOML) — the temp file is created 0600 by mkstemp, so a secret
|
||||
is never briefly world-readable mid-write."""
|
||||
fd, tmp = tempfile.mkstemp(dir=str(p.parent), prefix=prefix, suffix=p.suffix or "")
|
||||
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)
|
||||
|
||||
|
||||
def write_config(path: str | os.PathLike, cfg: dict) -> Path | None:
|
||||
"""
|
||||
Atomically write `cfg` to `path` (2-space pretty JSON). Backs up any existing
|
||||
@@ -1075,14 +1094,7 @@ def write_config(path: str | os.PathLike, cfg: dict) -> Path | None:
|
||||
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)
|
||||
_atomic_write_text(p, payload, prefix=".tmp_mcp_")
|
||||
return backup
|
||||
|
||||
|
||||
@@ -3100,6 +3112,447 @@ def sidecar_permission_fix_target(
|
||||
return sidecar_path(spec, platform=platform, environ=environ, home=home)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Hot-reload — live external-change detection (issue #101, epic #94)
|
||||
#
|
||||
# #91/#93 compute the sidecar and permission advisories only at profile load /
|
||||
# server selection, so a config.toml created (or chmod-ed) WHILE BCC runs stays
|
||||
# invisible until a restart. The OS file-watcher, the focus-in re-check and the
|
||||
# debounce are GUI wiring (bcc.py). The testable core is two pure functions:
|
||||
# * which external paths to WATCH for a given server, and
|
||||
# * a comparable SNAPSHOT of the externally-observable state, so the GUI can
|
||||
# cheaply decide "did anything the user can see actually change?" and skip a
|
||||
# redundant (flicker-y) refresh on an unrelated write in the watched dir.
|
||||
# --------------------------------------------------------------------------- #
|
||||
def sidecar_watch_paths(
|
||||
data: dict,
|
||||
*,
|
||||
platform: str | None = None,
|
||||
environ: dict | None = None,
|
||||
home: str | os.PathLike | None = None,
|
||||
) -> list[Path]:
|
||||
"""External filesystem paths whose changes affect a server's advisories.
|
||||
|
||||
Returns the resolved sidecar path, its containing directory (so the file
|
||||
appearing or being deleted — which a file-only watch on a not-yet-existent
|
||||
path would miss — is still observed), and, when distinct, the README/doc
|
||||
path (the #91 wrong-path trap: a file a user placed where the docs say but
|
||||
the server never reads). Empty when the server has no sidecar. Order is
|
||||
stable and de-duplicated so the GUI can hand it straight to a watcher.
|
||||
"""
|
||||
spec = resolve_server_spec(data)
|
||||
if spec is None:
|
||||
return []
|
||||
out: list[Path] = []
|
||||
p = sidecar_path(spec, platform=platform, environ=environ, home=home)
|
||||
if p is not None:
|
||||
out.append(p)
|
||||
out.append(p.parent)
|
||||
doc = sidecar_doc_path(spec, home=home)
|
||||
if doc is not None and doc != p:
|
||||
out.append(doc)
|
||||
out.append(doc.parent)
|
||||
# De-dupe while preserving order (the sidecar file's own dir may equal the
|
||||
# doc dir, or on Linux the doc path coincides with the real one).
|
||||
seen: set[Path] = set()
|
||||
uniq: list[Path] = []
|
||||
for path in out:
|
||||
if path not in seen:
|
||||
seen.add(path)
|
||||
uniq.append(path)
|
||||
return uniq
|
||||
|
||||
|
||||
def sidecar_state_fingerprint(
|
||||
data: dict,
|
||||
*,
|
||||
platform: str | None = None,
|
||||
environ: dict | None = None,
|
||||
home: str | os.PathLike | None = None,
|
||||
exists=None,
|
||||
stat_mode=None,
|
||||
) -> tuple | None:
|
||||
"""A comparable snapshot of a server's externally-observable sidecar state.
|
||||
|
||||
Folds the #91 sidecar status (existence, precedence, wrong-path) and the #93
|
||||
permission status (mode + ok) into one hashable tuple. The GUI takes a
|
||||
fingerprint before and after a filesystem event and refreshes the advisories
|
||||
only when it changed — so an unrelated write in the watched directory doesn't
|
||||
thrash the UI. Returns None when the server has no sidecar (nothing to
|
||||
watch). All inputs are injectable so tests drive a virtual filesystem.
|
||||
"""
|
||||
status = sidecar_status(data, platform=platform, environ=environ, home=home, exists=exists)
|
||||
if status is None:
|
||||
return None
|
||||
perm = permission_status(status["path"], platform=platform, stat_mode=stat_mode)
|
||||
return (
|
||||
status["exists"],
|
||||
status["doc_exists"],
|
||||
status["args_inert"],
|
||||
status["wrong_path"],
|
||||
None if perm is None else perm["ok"],
|
||||
None if perm is None else perm["mode"],
|
||||
)
|
||||
|
||||
|
||||
def sidecar_state_changed(before: tuple | None, after: tuple | None) -> bool:
|
||||
"""True when two ``sidecar_state_fingerprint`` snapshots differ.
|
||||
|
||||
A thin, explicitly-named equality so the GUI's watcher/focus-in handlers read
|
||||
intentionally and the "should I refresh?" decision stays covered by tests.
|
||||
"""
|
||||
return before != after
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Sidecar TOML editing (issue #102, epic #94)
|
||||
#
|
||||
# #91 shipped read-only sidecar DETECTION; #102 lets a layperson EDIT that file
|
||||
# from inside BCC. The write target is a new file (ssh-mcp's config.toml) with a
|
||||
# different shape, so it gets its own writer (write_sidecar) that REUSES the
|
||||
# atomic-rename + rotating-backup machinery — it never routes through
|
||||
# apply_servers (the cardinal rule: that only ever writes mcpServers /
|
||||
# _disabledMcpServers).
|
||||
#
|
||||
# Dependency decision (see the PR): NO new runtime dep. CI's test job installs
|
||||
# only `pytest cryptography` (not requirements.txt) and the catalog-signature
|
||||
# job imports bcc_core with only cryptography present, and the workflow is
|
||||
# off-limits — so a top-level TOML import is impossible and any pip TOML dep
|
||||
# would leave this core untested/red on CI. Instead a stdlib-only, deliberately
|
||||
# minimal reader + a SURGICAL line-editing writer: the writer only ever rewrites
|
||||
# the specific key it is asked to change, so comments, formatting and every key
|
||||
# it does not understand round-trip untouched — strictly safer than a
|
||||
# parse->dict->reserialize pass (which tomli-w would also not comment-preserve).
|
||||
# The GUI keeps a raw-text fallback for anything the form doesn't model.
|
||||
# --------------------------------------------------------------------------- #
|
||||
_TOML_UNPARSED = object() # sentinel: a value this lenient reader won't decode
|
||||
|
||||
# A top-level `[table]` or `[[array-of-tables]]` header line.
|
||||
_TOML_HEADER_RE = re.compile(r"^\s*\[\[?\s*([^\[\]]+?)\s*\]\]?\s*(?:#.*)?$")
|
||||
# A `key = value` line: bare, "quoted" or 'quoted' key, capturing indent, the
|
||||
# key token verbatim, the '=' run (with its spacing) and the remaining RHS.
|
||||
_TOML_KV_RE = re.compile(r"^(\s*)((?:\"[^\"]*\")|(?:'[^']*')|[A-Za-z0-9_.\-]+)(\s*=\s*)(.*)$")
|
||||
|
||||
|
||||
def _toml_dump_scalar(value) -> str:
|
||||
"""Serialize a Python scalar to a TOML value token.
|
||||
|
||||
Supports the scalar types the editor writes: bool, int, float, str. A string
|
||||
is emitted as a TOML basic string with the standard escapes. Raises TypeError
|
||||
for anything else — the form only ever hands us managed scalars, and the
|
||||
surgical writer never re-serializes values it didn't originate.
|
||||
"""
|
||||
if isinstance(value, bool): # bool before int — bool is an int subclass
|
||||
return "true" if value else "false"
|
||||
if isinstance(value, int):
|
||||
return str(value)
|
||||
if isinstance(value, float):
|
||||
return repr(value)
|
||||
if isinstance(value, str):
|
||||
esc = (
|
||||
value.replace("\\", "\\\\")
|
||||
.replace('"', '\\"')
|
||||
.replace("\n", "\\n")
|
||||
.replace("\r", "\\r")
|
||||
.replace("\t", "\\t")
|
||||
)
|
||||
return f'"{esc}"'
|
||||
raise TypeError(f"unsupported TOML scalar type: {type(value).__name__}")
|
||||
|
||||
|
||||
def _toml_unescape_basic(s: str) -> str:
|
||||
"""Decode the escape sequences a TOML basic string can carry."""
|
||||
out: list[str] = []
|
||||
i = 0
|
||||
table = {"n": "\n", "t": "\t", "r": "\r", "\\": "\\", '"': '"', "b": "\b", "f": "\f", "0": "\0"}
|
||||
while i < len(s):
|
||||
ch = s[i]
|
||||
if ch == "\\" and i + 1 < len(s):
|
||||
out.append(table.get(s[i + 1], s[i + 1]))
|
||||
i += 2
|
||||
else:
|
||||
out.append(ch)
|
||||
i += 1
|
||||
return "".join(out)
|
||||
|
||||
|
||||
def _toml_parse_scalar(token: str):
|
||||
"""Parse a TOML scalar token into a Python value, or ``_TOML_UNPARSED``.
|
||||
|
||||
Handles basic/literal strings, integers, floats and booleans — the values
|
||||
the editor's managed fields use. Arrays, inline tables, dates and multiline
|
||||
strings return the sentinel so callers leave them untouched (the surgical
|
||||
writer preserves the original text regardless).
|
||||
"""
|
||||
t = token.strip()
|
||||
if not t:
|
||||
return _TOML_UNPARSED
|
||||
if t in ("true", "false"):
|
||||
return t == "true"
|
||||
if len(t) >= 2 and t[0] == t[-1] and t[0] in "\"'":
|
||||
inner = t[1:-1]
|
||||
return inner if t[0] == "'" else _toml_unescape_basic(inner)
|
||||
cleaned = t.replace("_", "")
|
||||
if re.fullmatch(r"[+-]?\d+", cleaned):
|
||||
try:
|
||||
return int(cleaned)
|
||||
except ValueError:
|
||||
return _TOML_UNPARSED
|
||||
if ("." in cleaned or "e" in cleaned.lower()) and re.fullmatch(
|
||||
r"[+-]?(?:\d+\.\d*|\.\d+|\d+)(?:[eE][+-]?\d+)?", cleaned
|
||||
):
|
||||
try:
|
||||
return float(cleaned)
|
||||
except ValueError:
|
||||
return _TOML_UNPARSED
|
||||
return _TOML_UNPARSED
|
||||
|
||||
|
||||
def _split_value_comment(rhs: str) -> tuple[str, str]:
|
||||
"""Split a RHS into (value_token, trailing) at the first unquoted ``#``.
|
||||
|
||||
``trailing`` includes the run of whitespace immediately before the ``#`` so a
|
||||
rewrite can splice a new value in front of it and preserve the exact spacing
|
||||
and the comment. Quotes are respected so a ``#`` inside a string is not
|
||||
mistaken for a comment.
|
||||
"""
|
||||
quote = None
|
||||
for i, ch in enumerate(rhs):
|
||||
if quote:
|
||||
if ch == quote:
|
||||
quote = None
|
||||
elif ch in "\"'":
|
||||
quote = ch
|
||||
elif ch == "#":
|
||||
j = i
|
||||
while j > 0 and rhs[j - 1] in " \t":
|
||||
j -= 1
|
||||
return rhs[:j].rstrip(), rhs[j:]
|
||||
return rhs.rstrip(), ""
|
||||
|
||||
|
||||
def _toml_section_span(lines: list[str], section: str | None):
|
||||
"""(header_idx, body_start, body_end) for a section's line range.
|
||||
|
||||
``section=None`` targets the top-level block (before the first header):
|
||||
header_idx is None, body is [0, first-header). A named section returns the
|
||||
header line index and its body [after-header, next-header). When a named
|
||||
section is absent, returns (None, None, None).
|
||||
"""
|
||||
if section is None:
|
||||
end = len(lines)
|
||||
for i, raw in enumerate(lines):
|
||||
if _TOML_HEADER_RE.match(raw):
|
||||
end = i
|
||||
break
|
||||
return (None, 0, end)
|
||||
for i, raw in enumerate(lines):
|
||||
h = _TOML_HEADER_RE.match(raw)
|
||||
if h and h.group(1).strip().strip("'\"") == section:
|
||||
body_end = len(lines)
|
||||
for j in range(i + 1, len(lines)):
|
||||
if _TOML_HEADER_RE.match(lines[j]):
|
||||
body_end = j
|
||||
break
|
||||
return (i, i + 1, body_end)
|
||||
return (None, None, None)
|
||||
|
||||
|
||||
def _find_key_line(lines: list[str], start: int, end: int, key: str) -> int | None:
|
||||
for i in range(start, end):
|
||||
m = _TOML_KV_RE.match(lines[i].rstrip("\r\n"))
|
||||
if m and m.group(2).strip().strip("'\"") == key:
|
||||
return i
|
||||
return None
|
||||
|
||||
|
||||
def read_toml_section(text: str, section: str | None = None) -> dict:
|
||||
"""Scalar ``key -> value`` map for one section (top-level when ``section`` is None).
|
||||
|
||||
Lenient and deliberately minimal: only simple scalars (string/int/float/bool)
|
||||
are returned. Keys whose values this reader can't confidently decode (arrays,
|
||||
inline tables, dates, multiline strings) are omitted from the result — the
|
||||
surgical writer preserves them regardless, and the GUI's raw-text fallback
|
||||
covers editing them by hand.
|
||||
"""
|
||||
out: dict = {}
|
||||
lines = (text or "").splitlines(keepends=True)
|
||||
_, body_start, body_end = _toml_section_span(lines, section)
|
||||
if body_start is None:
|
||||
return out
|
||||
for i in range(body_start, body_end):
|
||||
raw = lines[i].rstrip("\r\n")
|
||||
stripped = raw.strip()
|
||||
if not stripped or stripped.startswith("#"):
|
||||
continue
|
||||
m = _TOML_KV_RE.match(raw)
|
||||
if not m:
|
||||
continue
|
||||
key = m.group(2).strip().strip("'\"")
|
||||
value_token, _comment = _split_value_comment(m.group(4))
|
||||
value = _toml_parse_scalar(value_token)
|
||||
if value is not _TOML_UNPARSED:
|
||||
out[key] = value
|
||||
return out
|
||||
|
||||
|
||||
def toml_sections(text: str) -> list[str]:
|
||||
"""Ordered, de-duplicated list of the top-level section names in `text`.
|
||||
|
||||
A ``[server]`` / ``[[hosts]]`` / ``[prod.auth]`` header contributes its first
|
||||
dotted component (so ``[prod]`` and ``[prod.auth]`` are one profile group),
|
||||
mirroring ``count_toml_profiles``. Lets the editor offer the sections a user
|
||||
can target. Top-level (pre-header) keys are represented by ``""`` when present.
|
||||
"""
|
||||
names: list[str] = []
|
||||
seen: set[str] = set()
|
||||
saw_top = False
|
||||
for raw in (text or "").splitlines():
|
||||
line = raw.strip()
|
||||
if not line or line.startswith("#"):
|
||||
continue
|
||||
h = _TOML_HEADER_RE.match(raw)
|
||||
if h:
|
||||
top = h.group(1).split(".", 1)[0].strip().strip("'\"")
|
||||
if top and top not in seen:
|
||||
seen.add(top)
|
||||
names.append(top)
|
||||
elif not names and _TOML_KV_RE.match(raw):
|
||||
saw_top = True
|
||||
if saw_top:
|
||||
names.insert(0, "")
|
||||
return names
|
||||
|
||||
|
||||
def _rewrite_key_line(line: str, value, nl: str) -> str:
|
||||
"""Replace only the value on an existing ``key = value`` line.
|
||||
|
||||
Preserves indentation, the key's original spelling, the ``=`` spacing, any
|
||||
trailing comment and the line's own ending — everything but the value.
|
||||
"""
|
||||
stripped = line.rstrip("\r\n")
|
||||
ending = line[len(stripped) :] or nl
|
||||
m = _TOML_KV_RE.match(stripped)
|
||||
indent, keytok, eq, rest = m.group(1), m.group(2), m.group(3), m.group(4)
|
||||
_oldval, comment = _split_value_comment(rest)
|
||||
return f"{indent}{keytok}{eq}{_toml_dump_scalar(value)}{comment}{ending}"
|
||||
|
||||
|
||||
def set_toml_value(text: str, key: str, value, *, section: str | None = None) -> str:
|
||||
"""Return `text` with ``[section] key`` set to `value`, surgically.
|
||||
|
||||
* An existing key line has only its value replaced (comment/formatting kept).
|
||||
* A missing key is appended at the end of the section body.
|
||||
* A missing section is created (with the key) at the end of the file.
|
||||
* ``value=None`` deletes the key line (no-op if absent).
|
||||
|
||||
Everything else in the document — comments, blank lines, unknown keys and
|
||||
tables, ordering — round-trips untouched. This is the writer the cardinal
|
||||
rule calls for: it reuses nothing from apply_servers and only ever changes
|
||||
the one key it is asked to.
|
||||
"""
|
||||
text = text or ""
|
||||
nl = "\r\n" if "\r\n" in text else "\n"
|
||||
lines = text.splitlines(keepends=True)
|
||||
_, body_start, body_end = _toml_section_span(lines, section)
|
||||
|
||||
if body_start is None: # named section absent
|
||||
if value is None:
|
||||
return text
|
||||
chunk = ""
|
||||
if lines and not lines[-1].endswith(("\n", "\r")):
|
||||
chunk += nl # terminate a final line that lacked a newline
|
||||
if lines and lines[-1].strip():
|
||||
chunk += nl # readability blank line before the new section
|
||||
chunk += f"[{section}]{nl}{key} = {_toml_dump_scalar(value)}{nl}"
|
||||
return text + chunk
|
||||
|
||||
ki = _find_key_line(lines, body_start, body_end, key)
|
||||
if ki is not None:
|
||||
if value is None:
|
||||
del lines[ki]
|
||||
else:
|
||||
lines[ki] = _rewrite_key_line(lines[ki], value, nl)
|
||||
return "".join(lines)
|
||||
|
||||
if value is None:
|
||||
return text # nothing to delete
|
||||
|
||||
# Insert a new key at the end of the section body, before any trailing blank
|
||||
# lines (which usually separate it from the next section).
|
||||
insert_at = body_end
|
||||
while insert_at > body_start and lines[insert_at - 1].strip() == "":
|
||||
insert_at -= 1
|
||||
if insert_at > 0 and not lines[insert_at - 1].endswith(("\n", "\r")):
|
||||
lines[insert_at - 1] = lines[insert_at - 1] + nl
|
||||
lines.insert(insert_at, f"{key} = {_toml_dump_scalar(value)}{nl}")
|
||||
return "".join(lines)
|
||||
|
||||
|
||||
def update_toml(text: str, updates: dict, *, section: str | None = None) -> str:
|
||||
"""Apply several key updates to one section, surgically (see set_toml_value).
|
||||
|
||||
A ``None`` value deletes that key. Applied in order; each edit preserves the
|
||||
rest of the document, so the result round-trips every untouched line.
|
||||
"""
|
||||
for key, value in updates.items():
|
||||
text = set_toml_value(text, key, value, section=section)
|
||||
return text
|
||||
|
||||
|
||||
def validate_sidecar_values(values: dict, schema: dict) -> list[str]:
|
||||
"""Validate proposed managed values against a ServerSpec schema.
|
||||
|
||||
Only keys the schema *manages* are policed (the enum/range fields the editor
|
||||
renders as pick-lists / a numeric field). Any other key is passed through
|
||||
untouched — BCC preserves unknown keys rather than rejecting them, so an
|
||||
out-of-schema key a user already has never blocks a save. Returns a list of
|
||||
plain-language problems; empty == all good.
|
||||
"""
|
||||
problems: list[str] = []
|
||||
for key, val in values.items():
|
||||
rule = schema.get(key)
|
||||
if rule is None:
|
||||
continue # unmanaged key — preserved, not policed
|
||||
if isinstance(rule, list): # enum
|
||||
if val not in rule:
|
||||
allowed = ", ".join(str(x) for x in rule)
|
||||
problems.append(f"{key}: {val!r} is not allowed — choose one of: {allowed}")
|
||||
elif isinstance(rule, dict) and "min" in rule and "max" in rule: # numeric range
|
||||
lo, hi = rule["min"], rule["max"]
|
||||
if isinstance(val, bool) or not isinstance(val, int) or not (lo <= val <= hi):
|
||||
problems.append(f"{key}: must be a whole number between {lo} and {hi}")
|
||||
return problems
|
||||
|
||||
|
||||
def write_sidecar(
|
||||
path: str | os.PathLike,
|
||||
text: str,
|
||||
*,
|
||||
platform: str | None = None,
|
||||
chmod=None,
|
||||
) -> Path | None:
|
||||
"""Atomically write TOML `text` to a sidecar `path`, safely.
|
||||
|
||||
Reuses the same temp-write + atomic ``os.replace`` + rotating timestamped
|
||||
backup as ``write_config``, then tightens the file to 0600 and its directory
|
||||
to 0700 (POSIX; a clean no-op on Windows) via #93's ``fix_permissions``. It
|
||||
does **not** go through ``apply_servers`` — the cardinal rule reserves that
|
||||
for the JSON server keys; the sidecar is a different file with a different
|
||||
shape. Returns the backup path, or None if there was no prior file to back
|
||||
up. ``chmod`` is injectable for tests.
|
||||
"""
|
||||
p = Path(path)
|
||||
p.parent.mkdir(parents=True, exist_ok=True)
|
||||
backup = _make_backup(p) if p.is_file() else None
|
||||
payload = text if text.endswith("\n") else text + "\n"
|
||||
_atomic_write_text(p, payload, prefix=".tmp_sidecar_")
|
||||
# Tighten permissions after the file is in place (the temp file was already
|
||||
# 0600 from mkstemp, so the secret was never briefly world-readable).
|
||||
fix_permissions(p, platform=platform, chmod=chmod)
|
||||
return backup
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Validation
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
[project]
|
||||
name = "better-claude-config"
|
||||
version = "1.3.0"
|
||||
version = "1.4.0"
|
||||
description = "Cross-platform GUI for editing the mcpServers block of Claude Desktop and Claude Code configs"
|
||||
readme = "README.md"
|
||||
license = { file = "LICENSE" }
|
||||
|
||||
@@ -3645,6 +3645,277 @@ def test_sidecar_permission_fix_target():
|
||||
assert c.sidecar_permission_fix_target({"command": "npx", "args": ["other"]}) is None
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Hot-reload — live external-change detection (issue #101)
|
||||
# --------------------------------------------------------------------------- #
|
||||
def test_sidecar_watch_paths_covers_file_dir_and_doc():
|
||||
data = dict(_SSH)
|
||||
paths = c.sidecar_watch_paths(data, platform="darwin", environ={}, home=_HOME)
|
||||
posix = [p.as_posix() for p in paths]
|
||||
real = c.sidecar_path(c.SERVER_SPECS["ssh-mcp"], platform="darwin", environ={}, home=_HOME)
|
||||
# The resolved sidecar file AND its directory (so create/delete registers).
|
||||
assert real.as_posix() in posix
|
||||
assert real.parent.as_posix() in posix
|
||||
# The README/doc path is distinct on macOS -> also watched.
|
||||
doc = c.sidecar_doc_path(c.SERVER_SPECS["ssh-mcp"], home=_HOME)
|
||||
assert doc.as_posix() in posix
|
||||
# No duplicates.
|
||||
assert len(posix) == len(set(posix))
|
||||
|
||||
|
||||
def test_sidecar_watch_paths_dedupes_on_linux():
|
||||
# On Linux the doc path coincides with the real path, so the file + dir are
|
||||
# each listed once, not twice.
|
||||
data = dict(_SSH)
|
||||
paths = c.sidecar_watch_paths(
|
||||
data, platform="linux", environ={"XDG_CONFIG_HOME": "/cfg"}, home=_HOME
|
||||
)
|
||||
posix = [p.as_posix() for p in paths]
|
||||
assert posix == list(dict.fromkeys(posix)) # order-preserving de-dup is a no-op
|
||||
assert "/cfg/ssh-mcp/config.toml" in posix
|
||||
assert "/cfg/ssh-mcp" in posix
|
||||
|
||||
|
||||
def test_sidecar_watch_paths_empty_for_non_sidecar():
|
||||
assert c.sidecar_watch_paths({"command": "npx", "args": ["other"]}) == []
|
||||
|
||||
|
||||
def test_sidecar_state_fingerprint_reflects_appearance_and_perms():
|
||||
data = dict(_SSH)
|
||||
real = c.sidecar_path(c.SERVER_SPECS["ssh-mcp"], platform="darwin", environ={}, home=_HOME)
|
||||
|
||||
# File absent -> a fingerprint that encodes "not there, no perms".
|
||||
absent = c.sidecar_state_fingerprint(
|
||||
data, platform="darwin", environ={}, home=_HOME, exists=lambda p: False
|
||||
)
|
||||
# File present but world-readable.
|
||||
bad = c.sidecar_state_fingerprint(
|
||||
data,
|
||||
platform="darwin",
|
||||
environ={},
|
||||
home=_HOME,
|
||||
exists=lambda p: Path(p) == real,
|
||||
stat_mode=lambda p: 0o644 if Path(p) == real else 0o700,
|
||||
)
|
||||
# File present and tight.
|
||||
good = c.sidecar_state_fingerprint(
|
||||
data,
|
||||
platform="darwin",
|
||||
environ={},
|
||||
home=_HOME,
|
||||
exists=lambda p: Path(p) == real,
|
||||
stat_mode=lambda p: 0o600 if Path(p) == real else 0o700,
|
||||
)
|
||||
assert absent is not None
|
||||
# Each observable transition changes the fingerprint.
|
||||
assert c.sidecar_state_changed(absent, bad)
|
||||
assert c.sidecar_state_changed(bad, good)
|
||||
assert c.sidecar_state_changed(absent, good)
|
||||
# Stable when nothing changed.
|
||||
assert not c.sidecar_state_changed(good, good)
|
||||
|
||||
|
||||
def test_sidecar_state_fingerprint_none_for_non_sidecar():
|
||||
assert c.sidecar_state_fingerprint({"command": "npx", "args": ["other"]}) is None
|
||||
|
||||
|
||||
def test_sidecar_state_fingerprint_tracks_args_inert_flip():
|
||||
# args_inert only becomes true once the sidecar file exists AND managed args
|
||||
# are present; the fingerprint must capture that precedence flip.
|
||||
data = dict(_SSH) # carries --host/--user (managed args)
|
||||
real = c.sidecar_path(
|
||||
c.SERVER_SPECS["ssh-mcp"], platform="linux", environ={"XDG_CONFIG_HOME": "/cfg"}, home=_HOME
|
||||
)
|
||||
before = c.sidecar_state_fingerprint(
|
||||
data,
|
||||
platform="linux",
|
||||
environ={"XDG_CONFIG_HOME": "/cfg"},
|
||||
home=_HOME,
|
||||
exists=lambda p: False,
|
||||
stat_mode=lambda p: None,
|
||||
)
|
||||
after = c.sidecar_state_fingerprint(
|
||||
data,
|
||||
platform="linux",
|
||||
environ={"XDG_CONFIG_HOME": "/cfg"},
|
||||
home=_HOME,
|
||||
exists=lambda p: Path(p) == real,
|
||||
stat_mode=lambda p: 0o600,
|
||||
)
|
||||
assert c.sidecar_state_changed(before, after)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Sidecar TOML editing (issue #102)
|
||||
# --------------------------------------------------------------------------- #
|
||||
def test_read_toml_section_top_level_and_named():
|
||||
text = (
|
||||
"# a comment\n"
|
||||
'title = "root"\n'
|
||||
"port = 22\n"
|
||||
"enabled = true\n"
|
||||
"\n"
|
||||
"[server]\n"
|
||||
'host = "h"\n'
|
||||
"auth = 'key'\n" # literal string
|
||||
"weight = 1.5\n"
|
||||
)
|
||||
top = c.read_toml_section(text)
|
||||
assert top == {"title": "root", "port": 22, "enabled": True}
|
||||
srv = c.read_toml_section(text, "server")
|
||||
assert srv == {"host": "h", "auth": "key", "weight": 1.5}
|
||||
# An absent section reads as empty.
|
||||
assert c.read_toml_section(text, "nope") == {}
|
||||
|
||||
|
||||
def test_read_toml_section_skips_unparseable_values():
|
||||
text = '[server]\nhosts = [1, 2, 3]\ninline = {a = 1}\nname = "ok"\n'
|
||||
# Arrays / inline tables are omitted (surgical writer preserves them); the
|
||||
# simple scalar is read.
|
||||
assert c.read_toml_section(text, "server") == {"name": "ok"}
|
||||
|
||||
|
||||
def test_toml_sections_lists_groups_in_order():
|
||||
text = "top = 1\n[server]\nx=1\n[prod]\na=1\n[prod.auth]\nk=1\n[[hosts]]\nn=1\n"
|
||||
# top-level present -> "" first; [prod] and [prod.auth] collapse to one group.
|
||||
assert c.toml_sections(text) == ["", "server", "prod", "hosts"]
|
||||
assert c.toml_sections("[only]\nx=1\n") == ["only"]
|
||||
|
||||
|
||||
def test_set_toml_value_replaces_and_preserves_comment():
|
||||
text = '[server]\nport = 22 # ssh\nhost = "h"\n'
|
||||
out = c.set_toml_value(text, "port", 2222, section="server")
|
||||
assert "port = 2222 # ssh" in out
|
||||
# Nothing else changed.
|
||||
assert 'host = "h"' in out
|
||||
assert out.startswith("[server]\n")
|
||||
|
||||
|
||||
def test_set_toml_value_appends_missing_key_within_section():
|
||||
text = '[server]\nhost = "h"\n\n[other]\nx = 1\n'
|
||||
out = c.set_toml_value(text, "auth", "key", section="server")
|
||||
lines = out.splitlines()
|
||||
# The new key lands inside [server], before the blank line and [other].
|
||||
assert lines.index('auth = "key"') < lines.index("[other]")
|
||||
assert lines.index('auth = "key"') > lines.index("[server]")
|
||||
assert "[other]" in out and "x = 1" in out
|
||||
|
||||
|
||||
def test_set_toml_value_creates_missing_section():
|
||||
text = 'title = "root"\n'
|
||||
out = c.set_toml_value(text, "role", "admin", section="server")
|
||||
assert 'title = "root"' in out
|
||||
assert "[server]" in out
|
||||
assert 'role = "admin"' in out
|
||||
# And it round-trips back through the reader.
|
||||
assert c.read_toml_section(out, "server") == {"role": "admin"}
|
||||
|
||||
|
||||
def test_set_toml_value_top_level_key():
|
||||
text = "a = 1\n[server]\nb = 2\n"
|
||||
out = c.set_toml_value(text, "a", 5) # section=None -> top-level
|
||||
assert out.startswith("a = 5\n")
|
||||
# The [server] b is untouched.
|
||||
assert c.read_toml_section(out, "server") == {"b": 2}
|
||||
|
||||
|
||||
def test_set_toml_value_delete_key():
|
||||
text = '[server]\nhost = "h"\nport = 22\n'
|
||||
out = c.set_toml_value(text, "port", None, section="server")
|
||||
assert "port" not in out
|
||||
assert 'host = "h"' in out
|
||||
# Deleting an absent key is a no-op.
|
||||
assert c.set_toml_value(text, "ghost", None, section="server") == text
|
||||
|
||||
|
||||
def test_set_toml_value_preserves_crlf():
|
||||
text = "[server]\r\nport = 22\r\n"
|
||||
out = c.set_toml_value(text, "port", 23, section="server")
|
||||
assert "port = 23\r\n" in out
|
||||
assert "\r\n" in out
|
||||
|
||||
|
||||
def test_set_toml_value_escapes_strings():
|
||||
out = c.set_toml_value("", "path", 'C:\\a\\"b"', section="win")
|
||||
# Backslashes and quotes are escaped; it reads back to the exact original.
|
||||
assert c.read_toml_section(out, "win") == {"path": 'C:\\a\\"b"'}
|
||||
|
||||
|
||||
def test_update_toml_batch_and_roundtrip():
|
||||
text = '[server]\nhost = "h" # keep me\n'
|
||||
out = c.update_toml(text, {"host": "newhost", "port": 22, "auth": "key"}, section="server")
|
||||
assert c.read_toml_section(out, "server") == {"host": "newhost", "port": 22, "auth": "key"}
|
||||
# The comment on the pre-existing host line survives the value change.
|
||||
assert "# keep me" in out
|
||||
|
||||
|
||||
def test_validate_sidecar_values_enums_and_port():
|
||||
schema = c.SERVER_SPECS["ssh-mcp"].schema
|
||||
# All valid -> no problems.
|
||||
assert (
|
||||
c.validate_sidecar_values(
|
||||
{"auth": "key", "approvalMode": "ask-all", "role": "admin", "port": 22}, schema
|
||||
)
|
||||
== []
|
||||
)
|
||||
# Bad enum value.
|
||||
probs = c.validate_sidecar_values({"auth": "sshkey"}, schema)
|
||||
assert probs and "auth" in probs[0] and "sshkey" in probs[0]
|
||||
# Out-of-range port, and a bool is not a valid int port.
|
||||
assert c.validate_sidecar_values({"port": 70000}, schema)
|
||||
assert c.validate_sidecar_values({"port": True}, schema)
|
||||
assert c.validate_sidecar_values({"port": 1}, schema) == []
|
||||
|
||||
|
||||
def test_validate_sidecar_values_passes_through_unknown_keys():
|
||||
schema = c.SERVER_SPECS["ssh-mcp"].schema
|
||||
# An unmanaged key is preserved, never a save-blocker.
|
||||
assert c.validate_sidecar_values({"customThing": "whatever"}, schema) == []
|
||||
|
||||
|
||||
def test_write_sidecar_atomic_backup_and_chmod(tmp_path):
|
||||
p = tmp_path / "ssh-mcp" / "config.toml"
|
||||
p.parent.mkdir()
|
||||
p.write_text("[server]\nport = 22\n", encoding="utf-8")
|
||||
|
||||
chmods: list[tuple[str, int]] = []
|
||||
new_text = c.set_toml_value(p.read_text(encoding="utf-8"), "port", 2222, section="server")
|
||||
backup = c.write_sidecar(
|
||||
p,
|
||||
new_text,
|
||||
platform="linux",
|
||||
chmod=lambda path, mode: chmods.append((Path(path).as_posix(), mode)),
|
||||
)
|
||||
|
||||
# File updated atomically with the new content.
|
||||
assert c.read_toml_section(p.read_text(encoding="utf-8"), "server") == {"port": 2222}
|
||||
# A timestamped backup of the prior content was made, under .toml.
|
||||
assert backup is not None and backup.suffix == ".toml"
|
||||
assert c.read_toml_section(backup.read_text(encoding="utf-8"), "server") == {"port": 22}
|
||||
# Permissions tightened: file 0600, dir 0700 (via #93's fix_permissions).
|
||||
assert (p.as_posix(), 0o600) in chmods
|
||||
assert (p.parent.as_posix(), 0o700) in chmods
|
||||
|
||||
|
||||
def test_write_sidecar_new_file_no_backup(tmp_path):
|
||||
p = tmp_path / "config.toml"
|
||||
backup = c.write_sidecar(
|
||||
p, '[server]\nrole = "viewer"\n', platform="linux", chmod=lambda *_: None
|
||||
)
|
||||
assert backup is None # nothing pre-existing to back up
|
||||
assert p.read_text(encoding="utf-8").endswith("\n")
|
||||
assert c.read_toml_section(p.read_text(encoding="utf-8"), "server") == {"role": "viewer"}
|
||||
|
||||
|
||||
def test_write_sidecar_does_not_touch_apply_servers(tmp_path):
|
||||
# Guard the cardinal rule: write_sidecar writes ONLY the text it is given —
|
||||
# no mcpServers / _disabledMcpServers keys are introduced.
|
||||
p = tmp_path / "config.toml"
|
||||
c.write_sidecar(p, '[server]\nhost = "h"\n', platform="linux", chmod=lambda *_: None)
|
||||
body = p.read_text(encoding="utf-8")
|
||||
assert "mcpServers" not in body and "_disabledMcpServers" not in body
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Move to environment variable (issue #83)
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
Reference in New Issue
Block a user