Land ServerSpec cut: spine + sidecar + version + perms (#90–#93) #100

Merged
the_og merged 9 commits from feat/serverspec-cut into main 2026-08-12 23:52:59 -04:00
3 changed files with 1289 additions and 38 deletions
+130
View File
@@ -819,6 +819,27 @@ class ServerEditor(QFrame):
dep.addWidget(recheck) dep.addWidget(recheck)
outer.addLayout(dep) 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 # Collapsible diagnostics panel
self.diag_card = QFrame() self.diag_card = QFrame()
self.diag_card.setObjectName("diagCard") 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_warn, 1)
rf_row.addWidget(self.removed_flag_fix_btn) rf_row.addWidget(self.removed_flag_fix_btn)
v.addLayout(rf_row) 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")) v.addWidget(self._lbl("Environment variables"))
self.env = KeyValueTable( self.env = KeyValueTable(
"Variable", "Value", on_change=self._emit, before_change=self._before_change "Variable", "Value", on_change=self._emit, before_change=self._before_change
@@ -1000,6 +1045,9 @@ class ServerEditor(QFrame):
self.secret_warn.hide() self.secret_warn.hide()
self.removed_flag_warn.hide() self.removed_flag_warn.hide()
self.removed_flag_fix_btn.hide() self.removed_flag_fix_btn.hide()
self.sidecar_warn.hide()
self.perm_warn.hide()
self.perm_fix_btn.hide()
self._loading = False self._loading = False
return return
self.setEnabled(True) self.setEnabled(True)
@@ -1089,6 +1137,9 @@ class ServerEditor(QFrame):
self.secret_warn.hide() self.secret_warn.hide()
self.removed_flag_warn.hide() self.removed_flag_warn.hide()
self.removed_flag_fix_btn.hide() self.removed_flag_fix_btn.hide()
self.sidecar_warn.hide()
self.perm_warn.hide()
self.perm_fix_btn.hide()
return return
_, notes = core.split_suspicious_args(self._current_arg_lines()) _, notes = core.split_suspicious_args(self._current_arg_lines())
if notes: if notes:
@@ -1121,6 +1172,23 @@ class ServerEditor(QFrame):
else: else:
self.removed_flag_warn.hide() self.removed_flag_warn.hide()
self.removed_flag_fix_btn.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): def _fix_args(self):
if self._before_change: if self._before_change:
@@ -1141,11 +1209,30 @@ class ServerEditor(QFrame):
self.env.load(migrated.get("env", {})) self.env.load(migrated.get("env", {}))
self.args.setPlainText("\n".join(migrated.get("args", []))) 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 ------------------------------------------------------ # # --- dependency ------------------------------------------------------ #
def refresh_dependency(self, auto_open=False): def refresh_dependency(self, auto_open=False):
if not self.isEnabled(): if not self.isEnabled():
self._set_dep({"status": "unknown", "label": "—"}) self._set_dep({"status": "unknown", "label": "—"})
self.diag_text.clear() self.diag_text.clear()
for wdg in self.ver_row_widgets:
wdg.hide()
return return
data = self.dump_data() data = self.dump_data()
res = core.check_dependency(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) self.details_btn.setChecked(True) # opens panel (fills text via _toggle_diag)
if self.diag_card.isVisible(): if self.diag_card.isVisible():
self.diag_text.setPlainText(self._full_diag_text()) 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): def _set_dep(self, res: dict):
status = res.get("status", "unknown") status = res.get("status", "unknown")
+718 -34
View File
@@ -30,6 +30,7 @@ import tempfile
import threading import threading
import time import time
from dataclasses import dataclass from dataclasses import dataclass
from dataclasses import field as _field
from pathlib import Path from pathlib import Path
from typing import NamedTuple from typing import NamedTuple
from urllib.parse import urlparse 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 # BCC already models *which host reads a config* (ClientSpec). ServerSpec is the
# a major version (secrets on the command line are visible to any local user in # orthogonal axis: *which server package a definition runs*. A given server has
# process listings). A config written for the old version then fails hard on the # both — Claude Desktop (client) running ssh-mcp (server), say. #89 shipped the
# new one — e.g. ssh-mcp v2 exits with "These flags were removed in v2: # seed of this axis as FLAG_ENV_MIGRATIONS (a per-npm-package registry of removed
# --password". This registry lets BCC recognise that shape, warn about it, and # credential flags); #90 promotes it into a richer struct the sidecar (#91),
# offer a one-click migration that lifts the value out of `args` into `env`. # version-pin (#92), permission (#93) and future schema/auth work all hang off.
# #
# Per package: # Non-reversing constraint (epic #94): the migration LOGIC below
# "env" removed flag -> env var that now supplies it. BCC auto-migrates # (migrate_removed_flags / detect_migratable_package / removed_flag_warnings) is
# these: the flag (and its value) leave `args`, the value lands in # unchanged — only the DATA grows. FLAG_ENV_MIGRATIONS is kept as a derived view
# `env` under the mapped name. # so every existing reader keeps seeing the same shape.
# "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.
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
FLAG_ENV_MIGRATIONS: dict[str, dict[str, dict[str, str]]] = { @dataclass(frozen=True)
"ssh-mcp": { class ServerSpec:
"env": { """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", "--password": "SSH_MCP_PASSWORD",
"--sudoPassword": "SSH_MCP_SUDO_PASSWORD",
"--suPassword": "SSH_MCP_SUDO_PASSWORD",
}, },
"removed": { # --disableSudo is gone with no env replacement: sudo is now a role/policy.
"--sudoPassword": ( removed_flags={
"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": ( "--disableSudo": (
"removed in ssh-mcp v2 — sudo is now governed by the server " "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 return t
def detect_migratable_package(data: dict) -> str | None: def resolve_server_spec(data: dict) -> ServerSpec | None:
"""Return the FLAG_ENV_MIGRATIONS key this stdio server runs, or 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 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. ``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): if not isinstance(data, dict):
return None return None
@@ -2294,11 +2367,21 @@ def detect_migratable_package(data: dict) -> str | None:
tokens.extend(data.get("args") or []) tokens.extend(data.get("args") or [])
for tok in tokens: for tok in tokens:
name = _normalize_pkg_token(tok) name = _normalize_pkg_token(tok)
if name in FLAG_ENV_MIGRATIONS: if name in SERVER_SPECS:
return name return SERVER_SPECS[name]
return None 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]]: def migrate_removed_flags(data: dict) -> tuple[dict, list[str]]:
"""Move known removed credential flags out of `args` and into `env`. """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 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 # Validation
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
+440 -3
View File
@@ -3169,13 +3169,38 @@ def test_removed_flag_warnings_covers_migratable_and_manual():
assert "--password" in text assert "--password" in text
assert "sudoPassword" in text assert "sudoPassword" in text
assert "disableSudo" 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) new, _ = c.migrate_removed_flags(data)
assert new["env"] == {"SSH_MCP_PASSWORD": "p"} assert new["env"] == {"SSH_MCP_PASSWORD": "p", "SSH_MCP_SUDO_PASSWORD": "s"}
assert "--sudoPassword=s" in new["args"] assert "--sudoPassword=s" not in new["args"]
assert "--disableSudo" 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(): def test_removed_flag_warnings_empty_when_clean():
assert c.removed_flag_warnings({"command": "ssh-mcp", "args": ["--host=h"]}) == [] assert c.removed_flag_warnings({"command": "ssh-mcp", "args": ["--host=h"]}) == []
assert c.removed_flag_warning({"command": "ssh-mcp", "args": ["--host=h"]}) is None 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 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) # Move to environment variable (issue #83)
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #