10 Commits
Author SHA1 Message Date
the_og cdda60da1b Merge pull request 'Land ServerSpec cut: spine + sidecar + version + perms (#90–#93)' (#100) from feat/serverspec-cut into main
CI / Lint (ruff) (push) Successful in 9s
CI / Tests (py3.10 / ubuntu-latest) (push) Successful in 12s
CI / Tests (py3.12 / ubuntu-latest) (push) Successful in 11s
CI / Tests (py3.12 / windows-latest) (push) Successful in 35s
CI / Tests (py3.13 / ubuntu-latest) (push) Successful in 11s
CI / Catalog signature (push) Successful in 7s
2026-08-12 23:52:59 -04:00
t 0920846c2c style: ruff format the union-merge seam
CI / Lint (ruff) (pull_request) Successful in 7s
CI / Tests (py3.10 / ubuntu-latest) (pull_request) Successful in 12s
CI / Tests (py3.12 / ubuntu-latest) (pull_request) Successful in 11s
CI / Tests (py3.12 / windows-latest) (pull_request) Successful in 35s
CI / Tests (py3.13 / ubuntu-latest) (pull_request) Successful in 11s
CI / Catalog signature (pull_request) Successful in 7s
2026-08-12 14:20:53 +00:00
t ce82f7b5c3 Merge remote-tracking branch 'origin/feat/93' into integ 2026-08-12 14:18:07 +00:00
t d2c126a60c Merge remote-tracking branch 'origin/feat/92' into integ
# Conflicts:
#	bcc_core.py
#	tests/test_core.py
2026-08-12 14:18:07 +00:00
Cowork SupervisorandClaude Opus 4.8 603d24566d test(#93): key injected mode/chmod maps on as_posix() for the Windows runner
CI / Lint (ruff) (pull_request) Successful in 9s
CI / Tests (py3.12 / windows-latest) (pull_request) Successful in 23s
CI / Tests (py3.10 / ubuntu-latest) (pull_request) Successful in 12s
CI / Tests (py3.12 / ubuntu-latest) (pull_request) Successful in 16s
CI / Tests (py3.13 / ubuntu-latest) (pull_request) Successful in 13s
CI / Catalog signature (pull_request) Successful in 8s
permission_status/fix_permissions wrap paths in the running OS's Path class, so
on the windows-latest CI job str(p) uses backslashes and the fixture dict lookups
miss — the same portability trap fixed for #91. Normalise with Path(p).as_posix()
in the four affected lambdas. Pure test fix; production code unchanged.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-08-12 03:14:16 -04:00
Cowork SupervisorandClaude Opus 4.8 0b2827e6b8 feat(#93): filesystem permission pre-flight for credential configs (0600/0700)
CI / Tests (py3.10 / ubuntu-latest) (pull_request) Successful in 13s
CI / Tests (py3.12 / ubuntu-latest) (pull_request) Successful in 13s
CI / Lint (ruff) (pull_request) Successful in 8s
CI / Tests (py3.12 / windows-latest) (pull_request) Failing after 24s
CI / Tests (py3.13 / ubuntu-latest) (pull_request) Successful in 12s
CI / Catalog signature (pull_request) Successful in 8s
ssh-mcp refuses to start if its config is group/world-readable (mode & 0o077 →
throws, requiring dir 0700 / file 0600). A GUI user has no idea what chmod 600
means — they just get a dead server. This checks it and offers a one-click fix.

Core (pure, POSIX-only, injectable stat/chmod so tests never touch a real file):
- permission_status(path): the ssh-mcp rule — any group/other bit set (mode & 0o077)
  is not-ok; file must be 0600, its dir 0700. Returns None on Windows (modes don't
  apply) or when the file is absent (nothing to pre-flight). Plain-language problems
  naming the offending octal mode.
- fix_permissions(path): chmod file → 0600, dir → 0700. No-op on Windows; reports
  an OSError instead of raising.
- sidecar_permission_warnings() / sidecar_permission_fix_target(): tie the check to
  #91's sidecar path resolution so it knows WHICH file to inspect. Mirror the
  sidecar-warnings shape.

GUI: a warning label + "Fix permissions" button in the stdio editor (mirrors the
removed-flag surface), shown only when the sidecar exists and is too open. Hidden
on Windows and for non-sidecar servers. Smoke-tested headlessly.

pytest green (529 passed), ruff + format clean. Closes #93. Part of epic #94.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-08-12 03:10:29 -04:00
Cowork SupervisorandClaude Opus 4.8 2e5d0351b4 test(#91): assert sidecar paths via as_posix() for the Windows CI runner
CI / Lint (ruff) (pull_request) Successful in 9s
CI / Tests (py3.12 / windows-latest) (pull_request) Successful in 22s
CI / Tests (py3.10 / ubuntu-latest) (pull_request) Successful in 13s
CI / Tests (py3.12 / ubuntu-latest) (pull_request) Successful in 15s
CI / Tests (py3.13 / ubuntu-latest) (pull_request) Successful in 13s
CI / Catalog signature (pull_request) Successful in 8s
The CI matrix includes a windows-latest job where str(Path("/Users/…")) renders
with backslashes, so the platform-parameterised sidecar-path assertions failed
there (macOS/Linux passed). Compare .as_posix() instead — separator-normalised
and portable — and note why. Pure test fix; no production-code change.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-08-12 03:07:47 -04:00
Cowork SupervisorandClaude Opus 4.8 2cd8e0fb3b feat(#92): detect unpinned npx specs, resolve version, one-click pin, drift
CI / Lint (ruff) (pull_request) Successful in 9s
CI / Tests (py3.10 / ubuntu-latest) (pull_request) Successful in 13s
CI / Tests (py3.12 / windows-latest) (pull_request) Successful in 21s
CI / Tests (py3.12 / ubuntu-latest) (pull_request) Successful in 13s
CI / Tests (py3.13 / ubuntu-latest) (pull_request) Successful in 13s
CI / Catalog signature (pull_request) Successful in 9s
`npx -y ssh-mcp` resolves *latest* on every launch — in one session ssh-mcp went
v1 → v2 and the tool set changed under a running agent, mid-task, with no warning.
This detects that and turns it into a comprehensible pin/upgrade prompt.

Core (pure, no network — reuses parse_version / is_newer_version / the catalog
spec parsers; the resolved-version lookup is fully injectable for tests):
- server_package_spec / server_package_name — the npm spec an npx-style server runs.
- is_unpinned_spec — bare name or dist-tag (@latest/@next) is unpinned; an exact
  numeric version is pinned.
- resolved_npx_version — reads the local ~/.npm/_npx cache (highest version wins),
  degrades to None cleanly. NO network. find/read injectable.
- pin_spec_transform — rewrite the spec to name@version (mirrors pin_command_path's
  (new_data, note) contract). No-op when already pinned / bad version / not npx.
- version_drift_note — "moved X → Y since you pinned" via numeric is_newer_version.
- version_status — the badge's high-level dict (unpinned / pinned_version /
  resolved_version / can_pin / drift). A _UNSET sentinel lets callers force an
  explicit resolved=None ("unknown") vs. omitting it to do the local lookup.

GUI: a version badge row under the dependency status (mirrors that surface) with a
one-click "Pin to <version>" button, shown only for npx servers; a drift note when
a pinned version has been overtaken. Smoke-tested headlessly.

pytest green (520 passed), ruff + format clean. Closes #92. Part of epic #94.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-08-12 03:04:53 -04:00
Cowork SupervisorandClaude Opus 4.8 8c51c25211 feat(#91): sidecar config detection + precedence + verified paths
CI / Tests (py3.12 / windows-latest) (pull_request) Failing after 23s
CI / Tests (py3.10 / ubuntu-latest) (pull_request) Successful in 46s
CI / Catalog signature (pull_request) Successful in 8s
CI / Lint (ruff) (pull_request) Successful in 36s
CI / Tests (py3.12 / ubuntu-latest) (pull_request) Successful in 15s
CI / Tests (py3.13 / ubuntu-latest) (pull_request) Successful in 16s
ssh-mcp v2 reads a TOML sidecar and only falls back to CLI args when that file
is ABSENT — so BCC's managed --host/--user args can be completely inert while
the real config lives in a file BCC never looks at. This adds read-only
truth-telling for that (no sidecar writing).

Core (pure, fully injectable platform/environ/home/exists/read for testing):
- sidecar_path(spec): VERIFIED per-platform TOML location from the package source,
  NOT the README (macOS → ~/Library/Application Support/ssh-mcp, Windows → %APPDATA%,
  else → ${XDG_CONFIG_HOME:-~/.config}). sidecar_doc_path() is the README path.
- sidecar_status(): resolves exists / has_managed_args / args_inert / wrong_path.
- sidecar_warnings(): mirrors removed_flag_warnings' shape. Reports:
  * precedence — "these arguments are inert; the server reads <real path>"
  * wrong-path — a TOML at the README path the server never actually reads
  * #11 credential scoping — an unprefixed SSH_MCP_PASSWORD shared across 2+
    profiles in a multi-profile sidecar (count_toml_profiles is a documented
    3.10-safe heuristic; unscoped_credential_warning gates on it).

GUI: a read-only advisory label in the stdio editor (mirrors the removed-flag
label; no fix button — editing the sidecar is a separate deliberate action).
Uses the real platform/env/filesystem so it reflects this machine.

pytest green, ruff + format clean. Closes #91. Part of epic #94.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-08-12 02:59:49 -04:00
Cowork SupervisorandClaude Opus 4.8 e087107710 feat(#90): promote FLAG_ENV_MIGRATIONS into a ServerSpec spine
CI / Lint (ruff) (pull_request) Successful in 7s
CI / Tests (py3.10 / ubuntu-latest) (pull_request) Successful in 11s
CI / Tests (py3.12 / windows-latest) (pull_request) Successful in 25s
CI / Tests (py3.12 / ubuntu-latest) (pull_request) Successful in 10s
CI / Tests (py3.13 / ubuntu-latest) (pull_request) Successful in 11s
CI / Catalog signature (pull_request) Successful in 6s
Introduce the server-package axis (orthogonal to ClientSpec): a frozen
ServerSpec dataclass + SERVER_SPECS registry that the sidecar (#91),
version-pin (#92) and permission (#93) work all hang off.

- ServerSpec carries env_flags / removed_flags / drift_flags / sidecar_paths
  / sidecar_doc_path / schema. FLAG_ENV_MIGRATIONS is now a derived view of
  the registry, so every existing reader and the migration functions keep the
  exact shape #89 shipped — the migration LOGIC is unchanged, only the DATA grew.
- resolve_server_spec(data) is the ServerSpec entry point; detect_migratable_package
  is a thin name-only wrapper over it (behaviour identical).
- Add the verified #4 follow-ups: --sudoPassword AND --suPassword now auto-migrate
  to SSH_MCP_SUDO_PASSWORD (two flags → one var; existing no-clobber handles it).
  --disableSudo stays warn-only (sudo is now a role/policy, no env replacement).
- Add drift_warnings(): --maxChars=none changed meaning (v1 silently capped at
  5000 chars). Warn-only, no auto-fix; matches inline and separate arg forms.
- Seed ssh-mcp's verified per-platform sidecar TOML paths and zod enums
  (auth/approvalMode/role/port) as data for later issues.

Pure core + tests, no GUI. One #89 test (sudoPassword now migratable) updated to
reflect the intended data growth, with a comment citing the verified facts.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-08-12 02:54:42 -04:00
3 changed files with 1289 additions and 38 deletions
+130
View File
@@ -819,6 +819,27 @@ class ServerEditor(QFrame):
dep.addWidget(recheck)
outer.addLayout(dep)
# Version status row (#92): for an npx-style server, show the currently
# resolved version and, when the spec is unpinned, a one-click "pin".
# Mirrors the dependency-status surface above. Hidden for everything else.
ver = QHBoxLayout()
self.ver_dot = QLabel("○")
self.ver_label = QLabel("—")
self.ver_label.setObjectName("muted")
self.ver_label.setWordWrap(True)
self.pin_btn = QPushButton("Pin")
self.pin_btn.setToolTip(
"Pin the package spec to the currently-resolved version so it can't "
"change under you on the next launch"
)
self.pin_btn.clicked.connect(self._pin_version)
self.pin_btn.setVisible(False)
ver.addWidget(self.ver_dot)
ver.addWidget(self.ver_label, 1)
ver.addWidget(self.pin_btn)
self.ver_row_widgets = (self.ver_dot, self.ver_label, self.pin_btn)
outer.addLayout(ver)
# Collapsible diagnostics panel
self.diag_card = QFrame()
self.diag_card.setObjectName("diagCard")
@@ -899,6 +920,30 @@ class ServerEditor(QFrame):
rf_row.addWidget(self.removed_flag_warn, 1)
rf_row.addWidget(self.removed_flag_fix_btn)
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.
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)
# 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.
self.perm_warn = QLabel("")
self.perm_warn.setStyleSheet(f"color: {WARN};")
self.perm_warn.setWordWrap(True)
self.perm_warn.hide()
self.perm_fix_btn = QPushButton("Fix permissions")
self.perm_fix_btn.setToolTip("chmod the config file to 0600 and its directory to 0700")
self.perm_fix_btn.clicked.connect(self._fix_permissions)
self.perm_fix_btn.hide()
perm_row = QHBoxLayout()
perm_row.addWidget(self.perm_warn, 1)
perm_row.addWidget(self.perm_fix_btn)
v.addLayout(perm_row)
v.addWidget(self._lbl("Environment variables"))
self.env = KeyValueTable(
"Variable", "Value", on_change=self._emit, before_change=self._before_change
@@ -1000,6 +1045,9 @@ class ServerEditor(QFrame):
self.secret_warn.hide()
self.removed_flag_warn.hide()
self.removed_flag_fix_btn.hide()
self.sidecar_warn.hide()
self.perm_warn.hide()
self.perm_fix_btn.hide()
self._loading = False
return
self.setEnabled(True)
@@ -1089,6 +1137,9 @@ class ServerEditor(QFrame):
self.secret_warn.hide()
self.removed_flag_warn.hide()
self.removed_flag_fix_btn.hide()
self.sidecar_warn.hide()
self.perm_warn.hide()
self.perm_fix_btn.hide()
return
_, notes = core.split_suspicious_args(self._current_arg_lines())
if notes:
@@ -1121,6 +1172,23 @@ class ServerEditor(QFrame):
else:
self.removed_flag_warn.hide()
self.removed_flag_fix_btn.hide()
# Sidecar precedence / wrong-path / credential-scoping (#91). Uses the
# real platform + environment + filesystem so it reflects this machine.
sc_warnings = core.sidecar_warnings(stdio)
if sc_warnings:
self.sidecar_warn.setText("⚠ " + "\n\n".join(sc_warnings))
self.sidecar_warn.show()
else:
self.sidecar_warn.hide()
# Sidecar filesystem permissions (#93). Real platform/fs; no-op on Windows.
perm_warnings = core.sidecar_permission_warnings(stdio)
if perm_warnings:
self.perm_warn.setText("⚠ " + "\n".join(perm_warnings))
self.perm_warn.show()
self.perm_fix_btn.show()
else:
self.perm_warn.hide()
self.perm_fix_btn.hide()
def _fix_args(self):
if self._before_change:
@@ -1141,11 +1209,30 @@ class ServerEditor(QFrame):
self.env.load(migrated.get("env", {}))
self.args.setPlainText("\n".join(migrated.get("args", [])))
def _fix_permissions(self):
"""chmod the sidecar config to 0600 / its dir to 0700 (#93)."""
stdio = {
"command": self.command.text().strip(),
"args": self._current_arg_lines(),
"env": self.env.dump(),
}
target = core.sidecar_permission_fix_target(stdio)
if target is None:
return
changed, note = core.fix_permissions(target)
if not changed:
# Surface the failure in-place rather than silently doing nothing.
self.perm_warn.setText("⚠ " + (note or "could not change permissions"))
return
self._check_args() # re-check; the warning clears when perms are now tight
# --- dependency ------------------------------------------------------ #
def refresh_dependency(self, auto_open=False):
if not self.isEnabled():
self._set_dep({"status": "unknown", "label": "—"})
self.diag_text.clear()
for wdg in self.ver_row_widgets:
wdg.hide()
return
data = self.dump_data()
res = core.check_dependency(data)
@@ -1159,6 +1246,49 @@ class ServerEditor(QFrame):
self.details_btn.setChecked(True) # opens panel (fills text via _toggle_diag)
if self.diag_card.isVisible():
self.diag_text.setPlainText(self._full_diag_text())
self._refresh_version_badge(data)
def _refresh_version_badge(self, data: dict):
"""Show the resolved version / pin state for an npx-style server (#92)."""
st = core.version_status(data) if self.type.currentIndex() == 0 else None
if st is None:
for wdg in self.ver_row_widgets:
wdg.hide()
return
for wdg in self.ver_row_widgets:
wdg.show()
resolved = st["resolved_version"] or "unknown"
if st["drift"]:
glyph, color, text = "●", WARN, f"{st['package']} {st['drift']}"
elif st["unpinned"]:
glyph, color = "●", WARN
text = f"{st['package']} · resolved {resolved} · unpinned (resolves latest each launch)"
else:
glyph, color = "●", GOOD
text = f"{st['package']} · pinned {st['pinned_version']}"
self.ver_dot.setText(glyph)
self.ver_dot.setStyleSheet(f"color: {color}; font-size: 14px;")
self.ver_label.setText(text)
self.ver_label.setStyleSheet(f"color: {color};")
# Offer the pin only when unpinned AND we know what to pin to.
self.pin_btn.setVisible(st["can_pin"])
if st["can_pin"]:
self.pin_btn.setText(f"Pin to {resolved}")
self._version_resolved = st["resolved_version"]
def _pin_version(self):
if self._before_change:
self._before_change()
resolved = getattr(self, "_version_resolved", None)
if not resolved:
return
new_data, note = core.pin_spec_transform(self.dump_data(), resolved)
if not note:
return
self._loading = True
self.args.setPlainText("\n".join(str(a) for a in (new_data.get("args") or [])))
self._loading = False
self._emit() # writes back to the model and re-checks (badge now "pinned")
def _set_dep(self, res: dict):
status = res.get("status", "unknown")
+719 -35
View File
@@ -30,6 +30,7 @@ import tempfile
import threading
import time
from dataclasses import dataclass
from dataclasses import field as _field
from pathlib import Path
from typing import NamedTuple
from urllib.parse import urlparse
@@ -2218,46 +2219,116 @@ def _looks_like_multiple_args(a: str) -> bool:
# --------------------------------------------------------------------------- #
# Removed-flag → env migration
# ServerSpec — the server-package axis (issue #90, epic #94)
#
# Some MCP servers moved credential CLI flags into environment variables across
# a major version (secrets on the command line are visible to any local user in
# process listings). A config written for the old version then fails hard on the
# new one — e.g. ssh-mcp v2 exits with "These flags were removed in v2:
# --password". This registry lets BCC recognise that shape, warn about it, and
# offer a one-click migration that lifts the value out of `args` into `env`.
# BCC already models *which host reads a config* (ClientSpec). ServerSpec is the
# orthogonal axis: *which server package a definition runs*. A given server has
# both — Claude Desktop (client) running ssh-mcp (server), say. #89 shipped the
# seed of this axis as FLAG_ENV_MIGRATIONS (a per-npm-package registry of removed
# credential flags); #90 promotes it into a richer struct the sidecar (#91),
# version-pin (#92), permission (#93) and future schema/auth work all hang off.
#
# Per package:
# "env" removed flag -> env var that now supplies it. BCC auto-migrates
# these: the flag (and its value) leave `args`, the value lands in
# `env` under the mapped name.
# "removed" flags that no longer exist but have no confirmed 1:1 env
# replacement (e.g. a boolean, or an env name not documented here).
# BCC only warns; the user resolves these by hand.
#
# Extending it: add a package key and its flag maps. Keep "env" limited to
# mappings whose env-var name is verified, so the auto-fix stays trustworthy.
# Non-reversing constraint (epic #94): the migration LOGIC below
# (migrate_removed_flags / detect_migratable_package / removed_flag_warnings) is
# unchanged — only the DATA grows. FLAG_ENV_MIGRATIONS is kept as a derived view
# so every existing reader keeps seeing the same shape.
# --------------------------------------------------------------------------- #
FLAG_ENV_MIGRATIONS: dict[str, dict[str, dict[str, str]]] = {
"ssh-mcp": {
"env": {
@dataclass(frozen=True)
class ServerSpec:
"""Everything BCC knows about one MCP *server package*.
Frozen so the module-level registry entries are effectively singletons.
Every collection field defaults empty, so a spec need only fill in the axes
that apply to its package. Fields:
* ``env_flags`` — removed CLI flags whose value now belongs in an env var.
Auto-migratable: ``migrate_removed_flags`` lifts the flag's value out of
``args`` into ``env`` under the mapped name. Two flags may map to one var
(ssh-mcp's ``--sudoPassword``/``--suPassword`` → ``SSH_MCP_SUDO_PASSWORD``);
the no-clobber logic already handles that collision.
* ``removed_flags`` — removed flags with no confirmed 1:1 env replacement
(e.g. a boolean policy toggle). Warn-only; the user resolves by hand.
* ``drift_flags`` — flags/values whose *meaning* changed across a major
version without being removed. Keyed by the exact ``flag=value`` token; the
value is the human explanation. Not auto-fixed — meaning drift needs a human.
* ``sidecar_paths`` — per-platform sidecar config location, VERIFIED from the
package source (not its docs). Keys are ``"darwin"``/``"win32"``/``"posix"``;
values are templates expanded against ``$HOME`` + environment (see #91).
Empty == the package has no sidecar. Consumed by #91/#93.
* ``sidecar_doc_path`` — a path the package's README documents but does NOT
actually read on some platform. Used by #91 to flag a file a user placed by
following the (wrong) docs, which the server silently never reads.
* ``schema`` — validation enums shipped by the package (zod). Seeded as data
now; becomes editor pick-lists later (#7).
"""
package: str
env_flags: dict[str, str] = _field(default_factory=dict)
removed_flags: dict[str, str] = _field(default_factory=dict)
drift_flags: dict[str, str] = _field(default_factory=dict)
sidecar_paths: dict[str, str] = _field(default_factory=dict)
sidecar_doc_path: str = ""
schema: dict = _field(default_factory=dict)
# The registry. Keyed by the normalised npm package name (see _normalize_pkg_token).
# Verified ssh-mcp facts (from the package source, NOT its README — the README's
# sidecar path is wrong on macOS/Windows):
SERVER_SPECS: dict[str, ServerSpec] = {
"ssh-mcp": ServerSpec(
package="ssh-mcp",
# Credential flags removed in v2 — secrets on argv are visible to any
# local user in process listings, so v2 reads them from the environment.
env_flags={
"--password": "SSH_MCP_PASSWORD",
"--sudoPassword": "SSH_MCP_SUDO_PASSWORD",
"--suPassword": "SSH_MCP_SUDO_PASSWORD",
},
"removed": {
"--sudoPassword": (
"removed in ssh-mcp v2 — supply it via an environment variable "
"instead (see the ssh-mcp README's 'Migrating from v1' section)"
),
"--suPassword": (
"removed in ssh-mcp v2 — supply it via an environment variable "
"instead (see the ssh-mcp README's 'Migrating from v1' section)"
),
# --disableSudo is gone with no env replacement: sudo is now a role/policy.
removed_flags={
"--disableSudo": (
"removed in ssh-mcp v2 — sudo is now governed by the server "
"config/policy rather than a CLI flag"
"config/policy (a role/permission) rather than a CLI flag"
),
},
},
# --maxChars=none changed meaning: v1 silently parsed "none" to a 5000-char
# cap (parseInt("none") || 5000), so a config still carrying it doesn't mean
# "unlimited" the way a user reading the flag name would assume.
drift_flags={
"--maxChars=none": (
"changed meaning across ssh-mcp versions — v1 silently parsed "
"'none' as a 5000-character cap (parseInt('none') || 5000), not "
'"unlimited". Confirm the value still means what you intend on the '
"version you run."
),
},
# Sidecar TOML path, verified from the package code (see #91). The README
# documents ~/.config on every platform, but the code resolves the OS's
# native app-data dir on macOS/Windows.
sidecar_paths={
"darwin": "~/Library/Application Support/ssh-mcp/config.toml",
"win32": "${APPDATA}/ssh-mcp/config.toml",
"posix": "${XDG_CONFIG_HOME:-~/.config}/ssh-mcp/config.toml",
},
sidecar_doc_path="~/.config/ssh-mcp/config.toml",
# zod enums shipped by ssh-mcp v2 (data now; pick-lists later — #7).
schema={
"auth": ["agent", "key", "password", "keychain"],
"approvalMode": ["auto", "ask-destructive", "ask-all", "deny"],
"role": ["viewer", "operator", "admin"],
"port": {"min": 1, "max": 65535},
},
),
}
# Backwards-compatible derived view of the registry: {pkg: {"env": ..., "removed": ...}}.
# SERVER_SPECS is the source of truth; this alias preserves the exact shape #89
# shipped so every existing reader of FLAG_ENV_MIGRATIONS (and the migration
# functions below) keeps working with no logic change.
FLAG_ENV_MIGRATIONS: dict[str, dict[str, dict[str, str]]] = {
pkg: {"env": dict(spec.env_flags), "removed": dict(spec.removed_flags)}
for pkg, spec in SERVER_SPECS.items()
}
@@ -2282,11 +2353,13 @@ def _normalize_pkg_token(token: str) -> str:
return t
def detect_migratable_package(data: dict) -> str | None:
"""Return the FLAG_ENV_MIGRATIONS key this stdio server runs, or None.
def resolve_server_spec(data: dict) -> ServerSpec | None:
"""Return the ServerSpec for the package a stdio server runs, or None.
Scans the command and every arg (so ``npx -y ssh-mcp`` and a direct
``command: ssh-mcp`` both resolve), matching on the normalised package name.
This is the ServerSpec-axis entry point; ``detect_migratable_package`` is a
thin name-only wrapper kept for the existing migration callers.
"""
if not isinstance(data, dict):
return None
@@ -2294,11 +2367,21 @@ def detect_migratable_package(data: dict) -> str | None:
tokens.extend(data.get("args") or [])
for tok in tokens:
name = _normalize_pkg_token(tok)
if name in FLAG_ENV_MIGRATIONS:
return name
if name in SERVER_SPECS:
return SERVER_SPECS[name]
return None
def detect_migratable_package(data: dict) -> str | None:
"""Return the registry key for the package this stdio server runs, or None.
Scans the command and every arg (so ``npx -y ssh-mcp`` and a direct
``command: ssh-mcp`` both resolve), matching on the normalised package name.
"""
spec = resolve_server_spec(data)
return spec.package if spec else None
def migrate_removed_flags(data: dict) -> tuple[dict, list[str]]:
"""Move known removed credential flags out of `args` and into `env`.
@@ -2416,6 +2499,607 @@ def removed_flag_warning(data: dict) -> str | None:
return warnings[0] if warnings else None
def drift_warnings(data: dict) -> list[str]:
"""Advisory lines for flags whose *meaning* changed across a major version.
Distinct from ``removed_flag_warnings``: these flags still exist and still
parse, so nothing errors — but the value no longer means what the user set
it to mean (e.g. ssh-mcp's ``--maxChars=none``, which v1 silently capped at
5000 chars). There is no safe auto-fix: only a human knows the intent, so
this is warn-only. Matches both the inline ``--flag=value`` form and the
separate ``--flag value`` form. Empty list == nothing to flag.
"""
spec = resolve_server_spec(data)
if spec is None or not spec.drift_flags:
return []
# Reconstruct the set of "flag=value" tokens present, collapsing the two argv
# spellings into one so a drift key of the form "--flag=value" matches both.
args = [str(a) for a in (data.get("args") or [])]
present: set[str] = set()
i, n = 0, len(args)
while i < n:
a = args[i]
if a.startswith("-") and "=" in a:
present.add(a) # inline: --flag=value
i += 1
continue
if a.startswith("-") and i + 1 < n and not args[i + 1].startswith("-"):
present.add(f"{a}={args[i + 1]}") # separate: --flag value
i += 2
continue
i += 1
out: list[str] = []
for token in sorted(spec.drift_flags):
if token in present:
out.append(f"{spec.package}: {token} {spec.drift_flags[token]}")
return out
# --------------------------------------------------------------------------- #
# Sidecar config detection + precedence (issue #91, epic #94)
#
# Some MCP servers read a config *sidecar* (ssh-mcp v2 reads a TOML file) and
# only fall back to CLI args when that file is ABSENT. So BCC's carefully-managed
# --host/--user args can be completely inert while the real config lives in a file
# BCC never looks at. Two traps this surfaces (read-only — no sidecar writing):
#
# 1. Precedence: when the sidecar exists, the args are inert. Tell the user
# where the file the server actually reads is.
# 2. Verified vs documented paths: ssh-mcp's README says ~/.config/ssh-mcp/…,
# but the code resolves the OS-native app-data dir on macOS/Windows. A user
# following the README writes a file the server never reads, with no error.
#
# Plus the #11 credential-scoping trap: an unprefixed SSH_MCP_PASSWORD is offered
# to *every* profile in a multi-profile TOML — detect that and suggest scoping.
#
# All resolution is injectable (platform / environ / home / exists / read) so the
# logic is unit-testable with fixtures and never touches the real filesystem in CI.
# --------------------------------------------------------------------------- #
# ${VAR} and ${VAR:-default} expansion inside a sidecar path template.
_SIDECAR_VAR_RE = re.compile(r"\$\{([A-Z_][A-Z0-9_]*)(?::-([^}]*))?\}")
# The connection-defining args a sidecar would override (rendering them inert).
_SIDECAR_MANAGED_FLAGS = ("--host", "--user", "--port", "--identity", "--key")
def _sidecar_platform_key(platform: str) -> str:
"""Map a sys.platform-style string to a ServerSpec.sidecar_paths key."""
if platform == "darwin":
return "darwin"
if platform.startswith("win"):
return "win32"
return "posix"
def _expand_sidecar_template(tmpl: str, environ: dict, home: Path) -> str:
"""Expand ${VAR}/${VAR:-default} and a leading ~ in a path template.
Injectable so tests drive it with a fixture environ + home rather than the
real process environment. An unset ${VAR} with no default expands to "".
"""
def sub(m: re.Match) -> str:
var, default = m.group(1), m.group(2)
val = environ.get(var)
if val:
return val
return default if default is not None else ""
s = _SIDECAR_VAR_RE.sub(sub, tmpl)
# A leading ~ (either literal, or introduced by a ${XDG:-~/.config} default).
if s.startswith("~"):
s = str(home) + s[1:]
return s
def sidecar_path(
spec: ServerSpec | None,
*,
platform: str | None = None,
environ: dict | None = None,
home: str | os.PathLike | None = None,
) -> Path | None:
"""The VERIFIED sidecar config path for `spec` on `platform`, or None.
None when the package has no sidecar, or the platform has no entry. Uses the
package-source-derived paths in ServerSpec.sidecar_paths — NOT the README,
which is wrong on macOS/Windows. All inputs are injectable for testing.
"""
if spec is None or not spec.sidecar_paths:
return None
platform = platform if platform is not None else sys.platform
environ = environ if environ is not None else dict(os.environ)
home = Path(home) if home is not None else Path.home()
tmpl = spec.sidecar_paths.get(_sidecar_platform_key(platform))
if not tmpl:
return None
return Path(_expand_sidecar_template(tmpl, environ, home))
def sidecar_doc_path(
spec: ServerSpec | None,
*,
home: str | os.PathLike | None = None,
) -> Path | None:
"""The path the package's README documents (but may not actually read)."""
if spec is None or not spec.sidecar_doc_path:
return None
home = Path(home) if home is not None else Path.home()
return Path(_expand_sidecar_template(spec.sidecar_doc_path, {}, home))
def _has_managed_connection_args(data: dict) -> bool:
"""True if the server carries connection args a sidecar would make inert."""
return any(str(a).split("=", 1)[0] in _SIDECAR_MANAGED_FLAGS for a in data.get("args") or [])
def sidecar_status(
data: dict,
*,
platform: str | None = None,
environ: dict | None = None,
home: str | os.PathLike | None = None,
exists=None,
) -> dict | None:
"""Resolve sidecar presence + precedence for one server. Pure/injectable.
Returns None when the server isn't a package with a sidecar. Otherwise a
dict: ``{package, path, doc_path, exists, doc_exists, has_managed_args,
args_inert, wrong_path}``. ``exists`` is a callable ``Path -> bool`` so tests
inject a virtual filesystem; it defaults to a real ``Path.is_file`` check.
"""
spec = resolve_server_spec(data)
if spec is None:
return None
path = sidecar_path(spec, platform=platform, environ=environ, home=home)
if path is None:
return None
doc = sidecar_doc_path(spec, home=home)
exists = exists if exists is not None else (lambda p: Path(p).is_file())
file_exists = bool(exists(path))
# Only count the doc path as a separate "wrong place" when it's genuinely a
# different location from the real one (on Linux they coincide).
doc_exists = bool(doc is not None and doc != path and exists(doc))
has_managed = _has_managed_connection_args(data)
return {
"package": spec.package,
"path": path,
"doc_path": doc,
"exists": file_exists,
"doc_exists": doc_exists,
"has_managed_args": has_managed,
# Precedence: the sidecar wins, so managed args are inert only when it exists.
"args_inert": file_exists and has_managed,
# A file sits where the README says but not where the server actually reads.
"wrong_path": doc_exists and not file_exists,
}
def count_toml_profiles(text: str) -> int:
"""Best-effort count of profile sections in a TOML sidecar.
ssh-mcp's multi-profile mode defines several named tables; an unprefixed
credential is shared across all of them (the #11 trap). Without a TOML parser
on the 3.10 baseline, this counts top-level ``[table]`` and ``[[array]]``
headers (ignoring comments and dotted sub-keys) as a conservative proxy for
"how many profiles are defined". Heuristic — see #91 notes; the exact ssh-mcp
schema should be confirmed before this drives anything destructive.
"""
seen: set[str] = set()
count = 0
for raw in (text or "").splitlines():
line = raw.strip()
if not line or line.startswith("#"):
continue
m = re.match(r"\[\[?\s*([^\]]+?)\s*\]\]?", line)
if not m:
continue
# Top-level section name (first dotted component), deduped so repeated
# sub-tables of one profile don't inflate the count.
top = m.group(1).split(".", 1)[0].strip().strip("'\"")
if top and top not in seen:
seen.add(top)
count += 1
return count
def unscoped_credential_warning(data: dict, profile_count: int | None = None) -> str | None:
"""Advisory when a bare credential is shared across multiple TOML profiles (#11).
ssh-mcp offers an unprefixed ``SSH_MCP_PASSWORD`` to every profile, so a
single stored password is silently handed to every host defined in a
multi-profile sidecar. Fires only when a bare credential is set AND the
sidecar defines 2+ profiles. ``profile_count`` is injected by the aggregator
(which reads the file); None means "unknown", so nothing is claimed.
"""
spec = resolve_server_spec(data)
if spec is None or profile_count is None or profile_count < 2:
return None
env = data.get("env") or {}
bare = [k for k in ("SSH_MCP_PASSWORD", "SSH_MCP_SUDO_PASSWORD") if k in env]
if not bare:
return None
names = ", ".join(sorted(bare))
return (
f"{spec.package}: {names} is unprefixed, so it is offered to every one of "
f"the {profile_count} profiles in the sidecar. Scope the credential "
f"per-profile so one host's password isn't shared with the others."
)
def sidecar_warnings(
data: dict,
*,
platform: str | None = None,
environ: dict | None = None,
home: str | os.PathLike | None = None,
exists=None,
read_text=None,
) -> list[str]:
"""Human-readable sidecar advisories for a server. Empty list == nothing to flag.
Mirrors ``removed_flag_warnings``' shape so the GUI wiring is identical. Read-only:
reports precedence (args inert), the wrong-path trap, and the #11 unscoped-credential
trap. ``read_text`` is an injectable ``Path -> str`` for the sidecar body (used only
for the profile count); it defaults to a safe real read that degrades to "".
"""
status = sidecar_status(data, platform=platform, environ=environ, home=home, exists=exists)
if status is None:
return []
out: list[str] = []
if status["args_inert"]:
out.append(
f"{status['package']}: these arguments are currently inert — this server "
f"reads its config from {status['path']}, which already exists, and only "
f"falls back to CLI args when that file is absent. Edit the file instead."
)
if status["wrong_path"]:
out.append(
f"{status['package']}: a config file exists at {status['doc_path']} (the "
f"path the README documents) but this server actually reads "
f"{status['path']} — the file you wrote is never loaded. Move it there."
)
# #11: read the sidecar (if present) to count profiles for the credential check.
if status["exists"]:
if read_text is None:
def read_text(p):
try:
return Path(p).read_text(encoding="utf-8", errors="replace")
except OSError:
return ""
cred = unscoped_credential_warning(
data, profile_count=count_toml_profiles(read_text(status["path"]))
)
if cred:
out.append(cred)
return out
# Version resolve + pin + drift (issue #92, epic #94)
#
# `npx -y ssh-mcp` resolves *latest* on every launch. In one working session the
# package went v1 → v2 and the exposed tool set changed under a running agent,
# mid-task, with no warning. This surfaces that: detect an unpinned spec, show
# the currently-resolved version (read locally — NO network calls, degrade to
# "unknown" cleanly), offer a one-click pin, and re-check drift with
# is_newer_version. Reuses parse_version / is_newer_version / the catalog spec
# parsers — additive plumbing on existing primitives.
# --------------------------------------------------------------------------- #
# npx-style launchers that resolve a package spec at run time (the unpinned case).
_NPX_LAUNCHERS = {"npx", "bunx", "pnpx"}
# Sentinel so version_status can tell "argument omitted" (do the local lookup)
# from an explicit resolved=None ("caller knows the version is unknown").
_UNSET = object()
def server_package_spec(data: dict) -> str | None:
"""The npm package spec token an ``npx``-style stdio server launches, or None.
``{"command": "npx", "args": ["-y", "ssh-mcp"]}`` → ``"ssh-mcp"``;
``["-y", "ssh-mcp@2.1.0"]`` → ``"ssh-mcp@2.1.0"``. Only npx-style launchers
are considered — that is where "resolve latest each launch" bites. A direct
binary command (``{"command": "ssh-mcp"}``) has no run-time spec to pin.
"""
if not isinstance(data, dict):
return None
cmd = str(data.get("command", "")).strip()
base = cmd.rsplit("/", 1)[-1] if "/" in cmd else cmd
if base not in _NPX_LAUNCHERS:
return None
return _first_catalog_package_spec([str(a) for a in (data.get("args") or [])])
def server_package_name(data: dict) -> str | None:
"""The bare package name for an npx-style server (spec minus any @version)."""
spec = server_package_spec(data)
return _normalize_pkg_token(spec) if spec else None
def is_unpinned_spec(data: dict) -> bool:
"""True when an npx-style server resolves 'latest' each launch.
Unpinned == a bare name (``ssh-mcp``) or a dist-tag (``ssh-mcp@latest``,
``@next``): anything that is not an exact numeric version. An exact pin
(``ssh-mcp@2.1.0``) is stable and returns False.
"""
spec = server_package_spec(data)
if spec is None:
return False
ver = _catalog_package_spec_version(spec)
# parse_version("latest") -> () (falsy); parse_version("2.1.0") -> (2,1,0).
return not (ver and parse_version(ver))
def parse_package_json_version(text: str) -> str | None:
"""Extract the ``version`` string from a package.json blob, or None."""
try:
d = json.loads(text)
except (ValueError, TypeError):
return None
v = d.get("version") if isinstance(d, dict) else None
return v if isinstance(v, str) and v.strip() else None
def resolved_npx_version(
package: str,
*,
home: str | os.PathLike | None = None,
find=None,
read=None,
) -> str | None:
"""Best-effort *resolved* version of an npx-cached package. NO network.
npx unpacks each package under ``~/.npm/_npx/<hash>/node_modules/<pkg>/``; this
reads the ``version`` from that package.json. When several cache entries exist
(different launches), the highest version wins. ``find`` (a glob callable) and
``read`` (path → text) are injectable so tests drive it with fixtures instead
of a real cache; both default to safe real implementations. Returns None when
nothing is found or the package name is empty — the caller shows "unknown".
"""
if not package:
return None
home = Path(home) if home is not None else Path.home()
pattern = str(home / ".npm" / "_npx" / "*" / "node_modules" / package / "package.json")
if find is None:
find = glob.glob
if read is None:
def read(p):
try:
return Path(p).read_text(encoding="utf-8", errors="replace")
except OSError:
return ""
best: str | None = None
for path in find(pattern):
ver = parse_package_json_version(read(path))
if ver and (best is None or is_newer_version(best, ver)):
best = ver
return best
def pin_spec_transform(data: dict, version: str) -> tuple[dict, str | None]:
"""Rewrite an npx server's package spec to an exact ``name@version`` pin.
Returns ``(new_data, note)``; ``note`` is None when there is nothing to pin
(not an npx server, no resolvable spec, empty version, or already pinned to
that exact version). Mirrors ``pin_command_path``'s contract so the GUI wiring
is identical. Only the package token in ``args`` is touched.
"""
if not version or not parse_version(version):
return data, None
spec = server_package_spec(data)
if spec is None:
return data, None
package = _normalize_pkg_token(spec)
if not package:
return data, None
pinned = f"{package}@{version}"
if spec == pinned:
return data, None
args = [str(a) for a in (data.get("args") or [])]
new_args = [pinned if a == spec else a for a in args]
if new_args == args:
return data, None
new_data = dict(data)
new_data["args"] = new_args
return new_data, f"pinned {package} to {version}"
def version_drift_note(pinned: str | None, resolved: str | None) -> str | None:
"""A 'moved X → Y' note when the resolved version is newer than the pin.
Only fires on a forward move (a package the user pinned that the cache has
since advanced past). Equal or older resolved versions, or unknown inputs,
return None. Uses is_newer_version so the compare is numeric, never lexical.
"""
if not pinned or not resolved:
return None
if is_newer_version(pinned, resolved):
return f"moved {pinned} → {resolved} since you pinned"
return None
def version_status(data: dict, *, resolved: str | None = _UNSET, **lookup) -> dict | None:
"""High-level version state for a server, for the GUI badge. None if N/A.
Returns ``{package, spec, unpinned, pinned_version, resolved_version,
can_pin, drift}``. Pass ``resolved`` to supply the resolved version directly
(including an explicit ``None`` for "unknown"); omit it to have
``resolved_npx_version`` read the local npx cache (injectable via ``**lookup``
→ home/find/read). A "pin" action makes sense when the spec is unpinned AND a
resolved version is known (``can_pin``).
"""
spec = server_package_spec(data)
if spec is None:
return None
package = _normalize_pkg_token(spec)
pinned_version = _catalog_package_spec_version(spec)
if not (pinned_version and parse_version(pinned_version)):
pinned_version = None # a dist-tag ("latest") is not a real pin
if resolved is _UNSET:
resolved = resolved_npx_version(package, **lookup)
unpinned = pinned_version is None
return {
"package": package,
"spec": spec,
"unpinned": unpinned,
"pinned_version": pinned_version,
"resolved_version": resolved,
"can_pin": unpinned and bool(resolved),
"drift": version_drift_note(pinned_version, resolved),
}
# --------------------------------------------------------------------------- #
# Filesystem permission pre-flight (issue #93, epic #94)
#
# ssh-mcp refuses to start if its config is group/world-readable: it throws when
# `mode & 0o077` is set, requiring dir 0700 / file 0600. A GUI user has no idea
# what `chmod 600` means — they just get a dead server. Generalise: any config
# file that carries credentials should be permission-checked. POSIX only — modes
# don't apply on Windows, where this degrades to a clean no-op.
#
# Depends on #91's sidecar path resolution to know WHICH file to check.
# --------------------------------------------------------------------------- #
def _is_windows_platform(platform: str | None) -> bool:
p = platform if platform is not None else sys.platform
return p.startswith("win")
def _permission_mode(path, stat_mode) -> int | None:
"""The 0o777 permission bits of `path`, or None if it can't be stat-ed."""
if stat_mode is not None:
return stat_mode(path)
try:
return os.stat(path).st_mode & 0o777
except OSError:
return None
def permission_status(
path: str | os.PathLike,
*,
platform: str | None = None,
stat_mode=None,
) -> dict | None:
"""Check a credential-bearing config file's mode (and its directory's). Pure.
Returns None on Windows (POSIX modes don't apply) or when the file doesn't
exist (nothing to check). Otherwise a dict:
``{path, mode, file_ok, dir, dir_mode, dir_ok, ok, problems}``.
The rule mirrors ssh-mcp's own guard: any group/other bit set (``mode &
0o077``) is not-ok — the file must be 0600, its directory 0700. ``stat_mode``
is an injectable ``path -> int|None`` (the 0o777 bits) so tests never chmod a
real file; it defaults to a real ``os.stat``.
"""
if _is_windows_platform(platform):
return None
p = Path(path)
fmode = _permission_mode(p, stat_mode)
if fmode is None:
return None # file absent / unreadable -> nothing to pre-flight
dmode = _permission_mode(p.parent, stat_mode)
problems: list[str] = []
file_ok = not (fmode & 0o077)
if not file_ok:
problems.append(
f"the config file is readable by other users (mode {fmode:04o}); "
f"ssh-mcp requires 0600 and refuses to start otherwise"
)
# The directory is only judged when we could read its mode.
dir_ok = dmode is None or not (dmode & 0o077)
if dmode is not None and not dir_ok:
problems.append(
f"the containing directory is accessible to other users (mode {dmode:04o}); "
f"ssh-mcp requires 0700"
)
return {
"path": p,
"mode": fmode,
"file_ok": file_ok,
"dir": p.parent,
"dir_mode": dmode,
"dir_ok": dir_ok,
"ok": file_ok and dir_ok,
"problems": problems,
}
def fix_permissions(
path: str | os.PathLike,
*,
platform: str | None = None,
chmod=None,
) -> tuple[bool, str | None]:
"""Tighten a config file to 0600 and its directory to 0700. POSIX only.
Returns ``(changed, note)``. On Windows: ``(False, None)`` — nothing to do.
``chmod`` is an injectable ``(path, mode) -> None`` so tests don't touch real
files; it defaults to ``os.chmod``. Only the bits that are currently wrong are
reported, but both file and dir are set unconditionally (cheap and idempotent).
"""
if _is_windows_platform(platform):
return False, None
p = Path(path)
chmod = chmod if chmod is not None else os.chmod
try:
chmod(p, 0o600)
chmod(p.parent, 0o700)
except OSError as e:
return False, f"could not change permissions: {e}"
return True, "set the config file to 0600 and its directory to 0700"
def sidecar_permission_warnings(
data: dict,
*,
platform: str | None = None,
environ: dict | None = None,
home: str | os.PathLike | None = None,
stat_mode=None,
) -> list[str]:
"""Permission advisories for a server's sidecar config (#93 over #91's path).
Resolves the ServerSpec sidecar path, and — when that file exists and is
group/other-accessible — returns a plain-language warning per problem. Empty
list == fine, no sidecar, or Windows. Mirrors ``sidecar_warnings``' shape so
the GUI wiring is identical.
"""
if _is_windows_platform(platform):
return []
spec = resolve_server_spec(data)
if spec is None:
return []
path = sidecar_path(spec, platform=platform, environ=environ, home=home)
if path is None:
return []
status = permission_status(path, platform=platform, stat_mode=stat_mode)
if status is None or status["ok"]:
return []
return [f"{spec.package}: {prob}" for prob in status["problems"]]
def sidecar_permission_fix_target(
data: dict,
*,
platform: str | None = None,
environ: dict | None = None,
home: str | os.PathLike | None = None,
) -> Path | None:
"""The sidecar path a "fix permissions" action should chmod, or None."""
if _is_windows_platform(platform):
return None
spec = resolve_server_spec(data)
if spec is None:
return None
return sidecar_path(spec, platform=platform, environ=environ, home=home)
# --------------------------------------------------------------------------- #
# Validation
# --------------------------------------------------------------------------- #
+440 -3
View File
@@ -3169,13 +3169,38 @@ def test_removed_flag_warnings_covers_migratable_and_manual():
assert "--password" in text
assert "sudoPassword" in text
assert "disableSudo" in text
# migrate only touches the confirmed --password mapping
# #90: --password AND --sudoPassword are now both auto-migratable (the
# verified ssh-mcp facts map sudo/su → SSH_MCP_SUDO_PASSWORD). The migration
# LOGIC is unchanged — only the registry data grew — so both move into env.
# --disableSudo has no env replacement (sudo is now a role/policy) → warn-only.
new, _ = c.migrate_removed_flags(data)
assert new["env"] == {"SSH_MCP_PASSWORD": "p"}
assert "--sudoPassword=s" in new["args"]
assert new["env"] == {"SSH_MCP_PASSWORD": "p", "SSH_MCP_SUDO_PASSWORD": "s"}
assert "--sudoPassword=s" not in new["args"]
assert "--disableSudo" in new["args"]
def test_migrate_su_password_maps_to_sudo_env():
# --suPassword is the second flag that maps to the same SSH_MCP_SUDO_PASSWORD var.
data = {"command": "ssh-mcp", "args": ["--suPassword", "rootpw"]}
new, notes = c.migrate_removed_flags(data)
assert new["env"] == {"SSH_MCP_SUDO_PASSWORD": "rootpw"}
assert "args" not in new
assert notes
def test_migrate_sudo_and_su_two_flags_one_var_no_clobber():
# Both sudo flags map to one var; the no-clobber path keeps the first, drops
# the second (different value) with a note rather than silently overwriting.
data = {
"command": "ssh-mcp",
"args": ["--sudoPassword=first", "--suPassword=second"],
}
new, notes = c.migrate_removed_flags(data)
assert new["env"] == {"SSH_MCP_SUDO_PASSWORD": "first"}
assert "args" not in new
assert any("already set" in n for n in notes)
def test_removed_flag_warnings_empty_when_clean():
assert c.removed_flag_warnings({"command": "ssh-mcp", "args": ["--host=h"]}) == []
assert c.removed_flag_warning({"command": "ssh-mcp", "args": ["--host=h"]}) is None
@@ -3208,6 +3233,418 @@ def test_ssh_membermatters_end_to_end():
assert notes
# --------------------------------------------------------------------------- #
# ServerSpec spine (issue #90)
# --------------------------------------------------------------------------- #
def test_server_spec_registry_seeds_ssh_mcp():
spec = c.SERVER_SPECS["ssh-mcp"]
assert spec.package == "ssh-mcp"
# sudo/su both map to the one sudo env var (verified facts).
assert spec.env_flags["--sudoPassword"] == "SSH_MCP_SUDO_PASSWORD"
assert spec.env_flags["--suPassword"] == "SSH_MCP_SUDO_PASSWORD"
assert spec.env_flags["--password"] == "SSH_MCP_PASSWORD"
# disableSudo stays warn-only (no env replacement).
assert "--disableSudo" in spec.removed_flags
assert "--disableSudo" not in spec.env_flags
# Verified sidecar paths seeded per platform (README path differs → doc_path).
assert "ssh-mcp/config.toml" in spec.sidecar_paths["darwin"]
assert "Application Support" in spec.sidecar_paths["darwin"]
assert "APPDATA" in spec.sidecar_paths["win32"]
assert spec.sidecar_doc_path == "~/.config/ssh-mcp/config.toml"
# zod enums seeded as data (pick-lists later — #7).
assert spec.schema["auth"] == ["agent", "key", "password", "keychain"]
assert spec.schema["approvalMode"] == ["auto", "ask-destructive", "ask-all", "deny"]
assert spec.schema["role"] == ["viewer", "operator", "admin"]
assert spec.schema["port"] == {"min": 1, "max": 65535}
def test_flag_env_migrations_is_derived_from_server_specs():
# The legacy constant is now a derived view; it must mirror the registry.
assert set(c.FLAG_ENV_MIGRATIONS) == set(c.SERVER_SPECS)
assert c.FLAG_ENV_MIGRATIONS["ssh-mcp"]["env"] == c.SERVER_SPECS["ssh-mcp"].env_flags
assert c.FLAG_ENV_MIGRATIONS["ssh-mcp"]["removed"] == c.SERVER_SPECS["ssh-mcp"].removed_flags
def test_resolve_server_spec_matches_detect_migratable_package():
for data in (
{"command": "npx", "args": ["-y", "ssh-mcp"]},
{"command": "ssh-mcp", "args": []},
{"command": "/usr/local/bin/ssh-mcp", "args": []},
{"command": "npx", "args": ["some-other"]},
{},
{"url": "https://x"},
):
spec = c.resolve_server_spec(data)
pkg = c.detect_migratable_package(data)
assert (spec.package if spec else None) == pkg
def test_drift_warning_maxchars_none_inline_and_separate():
inline = {"command": "ssh-mcp", "args": ["--maxChars=none"]}
separate = {"command": "ssh-mcp", "args": ["--maxChars", "none"]}
for data in (inline, separate):
warnings = c.drift_warnings(data)
assert len(warnings) == 1
assert "--maxChars=none" in warnings[0]
assert "5000" in warnings[0]
def test_drift_warning_quiet_for_other_values_and_unknown_pkg():
# A real numeric cap is fine; only "none" drifts.
assert c.drift_warnings({"command": "ssh-mcp", "args": ["--maxChars=8000"]}) == []
assert c.drift_warnings({"command": "ssh-mcp", "args": ["--host=h"]}) == []
# Unknown package: nothing to say.
assert c.drift_warnings({"command": "npx", "args": ["other", "--maxChars=none"]}) == []
# --------------------------------------------------------------------------- #
# Sidecar config detection + precedence (issue #91)
# --------------------------------------------------------------------------- #
_SSH = {"command": "npx", "args": ["-y", "ssh-mcp", "--host=h", "--user=u"]}
_HOME = "/Users/tester"
def test_sidecar_path_verified_per_platform():
# NOTE: assert on .as_posix() — the CI matrix includes a Windows runner where
# str(Path("/Users/…")) would render with backslashes. as_posix() normalises
# separators so these platform-parameterised checks are portable.
spec = c.SERVER_SPECS["ssh-mcp"]
mac = c.sidecar_path(spec, platform="darwin", environ={}, home=_HOME)
assert mac.as_posix() == "/Users/tester/Library/Application Support/ssh-mcp/config.toml"
# Windows resolves under %APPDATA%, not ~/.config.
win = c.sidecar_path(
spec, platform="win32", environ={"APPDATA": "C:/Users/t/AppData/Roaming"}, home=_HOME
)
assert "ssh-mcp/config.toml" in win.as_posix()
assert "AppData/Roaming" in win.as_posix()
# POSIX honours XDG_CONFIG_HOME, else ~/.config.
xdg = c.sidecar_path(spec, platform="linux", environ={"XDG_CONFIG_HOME": "/cfg"}, home=_HOME)
assert xdg.as_posix() == "/cfg/ssh-mcp/config.toml"
default = c.sidecar_path(spec, platform="linux", environ={}, home=_HOME)
assert default.as_posix() == "/Users/tester/.config/ssh-mcp/config.toml"
def test_sidecar_path_none_for_non_sidecar_package():
assert c.sidecar_path(None) is None
assert c.sidecar_path(c.ServerSpec(package="nope")) is None
def test_sidecar_args_inert_when_file_exists():
real = c.sidecar_path(c.SERVER_SPECS["ssh-mcp"], platform="darwin", environ={}, home=_HOME)
st = c.sidecar_status(
_SSH, platform="darwin", environ={}, home=_HOME, exists=lambda p: p == real
)
assert st["exists"] is True
assert st["has_managed_args"] is True
assert st["args_inert"] is True
assert st["wrong_path"] is False
warns = c.sidecar_warnings(
_SSH, platform="darwin", environ={}, home=_HOME, exists=lambda p: p == real
)
assert warns and "inert" in warns[0]
assert str(real) in warns[0]
def test_sidecar_args_live_when_file_absent():
st = c.sidecar_status(_SSH, platform="darwin", environ={}, home=_HOME, exists=lambda p: False)
assert st["exists"] is False
assert st["args_inert"] is False
assert (
c.sidecar_warnings(_SSH, platform="darwin", environ={}, home=_HOME, exists=lambda p: False)
== []
)
def test_sidecar_wrong_path_flag_on_macos():
# A TOML written at the README's ~/.config path is never read on macOS.
doc = c.sidecar_doc_path(c.SERVER_SPECS["ssh-mcp"], home=_HOME)
assert doc.as_posix() == "/Users/tester/.config/ssh-mcp/config.toml"
st = c.sidecar_status(
_SSH, platform="darwin", environ={}, home=_HOME, exists=lambda p: p == doc
)
assert st["wrong_path"] is True
assert st["args_inert"] is False # the real file doesn't exist, so args still apply
warns = c.sidecar_warnings(
_SSH, platform="darwin", environ={}, home=_HOME, exists=lambda p: p == doc
)
assert warns and "never loaded" in warns[0]
def test_sidecar_no_wrong_path_on_linux_where_paths_coincide():
# On Linux the real path and the doc path are the same, so a file there is
# correctly loaded — no wrong-path warning.
real = c.sidecar_path(c.SERVER_SPECS["ssh-mcp"], platform="linux", environ={}, home=_HOME)
st = c.sidecar_status(
_SSH, platform="linux", environ={}, home=_HOME, exists=lambda p: p == real
)
assert st["wrong_path"] is False
assert st["args_inert"] is True
def test_count_toml_profiles():
assert c.count_toml_profiles("") == 0
assert c.count_toml_profiles("[server]\nhost='h'\n") == 1
text = "# comment\n[[hosts]]\nname='a'\n[[hosts]]\nname='b'\n[settings]\nx=1\n"
# two [[hosts]] share a top-level name -> one profile group; [settings] -> another.
assert c.count_toml_profiles(text) == 2
multi = "[prod]\nhost='p'\n[prod.auth]\nkey='k'\n[staging]\nhost='s'\n"
assert c.count_toml_profiles(multi) == 2
def test_unscoped_credential_warning_11():
data = dict(_SSH, env={"SSH_MCP_PASSWORD": "shared"})
# Single profile: no sharing concern.
assert c.unscoped_credential_warning(data, profile_count=1) is None
# Unknown count: claim nothing.
assert c.unscoped_credential_warning(data, profile_count=None) is None
# Multiple profiles + bare credential: warn.
w = c.unscoped_credential_warning(data, profile_count=3)
assert w and "every one of the 3 profiles" in w
# No bare credential: nothing to warn.
assert c.unscoped_credential_warning(_SSH, profile_count=3) is None
def test_sidecar_warnings_includes_credential_scoping():
real = c.sidecar_path(c.SERVER_SPECS["ssh-mcp"], platform="darwin", environ={}, home=_HOME)
data = dict(_SSH, env={"SSH_MCP_PASSWORD": "shared"})
toml = "[prod]\nhost='p'\n[staging]\nhost='s'\n"
warns = c.sidecar_warnings(
data,
platform="darwin",
environ={},
home=_HOME,
exists=lambda p: p == real,
read_text=lambda p: toml,
)
assert any("profiles" in w for w in warns)
def test_sidecar_status_none_for_unknown_package():
assert c.sidecar_status({"command": "npx", "args": ["other"]}) is None
assert c.sidecar_warnings({"command": "npx", "args": ["other"]}) == []
# Version resolve + pin + drift (issue #92)
# --------------------------------------------------------------------------- #
def test_server_package_spec_and_name():
assert c.server_package_spec({"command": "npx", "args": ["-y", "ssh-mcp"]}) == "ssh-mcp"
assert c.server_package_spec({"command": "npx", "args": ["ssh-mcp@2.1.0"]}) == "ssh-mcp@2.1.0"
assert c.server_package_name({"command": "npx", "args": ["ssh-mcp@2.1.0"]}) == "ssh-mcp"
# A direct binary launch has no run-time spec to pin.
assert c.server_package_spec({"command": "ssh-mcp", "args": []}) is None
assert c.server_package_spec({"url": "https://x"}) is None
def test_is_unpinned_spec():
assert c.is_unpinned_spec({"command": "npx", "args": ["-y", "ssh-mcp"]}) is True
assert c.is_unpinned_spec({"command": "npx", "args": ["ssh-mcp@latest"]}) is True
assert c.is_unpinned_spec({"command": "npx", "args": ["ssh-mcp@next"]}) is True
assert c.is_unpinned_spec({"command": "npx", "args": ["ssh-mcp@2.1.0"]}) is False
assert c.is_unpinned_spec({"command": "ssh-mcp", "args": []}) is False
def test_parse_package_json_version():
assert c.parse_package_json_version('{"name":"ssh-mcp","version":"2.1.0"}') == "2.1.0"
assert c.parse_package_json_version('{"name":"x"}') is None
assert c.parse_package_json_version("not json") is None
assert c.parse_package_json_version('{"version":""}') is None
def test_resolved_npx_version_reads_cache_highest_wins():
# Two cache entries for ssh-mcp; the highest version wins. No real filesystem.
files = {
"/h/.npm/_npx/aaa/node_modules/ssh-mcp/package.json": '{"version":"1.9.0"}',
"/h/.npm/_npx/bbb/node_modules/ssh-mcp/package.json": '{"version":"2.1.0"}',
}
got = c.resolved_npx_version(
"ssh-mcp",
home="/h",
find=lambda pat: list(files),
read=lambda p: files[p],
)
assert got == "2.1.0"
def test_resolved_npx_version_unknown_degrades_to_none():
assert (
c.resolved_npx_version("ssh-mcp", home="/h", find=lambda pat: [], read=lambda p: "") is None
)
assert c.resolved_npx_version("", home="/h") is None
def test_pin_spec_transform():
data = {"command": "npx", "args": ["-y", "ssh-mcp", "--host=h"]}
new, note = c.pin_spec_transform(data, "2.1.0")
assert new["args"] == ["-y", "ssh-mcp@2.1.0", "--host=h"]
assert note and "2.1.0" in note
# Already pinned to that exact version -> no-op.
again, note2 = c.pin_spec_transform(new, "2.1.0")
assert again == new
assert note2 is None
# Bad / empty version -> no-op.
assert c.pin_spec_transform(data, "")[1] is None
assert c.pin_spec_transform(data, "latest")[1] is None
# Not an npx server -> no-op.
assert c.pin_spec_transform({"command": "ssh-mcp", "args": []}, "2.1.0")[1] is None
def test_version_drift_note():
assert c.version_drift_note("2.1.0", "3.0.0") == "moved 2.1.0 → 3.0.0 since you pinned"
assert c.version_drift_note("3.0.0", "3.0.0") is None
assert c.version_drift_note("3.0.0", "2.1.0") is None # never a backwards "drift"
assert c.version_drift_note(None, "3.0.0") is None
assert c.version_drift_note("2.1.0", None) is None
# Numeric, not lexical: 2 < 10.
assert c.version_drift_note("2.0.0", "10.0.0") is not None
def test_version_status_unpinned_offers_pin():
st = c.version_status({"command": "npx", "args": ["-y", "ssh-mcp"]}, resolved="2.1.0")
assert st["package"] == "ssh-mcp"
assert st["unpinned"] is True
assert st["pinned_version"] is None
assert st["resolved_version"] == "2.1.0"
assert st["can_pin"] is True
assert st["drift"] is None
def test_version_status_pinned_reports_drift():
st = c.version_status({"command": "npx", "args": ["ssh-mcp@2.1.0"]}, resolved="3.0.0")
assert st["unpinned"] is False
assert st["pinned_version"] == "2.1.0"
assert st["can_pin"] is False # already pinned
assert st["drift"] == "moved 2.1.0 → 3.0.0 since you pinned"
def test_version_status_unknown_resolved_cannot_pin():
st = c.version_status({"command": "npx", "args": ["-y", "ssh-mcp"]}, resolved=None)
assert st["unpinned"] is True
assert st["resolved_version"] is None
assert st["can_pin"] is False # nothing to pin TO
assert st["drift"] is None
def test_version_status_none_for_non_npx():
assert c.version_status({"command": "ssh-mcp", "args": []}) is None
assert c.version_status({"url": "https://x"}) is None
# --------------------------------------------------------------------------- #
# Filesystem permission pre-flight (issue #93)
# --------------------------------------------------------------------------- #
def test_permission_status_ok_when_0600():
# NOTE: key the injected mode map on Path(p).as_posix() — on the Windows CI
# runner permission_status wraps the path in a WindowsPath, so str(p) would
# use backslashes and miss the lookup (the same portability trap as #91).
st = c.permission_status(
"/cfg/config.toml",
platform="darwin",
stat_mode=lambda p: {"/cfg/config.toml": 0o600, "/cfg": 0o700}.get(Path(p).as_posix()),
)
assert st["ok"] is True
assert st["mode"] == 0o600
assert st["problems"] == []
def test_permission_status_blocks_group_world_readable_file():
st = c.permission_status(
"/cfg/config.toml",
platform="linux",
stat_mode=lambda p: {"/cfg/config.toml": 0o644, "/cfg": 0o755}.get(Path(p).as_posix()),
)
assert st["ok"] is False
assert st["file_ok"] is False
assert st["dir_ok"] is False
# Two plain-language problems, naming the offending octal modes.
text = " ".join(st["problems"])
assert "0644" in text and "0600" in text
assert "0755" in text and "0700" in text
def test_permission_status_file_bad_dir_ok():
st = c.permission_status(
"/cfg/config.toml",
platform="linux",
stat_mode=lambda p: {"/cfg/config.toml": 0o640, "/cfg": 0o700}.get(Path(p).as_posix()),
)
assert st["file_ok"] is False
assert st["dir_ok"] is True
assert len(st["problems"]) == 1
def test_permission_status_none_on_windows_and_missing_file():
# Windows: POSIX modes don't apply -> None (clean no-op).
assert c.permission_status("/cfg/config.toml", platform="win32") is None
# Absent file -> nothing to pre-flight.
assert c.permission_status("/cfg/gone.toml", platform="linux", stat_mode=lambda p: None) is None
def test_fix_permissions_chmods_file_and_dir():
# as_posix() so the recorded paths compare equal on the Windows CI runner too.
calls = []
changed, note = c.fix_permissions(
"/cfg/config.toml",
platform="linux",
chmod=lambda p, m: calls.append((Path(p).as_posix(), m)),
)
assert changed is True
assert ("/cfg/config.toml", 0o600) in calls
assert ("/cfg", 0o700) in calls
assert note and "0600" in note
def test_fix_permissions_noop_on_windows():
calls = []
changed, note = c.fix_permissions(
"/cfg/config.toml", platform="win32", chmod=lambda p, m: calls.append((p, m))
)
assert changed is False
assert note is None
assert calls == []
def test_fix_permissions_reports_oserror():
def boom(p, m):
raise OSError("nope")
changed, note = c.fix_permissions("/cfg/config.toml", platform="linux", chmod=boom)
assert changed is False
assert "could not change permissions" in note
def test_sidecar_permission_warnings_over_real_path():
data = {"command": "npx", "args": ["-y", "ssh-mcp", "--host=h"]}
real = c.sidecar_path(c.SERVER_SPECS["ssh-mcp"], platform="darwin", environ={}, home="/Users/t")
def stat_mode(p):
return 0o644 if Path(p) == real else 0o700
warns = c.sidecar_permission_warnings(
data, platform="darwin", environ={}, home="/Users/t", stat_mode=stat_mode
)
assert warns and warns[0].startswith("ssh-mcp:")
assert "0600" in warns[0]
# Windows / unknown package -> nothing.
assert c.sidecar_permission_warnings(data, platform="win32") == []
assert c.sidecar_permission_warnings({"command": "npx", "args": ["other"]}) == []
def test_sidecar_permission_warnings_quiet_when_tight():
data = {"command": "npx", "args": ["-y", "ssh-mcp", "--host=h"]}
warns = c.sidecar_permission_warnings(
data, platform="darwin", environ={}, home="/Users/t", stat_mode=lambda p: 0o600
)
assert warns == []
def test_sidecar_permission_fix_target():
data = {"command": "npx", "args": ["-y", "ssh-mcp"]}
tgt = c.sidecar_permission_fix_target(data, platform="darwin", environ={}, home="/Users/t")
assert tgt is not None and tgt.name == "config.toml"
assert c.sidecar_permission_fix_target(data, platform="win32") is None
assert c.sidecar_permission_fix_target({"command": "npx", "args": ["other"]}) is None
# --------------------------------------------------------------------------- #
# Move to environment variable (issue #83)
# --------------------------------------------------------------------------- #