Compare commits
6
Commits
2e5d0351b4
...
0920846c2c
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0920846c2c | ||
|
|
ce82f7b5c3 | ||
|
|
d2c126a60c | ||
|
|
603d24566d | ||
|
|
0b2827e6b8 | ||
|
|
2cd8e0fb3b |
@@ -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")
|
||||
@@ -908,6 +929,21 @@ class ServerEditor(QFrame):
|
||||
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
|
||||
@@ -1010,6 +1046,8 @@ class ServerEditor(QFrame):
|
||||
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)
|
||||
@@ -1100,6 +1138,8 @@ class ServerEditor(QFrame):
|
||||
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:
|
||||
@@ -1140,6 +1180,15 @@ class ServerEditor(QFrame):
|
||||
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:
|
||||
@@ -1160,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)
|
||||
@@ -1178,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")
|
||||
|
||||
+323
@@ -2777,6 +2777,329 @@ def sidecar_warnings(
|
||||
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
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
@@ -3424,6 +3424,227 @@ def test_sidecar_status_none_for_unknown_package():
|
||||
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)
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
Reference in New Issue
Block a user