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