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
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
This commit was merged in pull request #100.
This commit is contained in:
@@ -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
@@ -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
@@ -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)
|
||||||
# --------------------------------------------------------------------------- #
|
# --------------------------------------------------------------------------- #
|
||||||
|
|||||||
Reference in New Issue
Block a user