Python: Model complex MCP AST handler parameters as remote sources (Complementary to #22703)

docs/codeql/reusables/supported-frameworks.rst

@@ -226,6 +226,8 @@ and the CodeQL library pack ``codeql/python-all`` (`changelog <https://github.co
    Tornado, Web framework
    Twisted, Web framework
    Gradio, Web framework
+   fastmcp, MCP server framework
+   mcp, MCP server framework
    starlette, Asynchronous Server Gateway Interface (ASGI)
    ldap3, Lightweight Directory Access Protocol (LDAP)
    python-ldap, Lightweight Directory Access Protocol (LDAP)

python/ql/lib/change-notes/2026-10-04-mcp-remote-sources.md

@@ -0,0 +1,4 @@
+---
+category: minorAnalysis
+---
+* Added complementary `RemoteFlowSource` modeling for MCP/`fastmcp` handlers that Models-as-Data misses: stacked custom decorators and instance registrations such as `@self.mcp.tool()`.

python/ql/lib/semmle/python/Frameworks.qll

@@ -50,6 +50,7 @@ private import semmle.python.frameworks.Libxml2
 private import semmle.python.frameworks.Lxml
 private import semmle.python.frameworks.Mako
 private import semmle.python.frameworks.MarkupSafe
+private import semmle.python.frameworks.Mcp
 private import semmle.python.frameworks.Multidict
 private import semmle.python.frameworks.Mysql
 private import semmle.python.frameworks.MySQLdb

python/ql/lib/semmle/python/frameworks/Mcp.qll

@@ -0,0 +1,139 @@
+/**
+ * Provides classes modeling security-relevant aspects of the `mcp` and `fastmcp` PyPI packages
+ * (Model Context Protocol server SDKs).
+ *
+ * Baseline `@mcp.tool` / `@mcp.prompt` / `@mcp.resource` handlers on standalone functions are
+ * covered by Models-as-Data. This module only captures programmatic edge cases MaD misses:
+ * stacked/wrapped custom decorators and instance registrations such as `@self.mcp.tool()`.
+ */
+
+private import python
+private import semmle.python.dataflow.new.DataFlow
+private import semmle.python.dataflow.new.RemoteFlowSources
+private import semmle.python.ApiGraphs
+
+/**
+ * Provides models for MCP server handler parameters as remote flow sources.
+ */
+module Mcp {
+  private string registrationMethod() { result in ["tool", "prompt", "resource"] }
+
+  /** Gets API nodes for MCP / FastMCP server classes used to register handlers. */
+  private API::Node mcpServerClass() {
+    result =
+      [
+        API::moduleImport("fastmcp").getMember("FastMCP"),
+        API::moduleImport("fastmcp").getMember("server").getMember("FastMCP"),
+        API::moduleImport("fastmcp").getMember("server").getMember("server").getMember("FastMCP"),
+        API::moduleImport("mcp").getMember("server").getMember("fastmcp").getMember("FastMCP"),
+        API::moduleImport("mcp")
+            .getMember("server")
+            .getMember("fastmcp")
+            .getMember("server")
+            .getMember("FastMCP"),
+        API::moduleImport("mcp").getMember("server").getMember("MCPServer"),
+        API::moduleImport("mcp").getMember("server").getMember("mcpserver").getMember("MCPServer"),
+        API::moduleImport("mcp")
+            .getMember("server")
+            .getMember("mcpserver")
+            .getMember("server")
+            .getMember("MCPServer"),
+      ]
+  }
+
+  /** An MCP `@tool`/`@prompt`/`@resource` decorator expression on `handler`. */
+  private predicate mcpRegistrationDecorator(Function handler, Expr dec) {
+    dec = handler.getADecorator() and
+    exists(API::Node cls | cls = mcpServerClass() |
+      exists(API::CallNode call |
+        call = cls.getAnInstance().getMember(registrationMethod()).getACall() and
+        call.getNode().getNode() = dec
+      )
+      or
+      exists(string name | name = registrationMethod() |
+        dec = cls.getAnInstance().getMember(name).getAValueReachableFromSource().asExpr()
+      )
+    )
+  }
+
+  /**
+   * Holds if `handler` is an MCP registration wrapped by another decorator.
+   * MaD YAML currently drops these stacked/custom project wrappers.
+   */
+  private predicate isWrappedMcpHandler(Function handler) {
+    exists(Expr mcpDec, Expr extra |
+      mcpRegistrationDecorator(handler, mcpDec) and
+      extra = handler.getADecorator() and
+      extra != mcpDec
+    )
+  }
+
+  /**
+   * Holds if `handler` is registered via a nested instance lookup such as `@self.mcp.tool()`.
+   * The callee object is itself an `Attribute`, not a simple name (`mcp_app.tool()`).
+   */
+  private predicate isInstanceLookupHandler(Function handler) {
+    exists(Expr dec | mcpRegistrationDecorator(handler, dec) |
+      exists(Attribute callee |
+        callee = dec.(Call).getFunc()
+        or
+        callee = dec
+      |
+        callee.getObject() instanceof Attribute
+      )
+    )
+  }
+
+  private predicate isMcpHandler(Function handler) {
+    isWrappedMcpHandler(handler)
+    or
+    isInstanceLookupHandler(handler)
+  }
+
+  private predicate isContextAnnotation(Expr ann) {
+    exists(API::Node ctx |
+      ctx in [
+          API::moduleImport("fastmcp").getMember("Context"),
+          API::moduleImport("mcp").getMember("server").getMember("fastmcp").getMember("Context"),
+        ] and
+      ann = ctx.getAValueReachableFromSource().asExpr()
+    )
+  }
+
+  private predicate isMcpContextParameter(Parameter p) {
+    exists(Expr ann | ann = p.getAnnotation() |
+      isContextAnnotation(ann)
+      or
+      exists(Subscript sub | sub = ann | isContextAnnotation(sub.getValue()))
+      or
+      exists(BinaryExpr bin | bin = ann and bin.getOp() instanceof BitOr |
+        isContextAnnotation(bin.getLeft()) or isContextAnnotation(bin.getRight())
+      )
+    )
+  }
+
+  private predicate isHandlerParameter(Function handler, Parameter p) {
+    (
+      p = handler.getAnArg()
+      or
+      p = handler.getAKeywordOnlyArg()
+      or
+      p = handler.getVararg()
+      or
+      p = handler.getKwarg()
+    ) and
+    not (handler.isMethod() and p = handler.getArg(0)) and
+    not isMcpContextParameter(p)
+  }
+
+  private class McpHandlerParameter extends RemoteFlowSource::Range, DataFlow::ParameterNode {
+    McpHandlerParameter() {
+      exists(Function handler |
+        isMcpHandler(handler) and
+        isHandlerParameter(handler, this.getParameter())
+      )
+    }
+
+    override string getSourceType() { result = "MCP server handler parameter" }
+  }
+}

python/ql/test/library-tests/dataflow/remote-flow-sources/RemoteFlowSources.expected

@@ -0,0 +1,3 @@
+| mcp_test.py:61:23:61:25 | ControlFlowNode for cmd | MCP server handler parameter |
+| mcp_test.py:80:29:80:32 | ControlFlowNode for host | MCP server handler parameter |
+| mcp_test.py:86:40:86:46 | ControlFlowNode for payload | MCP server handler parameter |

python/ql/test/library-tests/dataflow/remote-flow-sources/RemoteFlowSources.ql

@@ -0,0 +1,6 @@
+import python
+import semmle.python.dataflow.new.RemoteFlowSources
+
+from RemoteFlowSource src
+where exists(src.getLocation().getFile().getRelativePath())
+select src, src.getSourceType()

python/ql/test/library-tests/dataflow/remote-flow-sources/mcp_test.py

@@ -0,0 +1,87 @@
+from fastmcp import Context, FastMCP
+from fastmcp import tool as standalone_tool
+from mcp.server.fastmcp import FastMCP as McpFastMCP
+from mcp.server import MCPServer
+
+mcp_app = FastMCP("test")
+mcp_sdk = McpFastMCP("test")
+mcp_v2 = MCPServer("test")
+
+
+def other_decorator(fn):
+    return fn
+
+
+@mcp_app.tool()
+def sync_tool(command: str) -> str:
+    return command
+
+
+@mcp_app.tool
+def bare_tool(command: str) -> str:
+    return command
+
+
+@mcp_sdk.prompt()
+def sdk_prompt(topic: str, *, style: str = "short") -> str:
+    return topic + style
+
+
+@mcp_v2.resource("notes://{name}")
+def v2_resource(name: str) -> str:
+    return name
+
+
+async def async_fetch(url: str) -> str:
+    return url
+
+
+mcp_app.add_tool(async_fetch)
+
+
+def added_via_add(url: str) -> str:
+    return url
+
+
+mcp_sdk.add_tool(fn=added_via_add)
+
+
+@standalone_tool
+def module_tool(cmd: str) -> str:
+    return cmd
+
+
+@standalone_tool()
+def module_tool_called(cmd: str) -> str:
+    return cmd
+
+
+@other_decorator
+@mcp_app.tool()
+def stacked_decorator(cmd: str) -> str:
+    return cmd
+
+
+def not_registered(value: str) -> str:
+    return value
+
+
+@mcp_app.tool()
+def with_context(ctx: Context, data: str) -> str:
+    return data
+
+
+class Service:
+    def __init__(self):
+        self.mcp = FastMCP("svc")
+
+    def register(self):
+        @self.mcp.tool()
+        def instance_lookup(host: str) -> str:
+            return host
+
+
+@other_decorator
+@mcp_app.tool()
+def stacked_with_context(ctx: Context, payload: str) -> str:
+    return payload