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

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

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

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

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

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

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

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

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

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-08-12 03:04:53 -04:00
3 changed files with 655 additions and 0 deletions
+111
View File
@@ -819,6 +819,27 @@ class ServerEditor(QFrame):
dep.addWidget(recheck) dep.addWidget(recheck)
outer.addLayout(dep) outer.addLayout(dep)
# Version status row (#92): for an npx-style server, show the currently
# resolved version and, when the spec is unpinned, a one-click "pin".
# Mirrors the dependency-status surface above. Hidden for everything else.
ver = QHBoxLayout()
self.ver_dot = QLabel("○")
self.ver_label = QLabel("—")
self.ver_label.setObjectName("muted")
self.ver_label.setWordWrap(True)
self.pin_btn = QPushButton("Pin")
self.pin_btn.setToolTip(
"Pin the package spec to the currently-resolved version so it can't "
"change under you on the next launch"
)
self.pin_btn.clicked.connect(self._pin_version)
self.pin_btn.setVisible(False)
ver.addWidget(self.ver_dot)
ver.addWidget(self.ver_label, 1)
ver.addWidget(self.pin_btn)
self.ver_row_widgets = (self.ver_dot, self.ver_label, self.pin_btn)
outer.addLayout(ver)
# Collapsible diagnostics panel # Collapsible diagnostics panel
self.diag_card = QFrame() self.diag_card = QFrame()
self.diag_card.setObjectName("diagCard") self.diag_card.setObjectName("diagCard")
@@ -908,6 +929,21 @@ class ServerEditor(QFrame):
self.sidecar_warn.setWordWrap(True) self.sidecar_warn.setWordWrap(True)
self.sidecar_warn.hide() self.sidecar_warn.hide()
v.addWidget(self.sidecar_warn) 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
@@ -1010,6 +1046,8 @@ class ServerEditor(QFrame):
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.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)
@@ -1100,6 +1138,8 @@ class ServerEditor(QFrame):
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.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:
@@ -1140,6 +1180,15 @@ class ServerEditor(QFrame):
self.sidecar_warn.show() self.sidecar_warn.show()
else: else:
self.sidecar_warn.hide() 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:
@@ -1160,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)
@@ -1178,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")
+323
View File
@@ -2777,6 +2777,329 @@ def sidecar_warnings(
return out 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
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
+221
View File
@@ -3424,6 +3424,227 @@ def test_sidecar_status_none_for_unknown_package():
assert c.sidecar_warnings({"command": "npx", "args": ["other"]}) == [] 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)
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #