diff --git a/src/skillspector/nodes/analyzers/mcp_rug_pull.py b/src/skillspector/nodes/analyzers/mcp_rug_pull.py index 31b8f2486..5ede6bedb 100644 --- a/src/skillspector/nodes/analyzers/mcp_rug_pull.py +++ b/src/skillspector/nodes/analyzers/mcp_rug_pull.py @@ -124,9 +124,27 @@ def analyzer_exhausted(self) -> bool: # RP1: Unpinned MCP server references in code or manifest _RP1_NPX_CMD = re.compile( - r"npx\s+(?:-+\w+\s+)*((?:@?[a-zA-Z][\w.-]*/)?[a-zA-Z][\w.-]*)", + r"\bp?npx[ \t]+(?:-+\w+[ \t]+)*((?:@?[a-zA-Z][\w.-]*/)?[a-zA-Z][\w.-]*)", re.IGNORECASE, ) +_RP1_CONFIG_RUNNER = re.compile( + r"^(?P[ \t]*)(?P-[ \t]+)?(?:command|cmd)[ \t]*:[ \t]*" + r"(?P[\"']?)(?P(?:[^\s\"'#]*/)?p?npx)(?P=quote)[ \t]*(?:#.*)?$", + re.IGNORECASE, +) +_RP1_CONFIG_ARGS = re.compile( + r"^(?P[ \t]*)(?P-[ \t]+)?args[ \t]*:[ \t]*(?P.*)$", + re.IGNORECASE, +) +_RP1_CONFIG_KEY = re.compile( + r"^(?P[ \t]*)(?P-[ \t]+)?(?:[\w-]+|\"[^\"]+\"|'[^']+')[ \t]*:" +) +_RP1_CONFIG_ARG_TOKEN = re.compile(r"[\"']([^\"']*)[\"']|([^\s,\[\]#]+)") +_RP1_CONFIG_MAX_LINES = 8 +# Structural windows count sibling keys, not the physical lines in their values. +# Independently bound physical traversal and argument collection per command. +_RP1_CONFIG_MAX_PHYSICAL_LINES = 256 +_RP1_CONFIG_MAX_ARG_LINES = 128 _RP1_UVX_CMD = re.compile( r"(?:uvx|uv\s+tool\s+run)\s+(?:-+\w+\s+)*([a-zA-Z][\w.-]*)", re.IGNORECASE, @@ -161,6 +179,147 @@ def _find_line(content: str, pos: int) -> int: return content.count("\n", 0, pos) + 1 +def _strip_yaml_comment(line: str) -> str: + """Remove an unquoted YAML comment without treating quoted hashes as comments.""" + quote = "" + index = 0 + while index < len(line): + char = line[index] + if quote: + if quote == '"' and char == "\\": + index += 2 + continue + if char == quote: + if quote == "'" and index + 1 < len(line) and line[index + 1] == "'": + index += 2 + continue + quote = "" + elif char in "\"'": + quote = char + elif char == "#" and (index == 0 or line[index - 1].isspace()): + return line[:index] + index += 1 + return line + + +def _iter_config_npx_commands( + content: str, budget: _RugPullBudget, file_path: str +) -> list[tuple[int, str, str]]: + """Find bounded YAML MCP command/args pairs that run an npx-style runner.""" + lines = content.splitlines(keepends=True) + offsets: list[int] = [] + offset = 0 + for line in lines: + offsets.append(offset) + offset += len(line) + + matches: list[tuple[int, str, str]] = [] + for command_index, command_line in enumerate(lines): + budget.check_runtime(file_path) + command = _RP1_CONFIG_RUNNER.fullmatch(command_line.rstrip("\r\n")) + if command is None: + continue + + # The key in "- command:" starts after the sequence marker. Sibling + # keys align with that column, not with the marker's indentation. + command_indent = len(command.group("indent")) + len(command.group("item") or "") + args_index: int | None = None + args_match: re.Match[str] | None = None + for direction in (-1, 1): + if direction == -1 and command.group("item"): + continue # This command is already the first key in its list item. + sibling_count = 0 + for distance in range(1, _RP1_CONFIG_MAX_PHYSICAL_LINES + 1): + index = command_index + direction * distance + if not 0 <= index < len(lines): + break + budget.check_runtime(file_path) + candidate_line = lines[index].rstrip("\r\n") + stripped = candidate_line.strip() + if not stripped or stripped.startswith("#"): + continue + indent = len(candidate_line) - len(candidate_line.lstrip(" \t")) + key = _RP1_CONFIG_KEY.match(candidate_line) + key_indent = ( + len(key.group("indent")) + len(key.group("item") or "") + if key is not None + else indent + ) + # Backward traversal may reach the first key of this list item. + # Forward traversal must never enter the next item, even when + # its key has the same effective column. + first_item_key = key is not None and key.group("item") is not None + if indent < command_indent and not ( + direction == -1 and first_item_key and key_indent == command_indent + ): + break + if key_indent == command_indent: + if key is None: + # An indentless sequence belongs to a sibling value, + # not a new server mapping. Other scalar lines end it. + if not stripped.startswith("- ") and stripped != "-": + break + else: + sibling_count += 1 + if sibling_count > _RP1_CONFIG_MAX_LINES: + break + candidate_args = _RP1_CONFIG_ARGS.fullmatch(candidate_line) + if candidate_args is not None: + args_index = index + args_match = candidate_args + break + if direction == -1 and first_item_key and key_indent == command_indent: + break + if args_match is not None: + break + + if args_index is None or args_match is None: + continue + + args_indent = len(args_match.group("indent")) + len(args_match.group("item") or "") + args_lines = [_strip_yaml_comment(args_match.group("value"))] + args_end_index = args_index + for index in range( + args_index + 1, min(len(lines), args_index + _RP1_CONFIG_MAX_ARG_LINES + 1) + ): + budget.check_runtime(file_path) + candidate_line = lines[index].rstrip("\r\n") + stripped = candidate_line.strip() + if not stripped or stripped.startswith("#"): + continue + indent = len(candidate_line) - len(candidate_line.lstrip(" \t")) + indentless_item = indent == args_indent and ( + stripped.startswith("- ") or stripped == "-" + ) + if indent < args_indent or (indent == args_indent and not indentless_item): + break + # A mapping item is not a scalar package argument and may start + # another server; never consume its keys as command arguments. + if indentless_item and _RP1_CONFIG_KEY.match(candidate_line): + break + args_lines.append(_strip_yaml_comment(candidate_line.lstrip(" \t"))) + args_end_index = index + + args_text = " ".join(args_lines) + for token_match in _RP1_CONFIG_ARG_TOKEN.finditer(args_text): + quoted_token = token_match.group(1) + token = quoted_token if quoted_token is not None else token_match.group(2) + if token.startswith("-"): + continue + if not token: + # The first positional argument is empty, not a package name. + # Do not shift a later argument into its position or invent RP1. + break + start = offsets[command_index] + full_match = "".join( + lines[min(command_index, args_index) : max(command_index, args_end_index) + 1] + ).strip() + matches.append((start, full_match, token)) + break + + return matches + + def _normalize_string_list( lst: list[object] | None, budget: _RugPullBudget | None = None, @@ -260,6 +419,35 @@ def _check_rp1( ) ) + # YAML MCP configs often place the runner and package in separate fields. + for start, full_match, package in _iter_config_npx_commands(content, budget, file_path): + budget.check_runtime(file_path) + if _VERSION_PIN_RE.search(package): + continue + line_num = _find_line(content, start) + budget.emit( + Finding( + rule_id="RP1", + message=( + f"MCP server referenced without pinned version: '{full_match[:200]}'." + ), + severity="MEDIUM", + confidence=0.70, + file=file_path, + start_line=line_num, + category=_CATEGORY, + tags=list(_TAGS), + matched_text=full_match[:200], + match_fingerprint=compute_match_fingerprint("RP1", full_match), + explanation=( + "npx-style MCP commands without a version suffix " + "create a rug-pull risk if the upstream server is " + "compromised and publishes a malicious update." + ), + remediation="Pin the version in the args list: @scope/server@1.2.3", + ) + ) + # uvx without ==version for m in _RP1_UVX_CMD.finditer(content): budget.check_runtime(file_path) diff --git a/tests/test_mcp_rug_pull.py b/tests/test_mcp_rug_pull.py index 244936326..dac80c194 100644 --- a/tests/test_mcp_rug_pull.py +++ b/tests/test_mcp_rug_pull.py @@ -19,7 +19,17 @@ import json -from skillspector.nodes.analyzers.mcp_rug_pull import node +import pytest +import yaml + +from skillspector.nodes.analyzers.mcp_rug_pull import ( + _RP1_CONFIG_MAX_ARG_LINES, + _RP1_CONFIG_MAX_PHYSICAL_LINES, + _iter_config_npx_commands, + _RugPullBudget, + _strip_yaml_comment, + node, +) from skillspector.nodes.build_context import build_context from skillspector.nodes.deduplicate import deduplicate from skillspector.nodes.report import report @@ -55,6 +65,321 @@ def test_rp1_npx_unpinned(): assert issue["finding"] == "npx @scope/mcp-server" +def test_rp1_npx_match_does_not_cross_lines(): + """A trailing ``npx`` must not combine with the next line as a command.""" + for content in ( + "---\nname: npx\ndescription: repro\n---\n", + "Install it with npx\nthe package manager.\n", + ): + result = node(_state(file_cache={"SKILL.md": content})) + assert not [finding for finding in result["findings"] if finding.rule_id == "RP1"] + + +def test_rp1_pnpx_unpinned(): + """RP1 also detects pnpm's npx-style runner without a version pin.""" + result = node(_state(file_cache={"setup.sh": "pnpx @scope/mcp-server\n"})) + rp1 = [finding for finding in result["findings"] if finding.rule_id == "RP1"] + + assert len(rp1) == 1 + assert rp1[0].matched_text == "pnpx @scope/mcp-server" + + +def test_rp1_npx_requires_a_word_boundary(): + """An unrelated identifier ending in ``npx`` is not a command.""" + result = node(_state(file_cache={"setup.sh": "foonpx @scope/mcp-server\n"})) + + assert not [finding for finding in result["findings"] if finding.rule_id == "RP1"] + + +def test_rp1_yaml_mcp_config_unpinned(): + """RP1 detects unpinned npx-style commands in YAML MCP config args.""" + configs = ( + """mcpServers:\n fs:\n command: npx\n args: ["-y", "@scope/mcp-server"]\n""", + """servers:\n goose:\n cmd: pnpx\n args:\n - "-y"\n - "@scope/mcp-server"\n""", + ) + + for config in configs: + result = node(_state(file_cache={"mcp.yaml": config})) + rp1 = [finding for finding in result["findings"] if finding.rule_id == "RP1"] + + assert len(rp1) == 1 + assert "@scope/mcp-server" in rp1[0].matched_text + + +def test_rp1_yaml_mcp_config_pinned_no_finding(): + """RP1 skips YAML MCP args whose package token pins a version.""" + configs = ( + """mcpServers:\n fs:\n command: npx\n args: ["-y", "@scope/mcp-server@1.2.3"]\n""", + """servers:\n goose:\n cmd: pnpx\n args:\n - "-y"\n - "@scope/mcp-server@1.2.3"\n""", + ) + + for config in configs: + result = node(_state(file_cache={"mcp.yaml": config})) + + assert not [finding for finding in result["findings"] if finding.rule_id == "RP1"] + + +@pytest.mark.parametrize("style", ["flow", "block"]) +@pytest.mark.parametrize("quote", ['"', "'"]) +@pytest.mark.parametrize( + ("arguments", "expected"), + [ + (["-y", "@scope/server", ""], 1), + (["-y", "@scope/server@1.2.3", ""], 0), + (["", "@scope/server"], 0), + (["-y", "", "@scope/server"], 0), + (["-y", "", "@scope/server@1.2.3"], 0), + ([""], 0), + (["-y", ""], 0), + ], +) +def test_rp1_yaml_empty_arguments_do_not_crash_or_shift_package(style, quote, arguments, expected): + quoted = [quote + argument + quote for argument in arguments] + args = ( + " args: [" + ", ".join(quoted) + "]\n" + if style == "flow" + else " args:\n" + "".join(" - " + argument + "\n" for argument in quoted) + ) + result = node(_state(file_cache={"mcp.yaml": "mcpServers:\n fs:\n command: npx\n" + args})) + rp1 = [finding for finding in result["findings"] if finding.rule_id == "RP1"] + assert len(rp1) == expected + if rp1: + assert rp1[0].start_line == 3 + assert "@scope/server" in rp1[0].matched_text + + +def test_rp1_yaml_empty_package_does_not_abort_other_configs_or_files(): + content = ( + 'mcpServers:\n empty:\n command: npx\n args: ["-y", ""]\n' + ' real:\n command: pnpx\n args: ["@scope/server"]\n' + ) + result = node(_state(file_cache={"mcp.yaml": content, "setup.sh": "npx another-server\n"})) + rp1 = [finding for finding in result["findings"] if finding.rule_id == "RP1"] + assert [(finding.file, finding.start_line) for finding in rp1] == [ + ("mcp.yaml", 6), + ("setup.sh", 1), + ] + assert all(event["outcome"] == "completed" for event in result["inspection_ledger"]) + + +@pytest.mark.parametrize( + "layout", + [ + 'mcpServers:\n fs:\n command: npx\n env:\n FOO: bar\n args: ["-y", "PACKAGE"]\n', + 'mcpServers:\n fs:\n command: npx\n type: stdio\n cwd: /tmp\n description: server\n args: ["-y", "PACKAGE"]\n', + 'servers:\n - command: npx\n args: ["-y", "PACKAGE"]\n', + 'servers:\n - command: pnpx\n type: stdio\n args:\n - "-y"\n - "PACKAGE"\n', + 'mcpServers:\n fs:\n args: ["-y", "PACKAGE"]\n env: {}\n command: npx\n', + 'mcpServers:\n fs:\n args:\n - "-y"\n - "PACKAGE"\n command: npx\n', + 'servers:\n - args: ["-y", "PACKAGE"]\n command: npx\n', + 'servers:\n - name: fs\n args:\n - "-y"\n - "PACKAGE"\n command: npx\n', + 'mcpServers:\n fs:\n command: /usr/local/bin/npx\n args: ["-y", "PACKAGE"]\n', + 'mcpServers:\n fs:\n args: ["-y", "PACKAGE"]\n command: "./node_modules/.bin/pnpx"\n', + ], +) +@pytest.mark.parametrize("pinned", [False, True]) +def test_rp1_yaml_sibling_layouts_preserve_pin_behavior(layout, pinned): + package = "@scope/server@1.2.3" if pinned else "@scope/server" + content = layout.replace("PACKAGE", package) + rp1 = [ + f for f in node(_state(file_cache={"mcp.yaml": content}))["findings"] if f.rule_id == "RP1" + ] + assert len(rp1) == (0 if pinned else 1) + if rp1: + assert "@scope/server" in rp1[0].matched_text + assert rp1[0].start_line == next( + index for index, line in enumerate(content.splitlines(), 1) if "command:" in line + ) + + +@pytest.mark.parametrize("pinned", [False, True]) +@pytest.mark.parametrize("reverse", [False, True]) +@pytest.mark.parametrize("newline", ["\n", "\r\n"]) +@pytest.mark.parametrize( + ("header", "command", "middle", "args"), + [ + ( + "mcpServers:\n fs:\n", + " command: npx\n", + "", + ' args:\n - -y\n - "PACKAGE"\n', + ), + ("servers:\n- name: fs\n", " command: pnpx\n", "", ' args:\n - -y\n - "PACKAGE"\n'), + ( + "mcpServers:\n fs:\n", + " command: npx\n", + " autoApprove:\n - read_file\n", + ' args: ["PACKAGE"]\n', + ), + ( + "mcpServers:\n fs:\n", + " command: npx\n", + " env:\n" + "".join(f" KEY{i}: value\n" for i in range(12)), + ' args: ["PACKAGE"]\n', + ), + ( + "mcpServers:\n fs:\n", + " command: npx\n", + " description: |\n" + " description text\n" * 12, + ' args: ["PACKAGE"]\n', + ), + ( + "mcpServers:\n fs:\n", + " command: npx\n", + "", + " args:\n" + " - -y\n" * 12 + ' - "PACKAGE"\n', + ), + ], +) +def test_rp1_yaml_indentless_and_long_values( + header, command, middle, args, pinned, reverse, newline +): + package = "@scope/server@1.2.3" if pinned else "@scope/server" + pair = args + middle + command if reverse else command + middle + args + content = (header + pair).replace("PACKAGE", package).replace("\n", newline) + parsed = yaml.safe_load(content) + server = parsed["servers"][0] if "servers" in parsed else parsed["mcpServers"]["fs"] + assert server["args"][-1] == package + rp1 = [ + f for f in node(_state(file_cache={"mcp.yaml": content}))["findings"] if f.rule_id == "RP1" + ] + assert len(rp1) == (0 if pinned else 1) + if rp1: + assert rp1[0].start_line == next( + i for i, line in enumerate(content.splitlines(), 1) if "command:" in line + ) + + +@pytest.mark.parametrize("pinned", [False, True]) +def test_rp1_yaml_indentless_args_in_first_key_list_item(pinned): + package = "@scope/server@1.2.3" if pinned else "@scope/server" + content = f'servers:\n- command: npx\n args:\n - -y\n - "{package}"\n' + rp1 = [ + f for f in node(_state(file_cache={"mcp.yaml": content}))["findings"] if f.rule_id == "RP1" + ] + assert len(rp1) == (0 if pinned else 1) + + +@pytest.mark.parametrize("direction", ["before", "after"]) +@pytest.mark.parametrize("at_limit", [True, False]) +def test_rp1_yaml_physical_search_hard_limit(direction, at_limit): + distance = _RP1_CONFIG_MAX_PHYSICAL_LINES + (0 if at_limit else 1) + filler = " # padding\n" * (distance - 1) + command = " command: npx\n" + args = ' args: ["@scope/server"]\n' + pair = args + filler + command if direction == "before" else command + filler + args + matches = _iter_config_npx_commands( + "mcpServers:\n fs:\n" + pair, _RugPullBudget({}), "mcp.yaml" + ) + assert len(matches) == (1 if at_limit else 0) + + +@pytest.mark.parametrize("indent", [" ", " "]) +@pytest.mark.parametrize("at_limit", [True, False]) +def test_rp1_yaml_argument_collection_hard_limit(indent, at_limit): + distance = _RP1_CONFIG_MAX_ARG_LINES + (0 if at_limit else 1) + content = "mcpServers:\n fs:\n command: npx\n args:\n" + content += f"{indent}- -y\n" * (distance - 1) + f'{indent}- "@scope/server"\n' + matches = _iter_config_npx_commands(content, _RugPullBudget({}), "mcp.yaml") + assert len(matches) == (1 if at_limit else 0) + + +def test_rp1_yaml_long_value_traversal_checks_runtime(monkeypatch): + calls = 0 + + def check_runtime(self, path=None): + nonlocal calls + calls += 1 + if calls == 20: + raise RuntimeError("test deadline") + + monkeypatch.setattr(_RugPullBudget, "check_runtime", check_runtime) + content = " command: npx\n env:\n" + " KEY: value\n" * 100 + with pytest.raises(RuntimeError, match="test deadline"): + _iter_config_npx_commands(content, _RugPullBudget({}), "mcp.yaml") + assert calls == 20 + + +@pytest.mark.parametrize( + "args", + [ + 'args: # "@scope/server@1.2.3"\n - "@scope/server"', + 'args:\n - "-y" # "decoy@1.2.3"\n - "@scope/server"', + 'args: ["@scope/server"] # "decoy@1.2.3"', + ], +) +def test_rp1_yaml_comments_cannot_supply_a_fake_package_pin(args): + content = "mcpServers:\n fs:\n command: npx\n " + args + "\n" + rp1 = [ + f for f in node(_state(file_cache={"mcp.yaml": content}))["findings"] if f.rule_id == "RP1" + ] + assert len(rp1) == 1 + + +@pytest.mark.parametrize( + "content", + [ + 'servers:\n - command: npx\n - args: ["@scope/server"]\n', + 'servers:\n - args: ["@scope/server"]\n - command: npx\n', + 'mcpServers:\n first:\n command: npx\n second:\n args: ["@scope/server"]\n', + 'mcpServers:\n first:\n args: ["@scope/server"]\n second:\n command: npx\n', + 'mcpServers:\n fs:\n command: npx\n env:\n args: ["@scope/server"]\n', + ], +) +def test_rp1_yaml_does_not_bind_args_from_another_mapping(content): + assert not [ + f for f in node(_state(file_cache={"mcp.yaml": content}))["findings"] if f.rule_id == "RP1" + ] + + +@pytest.mark.parametrize("direction", ["before", "after"]) +@pytest.mark.parametrize("distance", [8, 9]) +def test_rp1_yaml_sibling_search_remains_bounded(direction, distance): + command = " command: npx\n" + args = ' args: ["@scope/server"]\n' + intervening = "".join(f" field{i}: value\n" for i in range(distance - 1)) + pair = args + intervening + command if direction == "before" else command + intervening + args + content = "mcpServers:\n fs:\n" + pair + rp1 = [ + f for f in node(_state(file_cache={"mcp.yaml": content}))["findings"] if f.rule_id == "RP1" + ] + assert len(rp1) == (1 if distance == 8 else 0) + + +def test_rp1_yaml_nested_pinned_args_do_not_hide_sibling_package(): + content = ( + "servers:\r\n - command: npx\r\n env:\r\n" + ' args: ["decoy@1.2.3"]\r\n args: ["@scope/server"]\r\n' + ) + rp1 = [ + f for f in node(_state(file_cache={"mcp.yaml": content}))["findings"] if f.rule_id == "RP1" + ] + assert len(rp1) == 1 + assert rp1[0].start_line == 2 + + +@pytest.mark.parametrize( + ("line", "expected"), + [ + ('"pkg#fragment" # "decoy@1.2.3"', '"pkg#fragment" '), + ("'it''s # quoted' # tail", "'it''s # quoted' "), + ('"escaped\\" # quoted" # tail', '"escaped\\" # quoted" '), + ("pkg#fragment", "pkg#fragment"), + ], +) +def test_yaml_comment_stripping_preserves_quoted_content(line, expected): + assert _strip_yaml_comment(line) == expected + + +def test_rp1_npx_still_matches_flags_on_the_same_line(): + """Common npx flags remain supported after restricting whitespace.""" + result = node(_state(file_cache={"setup.sh": "npx -y @scope/mcp-server\n"})) + rp1 = [finding for finding in result["findings"] if finding.rule_id == "RP1"] + + assert len(rp1) == 1 + assert rp1[0].matched_text == "npx -y @scope/mcp-server" + + def test_rp1_scans_cached_files_without_a_manifest(): """Cache-based RP1 checks remain applicable when manifest parsing failed.""" result = node(_state(file_cache={"setup.sh": "npx @scope/mcp-server\n"}))