From 6726bb196f368f6bf5bab2fbeebfcfa8790b986e Mon Sep 17 00:00:00 2001 From: Zach Wasserman Date: Tue, 21 Oct 2025 11:25:11 -0700 Subject: [PATCH] Add `mcp_listening_servers` table (#34286) **Related issue:** Resolves #34330 # Checklist for submitter If some of the following don't apply, delete the relevant line. - [x] Changes file added for user-visible changes in `changes/`, `orbit/changes/` or `ee/fleetd-chrome/changes`. See [Changes files](https://github.com/fleetdm/fleet/blob/main/docs/Contributing/guides/committing-changes.md#changes-files) for more information. - [x] Input data is properly validated, `SELECT *` is avoided, SQL injection is prevented (using placeholders for values in statements) ## Testing - [x] Added/updated automated tests - [x] QA'd all new/changed functionality manually (so far just macOS) ## fleetd/orbit/Fleet Desktop - [x] Verified compatibility with the latest released version of Fleet (see [Must rule](https://github.com/fleetdm/fleet/blob/main/docs/Contributing/workflows/fleetd-development-and-release-strategy.md)) - [x] Verified that fleetd runs on macOS, Linux and Windows ## Summary by CodeRabbit * **New Features** * Added a built-in mcp_listening_servers table to discover MCP servers by inspecting listening ports and probing endpoints; returns process info, server metadata, capabilities, tools, prompts, and resources (supports macOS, Windows, Linux). * **Tests** * Added comprehensive unit tests covering detection, IPv6 handling, SSE responses, and session lifecycle. --- orbit/changes/34286-mcp-listening-servers | 1 + orbit/cmd/fleetd_tables/fleetd_tables.go | 2 +- orbit/pkg/installer/installer.go | 1 - orbit/pkg/table/extension.go | 14 +- orbit/pkg/table/extension_stub.go | 2 +- orbit/pkg/table/extension_windows.go | 1 + .../mcp_listening_servers.go | 449 ++++++++++++++++++ .../mcp_listening_servers_test.go | 427 +++++++++++++++++ schema/osquery_fleet_schema.json | 106 +++++ schema/tables/mcp_listening_servers.yml | 76 +++ 10 files changed, 1073 insertions(+), 6 deletions(-) create mode 100644 orbit/changes/34286-mcp-listening-servers create mode 100644 orbit/pkg/table/mcp_listening_servers/mcp_listening_servers.go create mode 100644 orbit/pkg/table/mcp_listening_servers/mcp_listening_servers_test.go create mode 100644 schema/tables/mcp_listening_servers.yml diff --git a/orbit/changes/34286-mcp-listening-servers b/orbit/changes/34286-mcp-listening-servers new file mode 100644 index 0000000000..1298ec6093 --- /dev/null +++ b/orbit/changes/34286-mcp-listening-servers @@ -0,0 +1 @@ +* Add `mcp_listening_servers` table to find MCP servers listening over HTTP. \ No newline at end of file diff --git a/orbit/cmd/fleetd_tables/fleetd_tables.go b/orbit/cmd/fleetd_tables/fleetd_tables.go index 5c2deda5e2..232c180803 100644 --- a/orbit/cmd/fleetd_tables/fleetd_tables.go +++ b/orbit/cmd/fleetd_tables/fleetd_tables.go @@ -50,10 +50,10 @@ func main() { log.Fatalln(err) } - plugins := orbittable.OrbitDefaultTables() opts := orbittable.PluginOpts{ Socket: *socket, } + plugins := orbittable.OrbitDefaultTables(opts) platformTables, err := orbittable.PlatformTables(opts) if err != nil { log.Fatalln(err) diff --git a/orbit/pkg/installer/installer.go b/orbit/pkg/installer/installer.go index 78091fee51..a5ee17b819 100644 --- a/orbit/pkg/installer/installer.go +++ b/orbit/pkg/installer/installer.go @@ -330,7 +330,6 @@ func (r *Runner) installWithRetry(ctx context.Context, installer *fleet.Software // attemptInstall performs a single installation attempt (download + install) func (r *Runner) attemptInstall(ctx context.Context, installer *fleet.SoftwareInstallDetails, payload *fleet.HostSoftwareInstallResultPayload, logger zerolog.Logger) (*fleet.HostSoftwareInstallResultPayload, error) { - tmpDirFn := r.tempDirFn if tmpDirFn == nil { tmpDirFn = os.MkdirTemp diff --git a/orbit/pkg/table/extension.go b/orbit/pkg/table/extension.go index 45f0c71255..7455a6515c 100644 --- a/orbit/pkg/table/extension.go +++ b/orbit/pkg/table/extension.go @@ -12,6 +12,7 @@ import ( "github.com/fleetdm/fleet/v4/orbit/pkg/table/dataflattentable" "github.com/fleetdm/fleet/v4/orbit/pkg/table/firefox_preferences" "github.com/fleetdm/fleet/v4/orbit/pkg/table/fleetd_logs" + "github.com/fleetdm/fleet/v4/orbit/pkg/table/mcp_listening_servers" "github.com/fleetdm/fleet/v4/orbit/pkg/table/sntp_request" "github.com/macadmins/osquery-extension/tables/chromeuserprofiles" "github.com/macadmins/osquery-extension/tables/fileline" @@ -107,9 +108,8 @@ func (r *Runner) Execute() error { } } - plugins := OrbitDefaultTables() - opts := PluginOpts{Socket: r.socket} + plugins := OrbitDefaultTables(opts) platformTables, err := PlatformTables(opts) if err != nil { return fmt.Errorf("populating platform tables: %w", err) @@ -132,7 +132,7 @@ func (r *Runner) Execute() error { return nil } -func OrbitDefaultTables() []osquery.OsqueryPlugin { +func OrbitDefaultTables(opts PluginOpts) []osquery.OsqueryPlugin { plugins := []osquery.OsqueryPlugin{ // MacAdmins extensions. table.NewPlugin("puppet_info", puppet.PuppetInfoColumns(), puppet.PuppetInfoGenerate), @@ -157,6 +157,14 @@ func OrbitDefaultTables() []osquery.OsqueryPlugin { dataflattentable.TablePlugin(log.Logger, dataflattentable.XmlType), // table name is "parse_xml" dataflattentable.TablePlugin(log.Logger, dataflattentable.IniType), // table name is "parse_ini" + // mcp_listening_servers: lists running processes from core processes table via osquery client + table.NewPlugin( + "mcp_listening_servers", + mcp_listening_servers.Columns(), + func(ctx context.Context, queryContext table.QueryContext) ([]map[string]string, error) { + return mcp_listening_servers.Generate(ctx, queryContext, opts.Socket) + }, + ), } return plugins } diff --git a/orbit/pkg/table/extension_stub.go b/orbit/pkg/table/extension_stub.go index 37d71c8284..6654d53d1f 100644 --- a/orbit/pkg/table/extension_stub.go +++ b/orbit/pkg/table/extension_stub.go @@ -7,4 +7,4 @@ package table import "github.com/osquery/osquery-go" -func PlatformTables(_ PluginOpts) []osquery.OsqueryPlugin { return nil } +func PlatformTables(_ PluginOpts) ([]osquery.OsqueryPlugin, error) { return nil, nil } diff --git a/orbit/pkg/table/extension_windows.go b/orbit/pkg/table/extension_windows.go index 87dba1a2c7..5539fd26d1 100644 --- a/orbit/pkg/table/extension_windows.go +++ b/orbit/pkg/table/extension_windows.go @@ -4,6 +4,7 @@ package table import ( "fmt" + "github.com/fleetdm/fleet/v4/orbit/pkg/table/bitlocker_key_protectors" cisaudit "github.com/fleetdm/fleet/v4/orbit/pkg/table/cis_audit" mdmbridge "github.com/fleetdm/fleet/v4/orbit/pkg/table/mdm" diff --git a/orbit/pkg/table/mcp_listening_servers/mcp_listening_servers.go b/orbit/pkg/table/mcp_listening_servers/mcp_listening_servers.go new file mode 100644 index 0000000000..8ef6c18d88 --- /dev/null +++ b/orbit/pkg/table/mcp_listening_servers/mcp_listening_servers.go @@ -0,0 +1,449 @@ +package mcp_listening_servers + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net" + "net/http" + "time" + + "github.com/fleetdm/fleet/v4/orbit/pkg/build" + "github.com/fleetdm/fleet/v4/pkg/fleethttp" + osqclient "github.com/osquery/osquery-go" + "github.com/osquery/osquery-go/plugin/table" + "github.com/rs/zerolog/log" +) + +// Columns defines the schema for the mcp_listening_servers table. +func Columns() []table.ColumnDefinition { + return []table.ColumnDefinition{ + table.BigIntColumn("pid"), + table.TextColumn("name"), + table.TextColumn("cmdline"), + table.IntegerColumn("port"), + table.TextColumn("address"), + table.TextColumn("protocol_version"), + table.TextColumn("server_name"), + table.TextColumn("server_title"), + table.TextColumn("server_version"), + table.IntegerColumn("has_logging"), + table.IntegerColumn("has_completions"), + table.TextColumn("instructions"), + table.TextColumn("tools"), // JSON array of tool names + table.TextColumn("prompts"), // JSON array of prompt names + table.TextColumn("resources"), // JSON array of resource URIs + } +} + +// Generate connects to the running osqueryd over the provided socket and queries +// the listening_ports and processes tables to find processes with listening ports, +// then checks each port to see if an MCP server is responding. +func Generate(ctx context.Context, queryContext table.QueryContext, socket string) ([]map[string]string, error) { + scanner := newMCPScanner() + return generateWithScanner(ctx, queryContext, socket, scanner) +} + +// generateWithScanner is an internal helper that allows dependency injection for testing. +func generateWithScanner(ctx context.Context, queryContext table.QueryContext, socket string, scanner *mcpScanner) ([]map[string]string, error) { + // Ensure we don't hang forever if osquery is unresponsive. + ctx, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + + // Open an osquery client using the extension socket. + c, err := scanner.newClient(socket, 2*time.Second) + if err != nil { + return nil, fmt.Errorf("open osquery client: %w", err) + } + defer c.Close() + + // Get the running processes with listening ports + sql := ` + SELECT DISTINCT lp.pid, lp.port, lp.address, lp.family, p.name, p.cmdline + FROM listening_ports lp JOIN processes p ON lp.pid = p.pid + ` + + rows, err := c.QueryRowsContext(ctx, sql) + if err != nil { + return nil, err + } + + // For each row, check if there's an MCP server listening on that port + results := make([]map[string]string, 0, len(rows)) + for _, row := range rows { + port, ok := row["port"] + if !ok || port == "" { + continue + } + + address, ok := row["address"] + if !ok { + address = "127.0.0.1" + } + + // Check family and log a warning if it's not a known address family + if family, ok := row["family"]; ok { + if family != afUnix && family != afInet && family != afInet6 { + log.Warn().Str("family", family).Str("port", port).Str("pid", row["pid"]).Msg("unexpected family value") + } + } + + // Check if MCP server is active on this port + mcpInfo := scanner.checkMCPServer(ctx, address, port) + + // Only include rows where an MCP server is actually responding + if mcpInfo == nil { + continue + } + + // Convert lists to JSON + toolsJSON, _ := json.Marshal(mcpInfo.Tools) + promptsJSON, _ := json.Marshal(mcpInfo.Prompts) + resourcesJSON, _ := json.Marshal(mcpInfo.Resources) + + // Convert boolean to integer string (0 or 1) + hasLogging := "0" + if mcpInfo.HasLogging { + hasLogging = "1" + } + hasCompletions := "0" + if mcpInfo.HasCompletions { + hasCompletions = "1" + } + + // Create result row with all required columns + result := map[string]string{ + "pid": row["pid"], + "name": row["name"], + "cmdline": row["cmdline"], + "port": port, + "address": address, + "protocol_version": mcpInfo.ProtocolVersion, + "server_name": mcpInfo.ServerName, + "server_title": mcpInfo.ServerTitle, + "server_version": mcpInfo.ServerVersion, + "has_logging": hasLogging, + "has_completions": hasCompletions, + "instructions": mcpInfo.Instructions, + "tools": string(toolsJSON), + "prompts": string(promptsJSON), + "resources": string(resourcesJSON), + } + results = append(results, result) + } + + return results, nil +} + +const ( + // Socket address family constants (as strings for osquery comparison) + afUnix = "1" // AF_UNIX - Unix domain sockets + afInet = "2" // AF_INET - IPv4 + afInet6 = "10" // AF_INET6 - IPv6 +) + +// osqueryClient abstracts the osquery client for ease of testing. +type osqueryClient interface { + QueryRowContext(ctx context.Context, sql string) (map[string]string, error) + QueryRowsContext(ctx context.Context, sql string) ([]map[string]string, error) + Close() +} + +// mcpScanner holds the dependencies needed for scanning MCP servers. +type mcpScanner struct { + httpClient *http.Client + newClient func(socket string, timeout time.Duration) (osqueryClient, error) +} + +// newMCPScanner creates a new MCP scanner with default dependencies. +func newMCPScanner() *mcpScanner { + return &mcpScanner{ + httpClient: fleethttp.NewClient(fleethttp.WithTimeout(2 * time.Second)), + newClient: func(socket string, timeout time.Duration) (osqueryClient, error) { + return osqclient.NewClient(socket, timeout) + }, + } +} + +// mcpTool represents a tool with its metadata +type mcpTool struct { + Name string `json:"name"` + Description string `json:"description,omitempty"` +} + +// mcpPrompt represents a prompt with its metadata +type mcpPrompt struct { + Name string `json:"name"` + Description string `json:"description,omitempty"` +} + +// mcpResource represents a resource with its metadata +type mcpResource struct { + URI string `json:"uri"` + Name string `json:"name,omitempty"` + Description string `json:"description,omitempty"` +} + +// mcpServerInfo holds information about an MCP server +type mcpServerInfo struct { + ProtocolVersion string + ServerName string + ServerTitle string + ServerVersion string + HasPrompts bool + HasResources bool + HasTools bool + HasLogging bool + HasCompletions bool + Instructions string + Tools []mcpTool // List of tools with descriptions + Prompts []mcpPrompt // List of prompts with descriptions + Resources []mcpResource // List of resources with descriptions +} + +// mcpResponse represents the JSON-RPC response from an MCP server +type mcpResponse struct { + JSONRPC string `json:"jsonrpc"` + ID int `json:"id"` + Result struct { + ProtocolVersion string `json:"protocolVersion"` + Capabilities struct { + Prompts *json.RawMessage `json:"prompts"` + Resources *json.RawMessage `json:"resources"` + Tools *json.RawMessage `json:"tools"` + Logging *json.RawMessage `json:"logging"` + Completions *json.RawMessage `json:"completions"` + } `json:"capabilities"` + ServerInfo struct { + Name string `json:"name"` + Title string `json:"title"` + Version string `json:"version"` + } `json:"serverInfo"` + Instructions string `json:"instructions"` + } `json:"result"` +} + +// mcpListResponse represents responses to tools/list, prompts/list, resources/list +type mcpListResponse struct { + JSONRPC string `json:"jsonrpc"` + ID int `json:"id"` + Result struct { + Tools []mcpTool `json:"tools"` + Prompts []mcpPrompt `json:"prompts"` + Resources []mcpResource `json:"resources"` + } `json:"result"` + Error *struct { + Code int `json:"code"` + Message string `json:"message"` + } `json:"error"` +} + +// terminateSession sends an HTTP DELETE to the MCP endpoint to explicitly terminate the session. +// Per the MCP spec, the server MAY respond with 405 Method Not Allowed if it doesn't support +// explicit session termination, which is acceptable. +func (s *mcpScanner) terminateSession(ctx context.Context, url, sessionID string) { + if sessionID == "" { + return + } + + req, err := http.NewRequestWithContext(ctx, "DELETE", url, nil) + if err != nil { + return + } + + req.Header.Set("Mcp-Session-Id", sessionID) + + resp, err := s.httpClient.Do(req) + if err != nil { + // Best-effort cleanup, don't log errors + return + } + resp.Body.Close() +} + +// makeMCPRequest makes an MCP JSON-RPC request and returns the response body and session ID +func (s *mcpScanner) makeMCPRequest(ctx context.Context, url, method, sessionID string, params interface{}) ([]byte, string, error) { + reqBody := map[string]interface{}{ + "jsonrpc": "2.0", + "id": 1, + "method": method, + } + if params != nil { + reqBody["params"] = params + } + + bodyJSON, err := json.Marshal(reqBody) + if err != nil { + return nil, "", err + } + + req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewBuffer(bodyJSON)) + if err != nil { + return nil, "", err + } + + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "application/json,text/event-stream") + if sessionID != "" { + req.Header.Set("Mcp-Session-Id", sessionID) + } + + resp, err := s.httpClient.Do(req) + if err != nil { + return nil, "", err + } + defer resp.Body.Close() + + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return nil, "", fmt.Errorf("HTTP %d", resp.StatusCode) + } + + // Extract session ID from response header + responseSessionID := resp.Header.Get("Mcp-Session-Id") + + bodyBytes, err := io.ReadAll(resp.Body) + if err != nil { + return nil, "", err + } + + // Handle SSE format (Server-Sent Events) - responses may contain lines like: + // event: message + // id: ... + // data: {...} + jsonData := bodyBytes + + // Check if this looks like SSE format (contains "data: " on a line) + if bytes.Contains(bodyBytes, []byte("data: ")) { + // Extract JSON from SSE format + lines := bytes.Split(bodyBytes, []byte("\n")) + for _, line := range lines { + if bytes.HasPrefix(line, []byte("data: ")) { + jsonData = bytes.TrimPrefix(line, []byte("data: ")) + break + } + } + } + + return jsonData, responseSessionID, nil +} + +// checkMCPServer attempts to connect to an MCP server at the given address and port. +// It returns information about the MCP server if it responds successfully, or nil otherwise. +func (s *mcpScanner) checkMCPServer(ctx context.Context, address, port string) *mcpServerInfo { + // Build the URL - handle localhost and IPv6 addresses specially + host := address + + // Check if the address is IPv6 + isIPv6 := false + if ip := net.ParseIP(address); ip != nil { + isIPv6 = ip.To4() == nil // If To4() returns nil, it's IPv6 + } + + // Handle wildcard and loopback addresses + if address == "0.0.0.0" || address == "127.0.0.1" || address == "::" || address == "::1" || address == "" { + host = "localhost" + } else if isIPv6 { + // IPv6 addresses must be wrapped in brackets in URLs + host = "[" + address + "]" + } + + url := fmt.Sprintf("http://%s:%s/mcp", host, port) + + // MCP initialize request + // 2025-06-18 is the latest version of the MCP protocol as of 2025-10-14 + // TODO update this to the latest version of the MCP protocol when it is released + initParams := map[string]interface{}{ + "protocolVersion": "2025-06-18", + "capabilities": map[string]interface{}{ + "tools": map[string]interface{}{}, + "resources": map[string]interface{}{}, + "prompts": map[string]interface{}{}, + "roots": map[string]interface{}{}, + }, + "clientInfo": map[string]interface{}{ + "name": "fleetd", + "version": build.Version, + }, + } + + // Send initialize request without a session ID - server will provide one + jsonData, sessionID, err := s.makeMCPRequest(ctx, url, "initialize", "", initParams) + if err != nil { + return nil + } + + // Terminate the session when we're done, per MCP spec + defer s.terminateSession(ctx, url, sessionID) + + var mcpResp mcpResponse + if err := json.Unmarshal(jsonData, &mcpResp); err != nil { + // If JSON parsing fails, return nil so the row is not included + // (we only want rows where we can successfully identify an MCP server) + return nil + } + + // Build the server info struct + info := &mcpServerInfo{ + ProtocolVersion: mcpResp.Result.ProtocolVersion, + ServerName: mcpResp.Result.ServerInfo.Name, + ServerTitle: mcpResp.Result.ServerInfo.Title, + ServerVersion: mcpResp.Result.ServerInfo.Version, + HasPrompts: mcpResp.Result.Capabilities.Prompts != nil, + HasResources: mcpResp.Result.Capabilities.Resources != nil, + HasTools: mcpResp.Result.Capabilities.Tools != nil, + HasLogging: mcpResp.Result.Capabilities.Logging != nil, + HasCompletions: mcpResp.Result.Capabilities.Completions != nil, + Instructions: mcpResp.Result.Instructions, + // Initialize slices to empty so they marshal to [] instead of null + Tools: []mcpTool{}, + Prompts: []mcpPrompt{}, + Resources: []mcpResource{}, + } + + // Fetch lists of tools, prompts, and resources if capabilities are available + if info.HasTools { + if listData, _, err := s.makeMCPRequest(ctx, url, "tools/list", sessionID, nil); err == nil { + var listResp mcpListResponse + if err := json.Unmarshal(listData, &listResp); err == nil { + // Check if the response contains an error + if listResp.Error != nil { + log.Warn().Int("code", listResp.Error.Code).Str("message", listResp.Error.Message).Str("port", port).Msg("tools/list returned error") + } else { + info.Tools = append(info.Tools, listResp.Result.Tools...) + } + } + } + } + + if info.HasPrompts { + if listData, _, err := s.makeMCPRequest(ctx, url, "prompts/list", sessionID, nil); err == nil { + var listResp mcpListResponse + if err := json.Unmarshal(listData, &listResp); err == nil { + // Check if the response contains an error + if listResp.Error != nil { + log.Warn().Int("code", listResp.Error.Code).Str("message", listResp.Error.Message).Str("port", port).Msg("prompts/list returned error") + } else { + info.Prompts = append(info.Prompts, listResp.Result.Prompts...) + } + } + } + } + + if info.HasResources { + if listData, _, err := s.makeMCPRequest(ctx, url, "resources/list", sessionID, nil); err == nil { + var listResp mcpListResponse + if err := json.Unmarshal(listData, &listResp); err == nil { + // Check if the response contains an error + if listResp.Error != nil { + log.Warn().Int("code", listResp.Error.Code).Str("message", listResp.Error.Message).Str("port", port).Msg("resources/list returned error") + } else { + info.Resources = append(info.Resources, listResp.Result.Resources...) + } + } + } + } + + return info +} diff --git a/orbit/pkg/table/mcp_listening_servers/mcp_listening_servers_test.go b/orbit/pkg/table/mcp_listening_servers/mcp_listening_servers_test.go new file mode 100644 index 0000000000..7e44182cd3 --- /dev/null +++ b/orbit/pkg/table/mcp_listening_servers/mcp_listening_servers_test.go @@ -0,0 +1,427 @@ +package mcp_listening_servers + +import ( + "bytes" + "context" + "encoding/json" + "io" + "net/http" + "testing" + "time" + + "github.com/fleetdm/fleet/v4/pkg/fleethttp" + "github.com/osquery/osquery-go/plugin/table" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const mockSessionID = "test-session-id" + +type mockClient struct { + row map[string]string + rows []map[string]string + err error +} + +func (m *mockClient) QueryRowContext(ctx context.Context, sql string) (map[string]string, error) { + return m.row, m.err +} + +func (m *mockClient) QueryRowsContext(ctx context.Context, sql string) ([]map[string]string, error) { + return m.rows, m.err +} +func (m *mockClient) Close() {} + +func TestGenerate_WithMCPServerActive(t *testing.T) { + // Create a scanner with mock dependencies + scanner := &mcpScanner{ + newClient: func(socket string, timeout time.Duration) (osqueryClient, error) { + return &mockClient{rows: []map[string]string{ + {"pid": "1234", "port": "3001", "address": "127.0.0.1", "name": "node", "cmdline": "node mcp-server.js"}, + }}, nil + }, + httpClient: fleethttp.NewClient(fleethttp.WithTimeout(2 * time.Second)), + } + + scanner.httpClient.Transport = &mockTransport{ + t: t, + responses: map[string]string{ + "initialize": `{ + "jsonrpc": "2.0", + "id": 1, + "result": { + "protocolVersion": "2025-03-26", + "capabilities": { + "prompts": {}, + "resources": {"subscribe": true}, + "tools": {}, + "logging": {}, + "completions": {} + }, + "serverInfo": { + "name": "example-servers/everything", + "title": "Everything Example Server", + "version": "1.0.0" + }, + "instructions": "Testing and demonstration server for MCP protocol features." + } + }`, + "tools/list": `{ + "jsonrpc": "2.0", + "id": 1, + "result": { + "tools": [ + {"name": "get_weather", "description": "Get weather for a location"}, + {"name": "search_web", "description": "Search the web"} + ] + } + }`, + "prompts/list": `{ + "jsonrpc": "2.0", + "id": 1, + "result": { + "prompts": [ + {"name": "code_review", "description": "Review code for quality"}, + {"name": "summarize", "description": "Summarize content"} + ] + } + }`, + "resources/list": `{ + "jsonrpc": "2.0", + "id": 1, + "result": { + "resources": [ + {"uri": "file:///data/doc1.txt", "name": "Document 1", "description": "First document"}, + {"uri": "file:///data/doc2.txt", "name": "Document 2", "description": "Second document"} + ] + } + }`, + }, + } + + qc := table.QueryContext{Constraints: map[string]table.ConstraintList{}} + + rows, err := generateWithScanner(context.Background(), qc, "/tmp/osq", scanner) + require.NoError(t, err) + require.Len(t, rows, 1) + + assert.Equal(t, "3001", rows[0]["port"]) + assert.Equal(t, "2025-03-26", rows[0]["protocol_version"]) + assert.Equal(t, "example-servers/everything", rows[0]["server_name"]) + assert.Equal(t, "Everything Example Server", rows[0]["server_title"]) + assert.Equal(t, "1.0.0", rows[0]["server_version"]) + assert.Equal(t, "1", rows[0]["has_logging"]) + assert.Equal(t, "1", rows[0]["has_completions"]) + assert.Equal(t, "Testing and demonstration server for MCP protocol features.", rows[0]["instructions"]) + assert.Equal(t, `[{"name":"get_weather","description":"Get weather for a location"},{"name":"search_web","description":"Search the web"}]`, rows[0]["tools"]) + assert.Equal(t, `[{"name":"code_review","description":"Review code for quality"},{"name":"summarize","description":"Summarize content"}]`, rows[0]["prompts"]) + assert.Equal(t, `[{"uri":"file:///data/doc1.txt","name":"Document 1","description":"First document"},{"uri":"file:///data/doc2.txt","name":"Document 2","description":"Second document"}]`, rows[0]["resources"]) +} + +func TestGenerate_WithMCPServerInactive(t *testing.T) { + // Create a scanner with mock dependencies + scanner := &mcpScanner{ + newClient: func(socket string, timeout time.Duration) (osqueryClient, error) { + return &mockClient{rows: []map[string]string{ + {"pid": "5678", "port": "8080", "address": "0.0.0.0", "name": "nginx", "cmdline": "nginx"}, + }}, nil + }, + httpClient: fleethttp.NewClient(fleethttp.WithTimeout(2 * time.Second)), + } + + scanner.httpClient.Transport = &mockTransport{ + t: t, + err: http.ErrServerClosed, + } + + qc := table.QueryContext{Constraints: map[string]table.ConstraintList{}} + + rows, err := generateWithScanner(context.Background(), qc, "/tmp/osq", scanner) + require.NoError(t, err) + // Should return 0 rows since no MCP server is active + assert.Empty(t, rows) +} + +func TestGenerate_MultipleActiveServers(t *testing.T) { + // Create a scanner with mock dependencies + scanner := &mcpScanner{ + newClient: func(socket string, timeout time.Duration) (osqueryClient, error) { + return &mockClient{rows: []map[string]string{ + {"pid": "1234", "port": "3001", "address": "127.0.0.1", "name": "node", "cmdline": "node mcp1.js"}, + {"pid": "5678", "port": "3002", "address": "127.0.0.1", "name": "node", "cmdline": "node mcp2.js"}, + }}, nil + }, + httpClient: fleethttp.NewClient(fleethttp.WithTimeout(2 * time.Second)), + } + + scanner.httpClient.Transport = &mockTransport{ + t: t, + responses: map[string]string{ + "initialize": `{ + "jsonrpc": "2.0", + "id": 1, + "result": { + "protocolVersion": "2025-03-26", + "capabilities": { + "prompts": {}, + "resources": {"subscribe": true}, + "tools": {}, + "logging": {}, + "completions": {} + }, + "serverInfo": { + "name": "example-servers/everything", + "title": "Everything Example Server", + "version": "1.0.0" + }, + "instructions": "Testing and demonstration server for MCP protocol features." + } + }`, + "tools/list": `{"jsonrpc":"2.0","id":1,"result":{"tools":[]}}`, + "prompts/list": `{"jsonrpc":"2.0","id":1,"result":{"prompts":[]}}`, + "resources/list": `{"jsonrpc":"2.0","id":1,"result":{"resources":[]}}`, + }, + statusCode: 200, + } + + qc := table.QueryContext{Constraints: map[string]table.ConstraintList{}} + + rows, err := generateWithScanner(context.Background(), qc, "/tmp/osq", scanner) + require.NoError(t, err) + assert.Len(t, rows, 2) +} + +func TestGenerate_WithSSEResponse(t *testing.T) { + // Create a scanner with mock dependencies + scanner := &mcpScanner{ + newClient: func(socket string, timeout time.Duration) (osqueryClient, error) { + return &mockClient{rows: []map[string]string{ + {"pid": "1234", "port": "3001", "address": "127.0.0.1", "name": "node", "cmdline": "node mcp-server.js"}, + }}, nil + }, + httpClient: fleethttp.NewClient(fleethttp.WithTimeout(2 * time.Second)), + } + + // Full SSE format with event, id, and data lines + scanner.httpClient.Transport = &mockTransport{ + t: t, + responses: map[string]string{ + "initialize": `event: message +id: 4b74868e-307e-416c-951d-6f305856cb43_1760466849580_vmzdy66v +data: {"result":{"protocolVersion":"2025-03-26","capabilities":{"prompts":{},"resources":{"subscribe":true},"tools":{},"logging":{},"completions":{}},"serverInfo":{"name":"example-servers/everything","title":"Everything Example Server","version":"1.0.0"},"instructions":"Testing and demonstration server for MCP protocol features."},"jsonrpc":"2.0","id":1} +`, + "tools/list": `{"jsonrpc":"2.0","id":1,"result":{"tools":[{"name":"test_tool","description":"A test tool"}]}}`, + "prompts/list": `{"jsonrpc":"2.0","id":1,"result":{"prompts":[{"name":"test_prompt","description":"A test prompt"}]}}`, + "resources/list": `{"jsonrpc":"2.0","id":1,"result":{"resources":[{"uri":"test://resource","name":"Test Resource","description":"A test resource"}]}}`, + }, + statusCode: 200, + } + + qc := table.QueryContext{Constraints: map[string]table.ConstraintList{}} + + rows, err := generateWithScanner(context.Background(), qc, "/tmp/osq", scanner) + require.NoError(t, err) + require.Len(t, rows, 1) + + assert.Equal(t, "2025-03-26", rows[0]["protocol_version"]) + assert.Equal(t, "example-servers/everything", rows[0]["server_name"]) +} + +func TestGenerate_WithIPv6Address(t *testing.T) { + // Create a scanner with mock dependencies + scanner := &mcpScanner{ + newClient: func(socket string, timeout time.Duration) (osqueryClient, error) { + return &mockClient{rows: []map[string]string{ + {"pid": "1234", "port": "3001", "address": "::1", "family": afInet6, "name": "node", "cmdline": "node mcp-server.js"}, + {"pid": "5678", "port": "3002", "address": "2001:db8::1", "family": afInet6, "name": "node", "cmdline": "node mcp-server2.js"}, + {"pid": "9999", "port": "3003", "address": "::", "family": afInet6, "name": "node", "cmdline": "node mcp-server3.js"}, + }}, nil + }, + httpClient: fleethttp.NewClient(fleethttp.WithTimeout(2 * time.Second)), + } + + // Track which URLs were actually requested to verify IPv6 bracket handling + requestedURLs := []string{} + scanner.httpClient.Transport = &mockTransport{ + t: t, + requestedURLs: &requestedURLs, + responses: map[string]string{ + "initialize": `{ + "jsonrpc": "2.0", + "id": 1, + "result": { + "protocolVersion": "2025-03-26", + "capabilities": { + "tools": {} + }, + "serverInfo": { + "name": "test-server", + "title": "Test Server", + "version": "1.0.0" + } + } + }`, + "tools/list": `{"jsonrpc":"2.0","id":1,"result":{"tools":[]}}`, + }, + } + + qc := table.QueryContext{Constraints: map[string]table.ConstraintList{}} + + rows, err := generateWithScanner(context.Background(), qc, "/tmp/osq", scanner) + require.NoError(t, err) + require.Len(t, rows, 3) + + // Verify that IPv6 loopback (::1) was converted to localhost + assert.Contains(t, requestedURLs, "http://localhost:3001/mcp") + + // Verify that IPv6 wildcard (::) was converted to localhost + assert.Contains(t, requestedURLs, "http://localhost:3003/mcp") + + // Verify that regular IPv6 addresses are wrapped in brackets + assert.Contains(t, requestedURLs, "http://[2001:db8::1]:3002/mcp") + + // All should return the same server info + assert.Equal(t, "test-server", rows[0]["server_name"]) + assert.Equal(t, "test-server", rows[1]["server_name"]) + assert.Equal(t, "test-server", rows[2]["server_name"]) +} + +func TestSessionTermination(t *testing.T) { + tests := []struct { + name string + deleteStatusCode int + }{ + { + name: "server supports termination (200 OK)", + deleteStatusCode: http.StatusOK, + }, + { + name: "server doesn't support termination (405 Method Not Allowed)", + deleteStatusCode: http.StatusMethodNotAllowed, + }, + { + name: "server returns 204 No Content", + deleteStatusCode: http.StatusNoContent, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + deleteRequestReceived := false + + scanner := &mcpScanner{ + newClient: func(socket string, timeout time.Duration) (osqueryClient, error) { + return &mockClient{rows: []map[string]string{ + {"pid": "1234", "port": "3001", "address": "127.0.0.1", "name": "node", "cmdline": "node mcp-server.js"}, + }}, nil + }, + httpClient: fleethttp.NewClient(fleethttp.WithTimeout(2 * time.Second)), + } + + scanner.httpClient.Transport = &mockTransport{ + t: t, + deleteRequestReceived: &deleteRequestReceived, + deleteStatusCode: tt.deleteStatusCode, + responses: map[string]string{ + "initialize": `{ + "jsonrpc": "2.0", + "id": 1, + "result": { + "protocolVersion": "2025-03-26", + "capabilities": {}, + "serverInfo": { + "name": "test-server", + "title": "Test Server", + "version": "1.0.0" + } + } + }`, + }, + } + + qc := table.QueryContext{Constraints: map[string]table.ConstraintList{}} + + rows, err := generateWithScanner(context.Background(), qc, "/tmp/osq", scanner) + require.NoError(t, err) + require.Len(t, rows, 1) + + // Verify DELETE request was sent + assert.True(t, deleteRequestReceived, "DELETE request should have been sent for session termination") + }) + } +} + +// mockTransport is a unified HTTP transport for testing MCP servers +type mockTransport struct { + t *testing.T + responses map[string]string // method -> response for POST requests + statusCode int // status code for POST requests (default 200) + err error // error to return instead of response + requestedURLs *[]string // if non-nil, track requested URLs + deleteRequestReceived *bool // if non-nil, track DELETE requests + deleteStatusCode int // status code for DELETE requests (default 200) +} + +func (m *mockTransport) RoundTrip(req *http.Request) (*http.Response, error) { + if m.err != nil { + return nil, m.err + } + + // Track the requested URL if tracking is enabled + if m.requestedURLs != nil { + *m.requestedURLs = append(*m.requestedURLs, req.URL.String()) + } + + // Handle DELETE requests for session termination + if req.Method == "DELETE" { + if m.deleteRequestReceived != nil { + *m.deleteRequestReceived = true + } + // Verify that the session ID header is correct + sessionID := req.Header.Get("Mcp-Session-Id") + assert.Equal(m.t, mockSessionID, sessionID, "DELETE request should have correct Mcp-Session-Id header") + deleteStatus := http.StatusOK + if m.deleteStatusCode != 0 { + deleteStatus = m.deleteStatusCode + } + return &http.Response{ + StatusCode: deleteStatus, + Body: io.NopCloser(bytes.NewBufferString("")), + Header: http.Header{}, + }, nil + } + + // Parse the request to determine which method is being called + var reqBody map[string]interface{} + bodyBytes, _ := io.ReadAll(req.Body) + _ = json.Unmarshal(bodyBytes, &reqBody) + + method, _ := reqBody["method"].(string) + + // Verify session ID header requirements: + // - initialize request should NOT have a session ID (it's the first request) + // - all other requests (tools/list, prompts/list, resources/list) MUST have a session ID + sessionID := req.Header.Get("Mcp-Session-Id") + if method == "initialize" { + assert.Equal(m.t, "", sessionID, "initialize request should not have Mcp-Session-Id header") + } else if method != "" { + // All non-initialize methods should have the correct session ID + assert.Equal(m.t, mockSessionID, sessionID, "%s request should have correct Mcp-Session-Id header", method) + } + + responseBody := m.responses[method] + if responseBody == "" { + responseBody = `{"jsonrpc":"2.0","id":1,"result":{}}` + } + + postStatus := http.StatusOK + if m.statusCode != 0 { + postStatus = m.statusCode + } + + return &http.Response{ + StatusCode: postStatus, + Body: io.NopCloser(bytes.NewBufferString(responseBody)), + Header: http.Header{"Mcp-Session-Id": []string{mockSessionID}}, + }, nil +} diff --git a/schema/osquery_fleet_schema.json b/schema/osquery_fleet_schema.json index 127b3db486..40f5f6b4f1 100644 --- a/schema/osquery_fleet_schema.json +++ b/schema/osquery_fleet_schema.json @@ -16369,6 +16369,112 @@ ], "fleetRepoUrl": "https://github.com/fleetdm/fleet/blob/main/schema/tables/managed_policies.yml" }, + { + "name": "mcp_listening_servers", + "notes": "This table is a fleetd table. Fleetd tables are built into Fleet's agent (fleetd).", + "description": "Lists processes with listening ports that are responding as [Model Context Protocol (MCP)](https://modelcontextprotocol.io/) servers via HTTP (Many MCP servers use stdio and will not be detected by this table). This table queries the `listening_ports` and `processes` tables to find listening ports, then probes each port to identify active MCP servers.", + "platforms": [ + "darwin", + "windows", + "linux" + ], + "evented": false, + "examples": "List all MCP servers currently running on the system.\n\n```\nSELECT * FROM mcp_listening_servers;\n```", + "columns": [ + { + "name": "pid", + "description": "The process ID of the process running the MCP server.", + "required": false, + "type": "bigint" + }, + { + "name": "name", + "description": "The name of the process running the MCP server.", + "required": false, + "type": "text" + }, + { + "name": "cmdline", + "description": "The full command line of the process running the MCP server.", + "required": false, + "type": "text" + }, + { + "name": "port", + "description": "The port number on which the MCP server is listening.", + "required": false, + "type": "integer" + }, + { + "name": "address", + "description": "The network address on which the MCP server is listening (e.g., \"127.0.0.1\", \"0.0.0.0\", \"::1\").", + "required": false, + "type": "text" + }, + { + "name": "protocol_version", + "description": "The MCP protocol version supported by the server (e.g., \"2025-06-18\").", + "required": false, + "type": "text" + }, + { + "name": "server_name", + "description": "The name identifier of the MCP server.", + "required": false, + "type": "text" + }, + { + "name": "server_title", + "description": "The human-readable title of the MCP server.", + "required": false, + "type": "text" + }, + { + "name": "server_version", + "description": "The version of the MCP server implementation.", + "required": false, + "type": "text" + }, + { + "name": "has_logging", + "description": "Indicates if the MCP server supports logging capabilities (1 = true, 0 = false).", + "required": false, + "type": "integer" + }, + { + "name": "has_completions", + "description": "Indicates if the MCP server supports completion capabilities (1 = true, 0 = false).", + "required": false, + "type": "integer" + }, + { + "name": "instructions", + "description": "Instructions or description text provided by the MCP server about its purpose and usage.", + "required": false, + "type": "text" + }, + { + "name": "tools", + "description": "JSON array of tools provided by the MCP server, including their names and descriptions.", + "required": false, + "type": "text" + }, + { + "name": "prompts", + "description": "JSON array of prompts provided by the MCP server, including their names and descriptions.", + "required": false, + "type": "text" + }, + { + "name": "resources", + "description": "JSON array of resources provided by the MCP server, including their URIs, names, and descriptions.", + "required": false, + "type": "text" + } + ], + "url": "https://fleetdm.com/tables/mcp_listening_servers", + "fleetRepoUrl": "https://github.com/fleetdm/fleet/blob/main/schema/tables/mcp_listening_servers.yml" + }, { "name": "md_devices", "description": "Software RAID array settings.", diff --git a/schema/tables/mcp_listening_servers.yml b/schema/tables/mcp_listening_servers.yml new file mode 100644 index 0000000000..38becc258b --- /dev/null +++ b/schema/tables/mcp_listening_servers.yml @@ -0,0 +1,76 @@ +name: mcp_listening_servers +notes: This table is a fleetd table. Fleetd tables are built into Fleet's agent (fleetd). +description: Lists processes with listening ports that are responding as [Model Context Protocol (MCP)](https://modelcontextprotocol.io/) servers via HTTP (Many MCP servers use stdio and will not be detected by this table). This table queries the `listening_ports` and `processes` tables to find listening ports, then probes each port to identify active MCP servers. +platforms: + - darwin + - windows + - linux +evented: false +examples: |- + List all MCP servers currently running on the system. + + ``` + SELECT * FROM mcp_listening_servers; + ``` +columns: + - name: pid + description: The process ID of the process running the MCP server. + required: false + type: bigint + - name: name + description: The name of the process running the MCP server. + required: false + type: text + - name: cmdline + description: The full command line of the process running the MCP server. + required: false + type: text + - name: port + description: The port number on which the MCP server is listening. + required: false + type: integer + - name: address + description: The network address on which the MCP server is listening (e.g., "127.0.0.1", "0.0.0.0", "::1"). + required: false + type: text + - name: protocol_version + description: The MCP protocol version supported by the server (e.g., "2025-06-18"). + required: false + type: text + - name: server_name + description: The name identifier of the MCP server. + required: false + type: text + - name: server_title + description: The human-readable title of the MCP server. + required: false + type: text + - name: server_version + description: The version of the MCP server implementation. + required: false + type: text + - name: has_logging + description: Indicates if the MCP server supports logging capabilities (1 = true, 0 = false). + required: false + type: integer + - name: has_completions + description: Indicates if the MCP server supports completion capabilities (1 = true, 0 = false). + required: false + type: integer + - name: instructions + description: Instructions or description text provided by the MCP server about its purpose and usage. + required: false + type: text + - name: tools + description: JSON array of tools provided by the MCP server, including their names and descriptions. + required: false + type: text + - name: prompts + description: JSON array of prompts provided by the MCP server, including their names and descriptions. + required: false + type: text + - name: resources + description: JSON array of resources provided by the MCP server, including their URIs, names, and descriptions. + required: false + type: text +