Merge pull request 'feat(#102): in-app sidecar TOML editor (stacked on #101)' (#104) from feat/102 into main
CI / Lint (ruff) (push) Successful in 7s
CI / Tests (py3.12 / windows-latest) (push) Successful in 21s
CI / Tests (py3.10 / ubuntu-latest) (push) Successful in 11s
CI / Tests (py3.12 / ubuntu-latest) (push) Successful in 11s
CI / Tests (py3.13 / ubuntu-latest) (push) Successful in 11s
CI / Catalog signature (push) Successful in 6s

This commit was merged in pull request #104.
This commit is contained in:
2026-08-13 00:55:21 -04:00
3 changed files with 763 additions and 15 deletions
+219 -3
View File
@@ -49,8 +49,10 @@ from PySide6.QtWidgets import (
QDialog,
QDialogButtonBox,
QFileDialog,
QFormLayout,
QFrame,
QGridLayout,
QGroupBox,
QHBoxLayout,
QHeaderView,
QInputDialog,
@@ -63,6 +65,7 @@ from PySide6.QtWidgets import (
QMessageBox,
QPlainTextEdit,
QPushButton,
QSpinBox,
QSplitter,
QStackedWidget,
QStyledItemDelegate,
@@ -933,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.
@@ -1160,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
@@ -1202,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:
@@ -1248,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():
@@ -2162,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.
+373 -12
View File
@@ -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
@@ -3192,6 +3204,355 @@ def sidecar_state_changed(before: tuple | None, after: tuple | None) -> bool:
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
# --------------------------------------------------------------------------- #
+171
View File
@@ -3745,6 +3745,177 @@ def test_sidecar_state_fingerprint_tracks_args_inert_flip():
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)
# --------------------------------------------------------------------------- #