From 318bfddff8b32236502d5e793548c15d71b5a09d Mon Sep 17 00:00:00 2001 From: Sammy Tourani Date: Tue, 29 Sep 2026 09:32:38 -0400 Subject: [PATCH] fix(rp1): only accept a version pin attached to the package operand The npx, uvx and pip checks searched the rest of the line for anything that looked like a version pin, so an unrelated argument, a URL with a port, or a neighboring pinned command suppressed the finding for an unpinned package (for example `npx evil-package --label helper@1.2.3` or `npx @scope/mcp-server http://localhost:3000/sse`). Only the text attached to the package token now counts as its pin. Signed-off-by: Sammy Tourani --- .../nodes/analyzers/mcp_rug_pull.py | 16 +++++++++--- tests/test_mcp_rug_pull.py | 26 +++++++++++++++++++ 2 files changed, 39 insertions(+), 3 deletions(-) diff --git a/src/skillspector/nodes/analyzers/mcp_rug_pull.py b/src/skillspector/nodes/analyzers/mcp_rug_pull.py index fb14f6064..31b8f2486 100644 --- a/src/skillspector/nodes/analyzers/mcp_rug_pull.py +++ b/src/skillspector/nodes/analyzers/mcp_rug_pull.py @@ -207,6 +207,16 @@ def _get_parameters_map( # --------------------------------------------------------------------------- +def _operand_has_version_pin(line_remainder: str) -> bool: + """Return whether a version pin is attached to the matched package operand. + + Only the operand's own token counts, so a version on a later argument or on a + neighboring command on the same line does not pin this package. + """ + operand_suffix = re.split(r"\s", line_remainder, maxsplit=1)[0] + return _VERSION_PIN_RE.search(operand_suffix) is not None + + def _check_rp1( manifest: dict, file_cache: dict[str, str], @@ -223,7 +233,7 @@ def _check_rp1( if line_end == -1: line_end = len(content) line_remainder = content[m.end() : min(line_end, m.end() + 256)] - if _VERSION_PIN_RE.search(full_match) or _VERSION_PIN_RE.search(line_remainder): + if _VERSION_PIN_RE.search(full_match) or _operand_has_version_pin(line_remainder): continue line_num = _find_line(content, m.start()) budget.emit( @@ -258,7 +268,7 @@ def _check_rp1( if line_end == -1: line_end = len(content) line_remainder = content[m.end() : min(line_end, m.end() + 256)] - if _VERSION_PIN_RE.search(full_match) or _VERSION_PIN_RE.search(line_remainder): + if _VERSION_PIN_RE.search(full_match) or _operand_has_version_pin(line_remainder): continue line_num = _find_line(content, m.start()) budget.emit( @@ -291,7 +301,7 @@ def _check_rp1( if line_end == -1: line_end = len(content) line_remainder = content[m.end() : min(line_end, m.end() + 256)] - if _VERSION_PIN_RE.search(full_match) or _VERSION_PIN_RE.search(line_remainder): + if _VERSION_PIN_RE.search(full_match) or _operand_has_version_pin(line_remainder): continue pkg = m.group(1) if "mcp" not in pkg.lower(): diff --git a/tests/test_mcp_rug_pull.py b/tests/test_mcp_rug_pull.py index 8f798fba4..244936326 100644 --- a/tests/test_mcp_rug_pull.py +++ b/tests/test_mcp_rug_pull.py @@ -87,6 +87,32 @@ def test_rp1_npx_pinned_no_finding(): assert len(rp1) == 0 +def test_rp1_unrelated_version_pin_does_not_suppress(): + """A pin on another argument or command on the same line does not pin the package.""" + for content, expected in ( + ("npx evil-package --label helper@1.2.3\n", "npx evil-package"), + ("npx @scope/mcp-server http://localhost:3000/sse\n", "npx @scope/mcp-server"), + ("npx @scope/server-a && npx @scope/server-b@1.2.3\n", "npx @scope/server-a"), + ("uvx my-mcp-server --with helper==1.2.3\n", "uvx my-mcp-server"), + ("pip install my-mcp-server other-package==1.2.3\n", "pip install my-mcp-server"), + ): + result = node(_state(file_cache={"setup.sh": content})) + rp1 = [f for f in result["findings"] if f.rule_id == "RP1"] + assert [f.matched_text for f in rp1] == [expected], content + + +def test_rp1_version_pin_attached_to_package_no_finding(): + """A pin attached to the package operand still counts when other arguments follow.""" + for content in ( + "npx -y @scope/mcp-server@1.2.3 --label helper\n", + "npx -p @scope/mcp-server@1.2.3 mcp-server\n", + "uvx my-mcp-server==1.2.3 --host 127.0.0.1:8000\n", + "pip install my-mcp-server[cli]==1.2.3\n", + ): + result = node(_state(file_cache={"setup.sh": content})) + assert not [f for f in result["findings"] if f.rule_id == "RP1"], content + + def test_rp1_uvx_unpinned(): """RP1 detects uvx without ==version.""" result = node(