diff --git a/bcc.py b/bcc.py index 5b87889..18ece9f 100644 --- a/bcc.py +++ b/bcc.py @@ -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 {path}
" + "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 0600." + ) + 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. diff --git a/bcc_core.py b/bcc_core.py index b65a9a6..2875d8e 100644 --- a/bcc_core.py +++ b/bcc_core.py @@ -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 # --------------------------------------------------------------------------- # diff --git a/tests/test_core.py b/tests/test_core.py index 68ee9ba..393db57 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -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) # --------------------------------------------------------------------------- #