From 52c7488750cc0635b3a4caabb188e4e4a6c45c69 Mon Sep 17 00:00:00 2001 From: King Star Date: Sat, 20 Jun 2026 23:01:47 +0800 Subject: [PATCH] fix: reduce Python semantic analyzer false positives --- src/mcts/analyzers/behavioral_static.py | 44 +++++++++++++++++++-- src/mcts/analyzers/command_execution.py | 5 --- src/mcts/sast/python/crossfile.py | 21 ++++++---- tests/test_analyzers.py | 21 +++++++++- tests/test_semantic.py | 51 +++++++++++++++++++++++++ 5 files changed, 125 insertions(+), 17 deletions(-) diff --git a/src/mcts/analyzers/behavioral_static.py b/src/mcts/analyzers/behavioral_static.py index c5b5ab3..63e92f2 100644 --- a/src/mcts/analyzers/behavioral_static.py +++ b/src/mcts/analyzers/behavioral_static.py @@ -69,6 +69,37 @@ "shutil": Severity.HIGH, } +_PYTHON_CALL_SINKS = { + "subprocess.run": ("subprocess", Severity.HIGH), + "subprocess.call": ("subprocess", Severity.HIGH), + "subprocess.Popen": ("subprocess", Severity.HIGH), + "subprocess.check_output": ("subprocess", Severity.HIGH), + "os.system": ("os.system", Severity.HIGH), + "os.popen": ("os.popen", Severity.HIGH), + "eval": ("eval", Severity.CRITICAL), + "exec": ("exec", Severity.CRITICAL), + "open": ("open", Severity.MEDIUM), + "requests.get": ("requests.", Severity.MEDIUM), + "requests.post": ("requests.", Severity.MEDIUM), + "requests.put": ("requests.", Severity.MEDIUM), + "requests.delete": ("requests.", Severity.MEDIUM), + "requests.request": ("requests.", Severity.MEDIUM), + "urllib.request.urlopen": ("urllib", Severity.MEDIUM), + "httpx.get": ("httpx", Severity.HIGH), + "httpx.post": ("httpx", Severity.HIGH), + "httpx.AsyncClient": ("http_client", Severity.HIGH), + "shutil.rmtree": ("shutil.rmtree", Severity.HIGH), + "os.remove": ("os.remove", Severity.HIGH), + "os.unlink": ("os.remove", Severity.HIGH), + "os.rmdir": ("os.remove", Severity.HIGH), + "shutil.copy": ("shutil", Severity.HIGH), + "shutil.move": ("shutil", Severity.HIGH), + "pickle.load": ("pickle", Severity.HIGH), + "pickle.loads": ("pickle", Severity.HIGH), + "socket.socket": ("socket", Severity.MEDIUM), + "Template": ("Template", Severity.HIGH), +} + class BehavioralStaticAnalyzer(BaseAnalyzer): """Detects description/implementation mismatches and taint flows in tool handlers.""" @@ -363,6 +394,8 @@ def _read_handler(path: str, line: int | None) -> str: def _detect_sinks(snippet: str, source_file: str | None) -> dict[str, Severity]: sinks: dict[str, Severity] = {} + if _looks_like_python(snippet, source_file): + return _detect_python_sinks(snippet) if _is_typescript(source_file) or _looks_like_typescript(snippet): for label in detect_typescript_sinks(snippet): sinks[label] = _SINK_MARKERS.get(label, Severity.MEDIUM) @@ -375,8 +408,11 @@ def _detect_sinks(snippet: str, source_file: str | None) -> dict[str, Severity]: for marker, severity in _SINK_MARKERS.items(): if marker in snippet: sinks[marker] = severity - if not _is_python(source_file): - return sinks + return sinks + + +def _detect_python_sinks(snippet: str) -> dict[str, Severity]: + sinks: dict[str, Severity] = {} try: tree = ast.parse(snippet) except SyntaxError: @@ -384,8 +420,8 @@ def _detect_sinks(snippet: str, source_file: str | None) -> dict[str, Severity]: for node in ast.walk(tree): if isinstance(node, ast.Call): name = _call_name(node.func) - if name in ("subprocess.run", "subprocess.Popen", "os.system", "eval", "exec"): - sinks[name] = Severity.HIGH + if name in _PYTHON_CALL_SINKS: + sinks[_PYTHON_CALL_SINKS[name][0]] = _PYTHON_CALL_SINKS[name][1] return sinks diff --git a/src/mcts/analyzers/command_execution.py b/src/mcts/analyzers/command_execution.py index d6dc82f..177c2ec 100644 --- a/src/mcts/analyzers/command_execution.py +++ b/src/mcts/analyzers/command_execution.py @@ -157,11 +157,6 @@ def _findings_from_snippet( def _snippet_matches_call(snippet: str, call: str, *, strict: bool) -> bool: - if strict: - pattern = _SNIPPET_PATTERNS.get(call) - return bool(pattern and pattern.search(snippet)) - if call.replace(".", "") in snippet.replace(".", "") or call in snippet: - return True pattern = _SNIPPET_PATTERNS.get(call) return bool(pattern and pattern.search(snippet)) diff --git a/src/mcts/sast/python/crossfile.py b/src/mcts/sast/python/crossfile.py index cbc2f73..ca5d10b 100644 --- a/src/mcts/sast/python/crossfile.py +++ b/src/mcts/sast/python/crossfile.py @@ -47,17 +47,24 @@ def _local_imports(source: str) -> list[str]: def _resolve_module(base_dir: Path, module_name: str, source_files: dict[str, str]) -> str | None: - candidates = [ - base_dir / f"{module_name}.py", - base_dir / module_name / "__init__.py", - ] + candidates = list(_module_candidates(base_dir, module_name)) for candidate in candidates: key = str(candidate) if key in source_files: return source_files[key] if candidate.exists(): return candidate.read_text(encoding="utf-8") - for path, content in source_files.items(): - if path.endswith(f"/{module_name}.py") or path.endswith(f"/{module_name}/__init__.py"): - return content return None + + +def _module_candidates(base_dir: Path, module_name: str) -> list[Path]: + roots = [base_dir, *base_dir.parents] + candidates: list[Path] = [] + for root in roots: + candidates.extend( + [ + root / f"{module_name}.py", + root / module_name / "__init__.py", + ] + ) + return list(dict.fromkeys(candidates)) diff --git a/tests/test_analyzers.py b/tests/test_analyzers.py index 8571000..4b07a19 100644 --- a/tests/test_analyzers.py +++ b/tests/test_analyzers.py @@ -6,7 +6,7 @@ from mcts.analyzers.data_leakage import DataLeakageAnalyzer from mcts.core.config import ScanConfig from mcts.discovery.static import StaticDiscovery -from mcts.mcp.models import MCPServerInfo +from mcts.mcp.models import MCPServerInfo, MCPTool from mcts.reporting.models import Severity @@ -21,6 +21,25 @@ def test_command_execution_detects_subprocess(example_server_path: Path) -> None assert all(isinstance((f.evidence or {}).get("facts"), list) and f.evidence["facts"] for f in findings) +def test_command_execution_ignores_substrings_without_calls() -> None: + server = MCPServerInfo( + tools=[ + MCPTool( + name="describe", + description="Describe execution status.", + handler_snippet=( + "def describe(name: str) -> str:\n" + " return f'No subprocess or exec call is used while fetching {name}'\n" + ), + ) + ] + ) + + findings = CommandExecutionAnalyzer().analyze(server) + + assert findings == [] + + def test_data_leakage_scans_source_files(example_server_path: Path) -> None: from mcts.core.scanner import Scanner diff --git a/tests/test_semantic.py b/tests/test_semantic.py index ada2533..e12c8b3 100644 --- a/tests/test_semantic.py +++ b/tests/test_semantic.py @@ -48,6 +48,57 @@ def test_behavioral_static_emits_semantic_without_code_sinks() -> None: assert any(f.id.startswith("behavioral-semantic") for f in findings) +def test_behavioral_static_does_not_expand_third_party_mcp_imports_to_local_modules() -> None: + source_file = "src/sibyl_mcp_server.py" + source = ''' +from mcp.server.fastmcp import FastMCP + +mcp = FastMCP("perseus") + +@mcp.tool() +def get_entity(name: str) -> str: + """Safe lookup helper.""" + return f"Error fetching entity {name}" +''' + tool = MCPTool( + name="get_entity", + description="Safe benign lookup helper.", + handler_snippet=source, + source_file=source_file, + ) + server = MCPServerInfo( + tools=[tool], + source_files={ + source_file: source, + "src/perseus/mcp.py": "import subprocess\nsubprocess.run(['echo', 'wrong'])\n", + }, + ) + + findings = BehavioralStaticAnalyzer().analyze(server) + + assert not any("subprocess" in (finding.id + finding.description) for finding in findings) + + +def test_behavioral_static_does_not_treat_fetching_string_as_fetch_sink() -> None: + source = ''' +def get_entity(name: str) -> str: + """Safe lookup helper.""" + return f"Error fetching entity {name}" +''' + tool = MCPTool( + name="get_entity", + description="Safe benign lookup helper.", + handler_snippet=source, + source_file="server.py", + ) + + findings = BehavioralStaticAnalyzer().analyze( + MCPServerInfo(tools=[tool], source_files={"server.py": source}) + ) + + assert not any("fetch" in (finding.id + finding.description) for finding in findings) + + def test_scanner_eval_recall_when_corpus_available() -> None: corpus = _behavioral_eval_corpus() if corpus is None: