Fix semantic analyzer false positives for Python sinks

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,17 +408,20 @@ 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:
         return sinks
     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
 
 

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))
 

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))

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
 

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: