Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
113 changes: 113 additions & 0 deletions cli/mcp_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@

import argparse
import json
import re
import sys
from pathlib import Path
from typing import Any
Expand Down Expand Up @@ -340,8 +341,104 @@ def _pairs(values: list[str], *, option: str) -> dict[str, str]:
"off": "disabled",
"env": "missing-env",
"cred": "sends-credential",
"pin": "unpinned-package",
}

#: Launchers whose argument names a registry package resolved at start time.
_UNPINNED_LAUNCHERS = frozenset({"npx", "uvx"})

#: Windows filename suffixes ``npx``/``uvx`` may be invoked with.
_LAUNCHER_SUFFIXES = (".cmd", ".exe", ".bat", ".ps1")

#: Options whose own value names the package, per launcher.
_PACKAGE_OPTIONS = {
"npx": ("--package", "-p"),
"uvx": ("--from",),
}

#: Options that take a value which is not the package, per launcher; the
#: next bare token is consumed as that value and never scanned as a package.
_VALUE_OPTIONS = {
"npx": frozenset({"--cache", "--registry", "--call", "-c"}),
"uvx": frozenset({"--python", "-p", "--with", "--index-url"}),
}

#: npm reads a partial version (``pkg@1``, ``pkg@1.x``) as a range, so only
#: a full ``x.y.z`` with an optional pre-release suffix pins it down.
_NPM_EXACT_VERSION = re.compile(r"\d+\.\d+\.\d+(?:-[0-9A-Za-z.-]+)?")


def _launcher_package_spec(args: list[str], launcher: str) -> str | None:
"""First registry-package token on a launcher command line, if any.

The package may arrive as a positional argument or through the
launcher's own option (``--package``/``-p`` for ``npx``, ``--from``
for ``uvx``); the option's value decides, inline ``--from=x`` alike.
Any other option that takes a value (``--cache``, ``--with``,
``--python``) consumes its value without scanning it, so a value can
never be mistaken for the package (``uvx --with httpx pkg`` scans
``pkg``, not ``httpx``), while a valueless flag (``-y``) consumes
nothing. URL tokens are skipped, and a path-like token
(``./server.js``, ``/opt/x.js``, ``C:\\x.js``) names a local file
rather than a package, so there is no version to pin and the scan
stops. Only the first package token decides: later positional
arguments are the server's own operands (``npx -y pkg /data``).
"""
package_options = _PACKAGE_OPTIONS.get(launcher, ())
value_options = _VALUE_OPTIONS.get(launcher, frozenset())
expect: str | None = None # next token is a package value or a skipped value
for arg in args:
if expect is not None:
if expect == "package":
return arg
expect = None
continue
if arg.startswith("-"):
name, eq, inline = arg.partition("=")
if name in package_options:
if eq:
return inline
expect = "package"
elif name in value_options and not eq:
expect = "skip"
continue
if "://" in arg:
continue
if arg.startswith((".", "/", "\\")) or (
len(arg) >= 2 and arg[1] == ":" and arg[0].isalpha()
):
return None
return arg
return None


def _package_spec_is_unpinned(spec: str, launcher: str) -> bool:
"""Whether a package spec floats: no version, or not one exact release.

``pkg`` and ``pkg@latest`` (or ``pkg@next``) resolve to whatever the
registry serves now, so the effective code was never chosen; ``pkg@1.2.3``
and pip's ``pkg==1.2.3`` name exact releases (an optional leading ``v``
is accepted). npm reads a partial version as a range (``pkg@1`` means
every ``1.x``), so ``npx`` counts only a full ``x.y.z`` -- optionally
with a pre-release suffix such as ``-beta.1`` -- as pinned; uvx's
``==`` matches exactly at any precision, so a leading digit still
counts there. A leading ``@scope/`` belongs to the name, so the ``@``
delimiter is only searched past position 0.
"""
double_equals = spec.find("==")
if double_equals >= 0:
version = spec[double_equals + 2 :]
else:
at = spec.find("@", 1) if spec.startswith("@") else spec.find("@")
if at < 0:
return True
version = spec[at + 1 :]
if version.startswith("v"):
version = version[1:]
if launcher == "npx":
return _NPM_EXACT_VERSION.fullmatch(version) is None
return not version or not version[0].isdigit()


def _mcp_concerns(server: dict[str, Any]) -> tuple[str, ...]:
"""Posture facts an operator should look at before trusting a server.
Expand All @@ -353,6 +450,12 @@ def _mcp_concerns(server: dict[str, Any]) -> tuple[str, ...]:
:func:`_posture_summary` states them once instead of repeating them on
every row. Only key *names* are ever read, never credential values (see
``core.config._resolve_env_refs``).

The package spec a stdio launcher runs is explicit configuration too:
``npx``/``uvx`` given no version (or a floating tag such as
``@latest``) fetch whatever the registry serves at each start, so that
choice is recorded on its row. Other commands run what they name and
are not scanned.
"""
concerns: list[str] = []
if not server["enabled"]:
Expand All @@ -361,6 +464,16 @@ def _mcp_concerns(server: dict[str, Any]) -> tuple[str, ...]:
concerns.append("env")
if server["credentialEnvKeys"]:
concerns.append("cred")
command = server["command"]
if command:
program = command.replace("\\", "/").rsplit("/", 1)[-1].lower()
for suffix in _LAUNCHER_SUFFIXES:
if program.endswith(suffix):
program = program[: -len(suffix)]
if program in _UNPINNED_LAUNCHERS:
spec = _launcher_package_spec(server["args"], program)
if spec is not None and _package_spec_is_unpinned(spec, program):
concerns.append("pin")
return tuple(concerns)


Expand Down
159 changes: 159 additions & 0 deletions tests/test_mcp_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -272,3 +272,162 @@ def test_mcp_list_names_concerns_and_states_default_posture_once(
"Default posture among 3 enabled servers: "
"2 expose all tools, 2 add no MCP approval gate."
)


def test_mcp_list_flags_unpinned_launcher_packages(
tmp_path: Path,
monkeypatch,
capsys,
) -> None:
home = tmp_path / "home"
home.mkdir()
monkeypatch.setenv("DEEPCODE_HOME", str(home))
(home / "deepcode_config.json").write_text(
json.dumps(
{
"mcpServers": {
"npx_tag": {
"type": "stdio",
"command": "npx",
"args": ["-y", "left-pad@latest"],
},
"npx_noversion": {
"type": "stdio",
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-filesystem"],
},
"npx_pinned": {
"type": "stdio",
"command": "npx",
"args": ["-y", "left-pad@1.2.3"],
},
"npx_flags_only": {
"type": "stdio",
"command": "npx",
"args": ["--version"],
},
"npx_local_file": {
"type": "stdio",
"command": "npx",
"args": ["./server.js"],
},
"npx_windows": {
"type": "stdio",
"command": "C:\\Program Files\\nodejs\\npx.cmd",
"args": ["-y", "left-pad"],
},
"uvx_noversion": {
"type": "stdio",
"command": "uvx",
"args": ["mcp-server-time"],
},
"uvx_pinned": {
"type": "stdio",
"command": "uvx",
"args": ["mcp-server-time==0.3.0"],
},
"python": {
"type": "stdio",
"command": "python3",
"args": ["server.py"],
},
"npx_cache_value": {
"type": "stdio",
"command": "npx",
"args": ["--cache", "/tmp/c", "-y", "left-pad"],
},
"npx_package_option_pinned": {
"type": "stdio",
"command": "npx",
"args": ["--package", "left-pad@1.2.3", "-y"],
},
"npx_package_short_unpinned": {
"type": "stdio",
"command": "npx",
"args": ["-y", "-p", "left-pad", "cowsay@2.0.0"],
},
"npx_major_only": {
"type": "stdio",
"command": "npx",
"args": ["-y", "left-pad@1"],
},
"npx_minor_wild": {
"type": "stdio",
"command": "npx",
"args": ["-y", "left-pad@1.x"],
},
"npx_prerelease": {
"type": "stdio",
"command": "npx",
"args": ["-y", "left-pad@1.2.3-beta.1"],
},
"uvx_python_option": {
"type": "stdio",
"command": "uvx",
"args": ["--python", "3.12", "mcp-server-time==0.3.0"],
},
"uvx_python_short": {
"type": "stdio",
"command": "uvx",
"args": ["-p", "3.12", "mcp-server-time==0.3.0"],
},
"uvx_with_option": {
"type": "stdio",
"command": "uvx",
"args": ["--with", "httpx", "mcp-server-time==0.3.0"],
},
"uvx_from_option": {
"type": "stdio",
"command": "uvx",
"args": ["--from", "mcp-server-time==0.3.0"],
},
}
}
),
encoding="utf-8",
)

assert run(["list"]) == 0
header, *rows, summary = capsys.readouterr().out.splitlines()
assert header.split() == ["CONFIG", "AUTH", "RUNTIME", "SCOPE", "NAME", "CONCERNS"]
concerns: dict[str, str] = {}
for row in rows:
fields = row.split(maxsplit=5)
concerns[fields[4]] = fields[5]
# The posture line counts only enabled servers whose configuration
# resolves, and command availability differs by host (a Windows path or
# ``uvx`` may be absent here), so derive the denominator from the rows.
readable = sum(
1 for row in rows if row.split(maxsplit=1)[0] not in {"invalid", "disabled"}
)
assert readable >= 1
assert summary.startswith(f"Default posture among {readable} enabled servers")
# ``npx``/``uvx`` resolve their package argument from the registry on
# every start, so a missing version or a floating tag is the concern;
# a pinned version, a local file, a non-launcher command and a command
# line without a package are not. The package is read from the
# launcher's own option when one names it, other options' values are
# skipped rather than scanned, and npm's partial versions (``@1``,
# ``@1.x``) are ranges -- only a full ``x.y.z``, with an optional
# pre-release suffix, counts as pinned there.
assert concerns == {
"npx_tag": "unpinned-package",
"npx_noversion": "unpinned-package",
"npx_pinned": "-",
"npx_flags_only": "-",
"npx_local_file": "-",
"npx_windows": "unpinned-package",
"npx_cache_value": "unpinned-package",
"npx_package_option_pinned": "-",
"npx_package_short_unpinned": "unpinned-package",
"npx_major_only": "unpinned-package",
"npx_minor_wild": "unpinned-package",
"npx_prerelease": "-",
"uvx_noversion": "unpinned-package",
"uvx_pinned": "-",
"uvx_python_option": "-",
"uvx_python_short": "-",
"uvx_with_option": "-",
"uvx_from_option": "-",
"python": "-",
}
Loading