diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 1bd38513a7..86b5850e80 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -16523,6 +16523,7 @@ dependencies = [ "serde", "serde_json", "sqlx", + "sse-stream", "tokio", "tokio-util", "tracing", diff --git a/backend/tests/fixtures/mcp_resource_authz.sql b/backend/tests/fixtures/mcp_resource_authz.sql index ecf3f01dc3..17fcc553b9 100644 --- a/backend/tests/fixtures/mcp_resource_authz.sql +++ b/backend/tests/fixtures/mcp_resource_authz.sql @@ -23,3 +23,19 @@ VALUES ( '{}', 'test-user' ); + +-- MCP resource whose URL lives in a variable. The host only appears in the +-- variable, so an error naming it proves the reference was resolved. +INSERT INTO variable (workspace_id, path, value, is_secret, description, extra_perms) +VALUES ('test-workspace', 'u/test-user/mcp_url', 'https://mcp-from-var.invalid.windmill.test/key', false, '', '{}'); + +INSERT INTO resource (workspace_id, path, value, description, resource_type, extra_perms, created_by) +VALUES ( + 'test-workspace', + 'u/test-user/var_mcp', + '{"name": "var_mcp", "url": "$var:u/test-user/mcp_url"}', + 'MCP resource with its URL in a variable', + 'mcp', + '{}', + 'test-user' +); diff --git a/backend/tests/mcp_resource_authz.rs b/backend/tests/mcp_resource_authz.rs index 0fa35933f4..6ccf5f4f2d 100644 --- a/backend/tests/mcp_resource_authz.rs +++ b/backend/tests/mcp_resource_authz.rs @@ -23,9 +23,9 @@ use windmill_common::client::AuthedClient; use windmill_test_utils::*; use windmill_worker::{load_mcp_tools, McpResourceConfig}; -fn config() -> Vec { +fn config(resource_path: &str) -> Vec { vec![McpResourceConfig { - resource_path: "$res:f/private/admin_mcp".to_string(), + resource_path: resource_path.to_string(), include_tools: None, exclude_tools: None, }] @@ -34,7 +34,16 @@ fn config() -> Vec { // The Ok variant ((HashMap<_, Arc>, Vec)) does not implement // Debug, so `expect_err` is unavailable — extract the error explicitly. async fn load_err(db: &Pool, client: &AuthedClient, ctx: &str) -> String { - match load_mcp_tools(db, "test-workspace", config(), client).await { + load_err_for(db, client, "$res:f/private/admin_mcp", ctx).await +} + +async fn load_err_for( + db: &Pool, + client: &AuthedClient, + resource_path: &str, + ctx: &str, +) -> String { + match load_mcp_tools(db, "test-workspace", config(resource_path), client).await { Ok(_) => panic!("{ctx}"), Err(e) => e.to_string(), } @@ -100,3 +109,31 @@ async fn test_mcp_resource_not_loaded_without_authorization( Ok(()) } + +/// A `$var:` in the resource's `url` is resolved before connecting, rather than +/// being handed to the client as a literal (and rejected as a relative URL). +#[sqlx::test(fixtures("base", "mcp_resource_authz"))] +async fn test_mcp_resource_url_variable_is_resolved(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + + let server = ApiServer::start(db.clone()).await?; + let client = AuthedClient::new( + format!("http://localhost:{}", server.addr.port()), + "test-workspace".to_string(), + "SECRET_TOKEN".to_string(), + None, + ); + let msg = load_err_for( + &db, + &client, + "$res:u/test-user/var_mcp", + "the variable's URL is non-resolvable, so the connect step must fail", + ) + .await; + assert!( + msg.contains("mcp-from-var.invalid.windmill.test"), + "the URL should come from the variable: {msg}" + ); + + Ok(()) +} diff --git a/backend/windmill-api/src/mcp_tools.rs b/backend/windmill-api/src/mcp_tools.rs index 1e07f374de..a9049951a3 100644 --- a/backend/windmill-api/src/mcp_tools.rs +++ b/backend/windmill-api/src/mcp_tools.rs @@ -125,8 +125,10 @@ async fn connect_mcp_client( None }; - // Header secrets go through the same permissioned path as the token. - for value in mcp_resource.headers.iter_mut().flatten().map(|(_, v)| v) { + // The URL and header secrets go through the same permissioned path as the token. + for value in std::iter::once(&mut mcp_resource.url) + .chain(mcp_resource.headers.iter_mut().flatten().map(|(_, v)| v)) + { if let Some(var_path) = value.strip_prefix("$var:") { check_scopes(authed, || format!("variables:read:{}", var_path))?; let db_authed = DbWithOptAuthed::from_authed(authed, db.clone(), Some(user_db.clone())); diff --git a/backend/windmill-mcp/Cargo.toml b/backend/windmill-mcp/Cargo.toml index 8ac23e885d..2c07b92d44 100644 --- a/backend/windmill-mcp/Cargo.toml +++ b/backend/windmill-mcp/Cargo.toml @@ -22,6 +22,7 @@ serde.workspace = true serde_json.workspace = true tracing.workspace = true rmcp.workspace = true +sse-stream = "0.2" sqlx = { workspace = true, optional = true } chrono = { workspace = true, optional = true } async-trait = { workspace = true, optional = true } diff --git a/backend/windmill-mcp/src/client/mod.rs b/backend/windmill-mcp/src/client/mod.rs index 5d13a40314..6f3928e112 100644 --- a/backend/windmill-mcp/src/client/mod.rs +++ b/backend/windmill-mcp/src/client/mod.rs @@ -3,11 +3,13 @@ //! This module provides functionality for connecting to external MCP servers //! and executing tools on them. +mod redacting; mod types; pub use types::{McpResource, McpToolSource}; use anyhow::{Context, Result}; +use redacting::UrlRedactingClient; use reqwest::header::{HeaderMap, HeaderName, HeaderValue}; use rmcp::{ model::{ @@ -97,7 +99,8 @@ impl McpClient { // Create the HTTP transport with custom client let config = StreamableHttpClientTransportConfig::with_uri(resource.url.as_str()); - let transport = StreamableHttpClientTransport::with_client(reqwest_client, config); + let transport = + StreamableHttpClientTransport::with_client(UrlRedactingClient(reqwest_client), config); // Set up client info let client_info = ClientInfo::new( diff --git a/backend/windmill-mcp/src/client/redacting.rs b/backend/windmill-mcp/src/client/redacting.rs new file mode 100644 index 0000000000..6abcc50f50 --- /dev/null +++ b/backend/windmill-mcp/src/client/redacting.rs @@ -0,0 +1,151 @@ +use std::{borrow::Cow, collections::HashMap, future::Future, sync::Arc}; + +use futures::stream::BoxStream; +use reqwest::header::{HeaderName, HeaderValue}; +use rmcp::{ + model::ClientJsonRpcMessage, + transport::streamable_http_client::{ + SseError, StreamableHttpClient, StreamableHttpError, StreamableHttpPostResponse, + }, +}; +use sse_stream::Sse; +use tracing::{instrument::WithSubscriber, Dispatch}; + +type Error = StreamableHttpError; + +/// reqwest names the URL it failed on in its errors, and an MCP server URL can +/// carry the server's key. rmcp logs those errors as well as returning them, so +/// the URL is dropped before the transport sees one, and rmcp's reqwest client, +/// which logs the errors it swallows, runs without a subscriber. +#[derive(Clone)] +pub(super) struct UrlRedactingClient(pub(super) reqwest::Client); + +async fn redacted(request: impl Future>) -> Result { + match request.with_subscriber(Dispatch::none()).await { + Err(StreamableHttpError::Client(e)) => Err(StreamableHttpError::Client(e.without_url())), + // An owned message is `HTTP : `, and the body is the server's + // to write: a common one echoes the request target (`Cannot POST /mcp/`). + Err(StreamableHttpError::UnexpectedServerResponse(Cow::Owned(message))) => { + let status = message.split_once(": ").map_or("", |(status, _)| status); + let status = status.to_string(); + Err(StreamableHttpError::UnexpectedServerResponse(status.into())) + } + r => r, + } +} + +impl StreamableHttpClient for UrlRedactingClient { + type Error = reqwest::Error; + + async fn post_message( + &self, + uri: Arc, + message: ClientJsonRpcMessage, + session_id: Option>, + auth_header: Option, + custom_headers: HashMap, + ) -> Result { + redacted( + self.0 + .post_message(uri, message, session_id, auth_header, custom_headers), + ) + .await + } + + async fn post_message_with_max_sse_event_size( + &self, + uri: Arc, + message: ClientJsonRpcMessage, + session_id: Option>, + auth_header: Option, + custom_headers: HashMap, + max_sse_event_size: usize, + ) -> Result { + redacted(self.0.post_message_with_max_sse_event_size( + uri, + message, + session_id, + auth_header, + custom_headers, + max_sse_event_size, + )) + .await + } + + async fn delete_session( + &self, + uri: Arc, + session_id: Arc, + auth_header: Option, + custom_headers: HashMap, + ) -> Result<(), Error> { + redacted( + self.0 + .delete_session(uri, session_id, auth_header, custom_headers), + ) + .await + } + + async fn get_stream( + &self, + uri: Arc, + session_id: Option>, + last_event_id: Option, + auth_header: Option, + custom_headers: HashMap, + ) -> Result>, Error> { + redacted( + self.0 + .get_stream(uri, session_id, last_event_id, auth_header, custom_headers), + ) + .await + } + + async fn get_stream_with_max_sse_event_size( + &self, + uri: Arc, + session_id: Option>, + last_event_id: Option, + auth_header: Option, + custom_headers: HashMap, + max_sse_event_size: usize, + ) -> Result>, Error> { + redacted(self.0.get_stream_with_max_sse_event_size( + uri, + session_id, + last_event_id, + auth_header, + custom_headers, + max_sse_event_size, + )) + .await + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn server_error_body_is_dropped() { + for (message, expected) in [ + ( + "HTTP 404 Not Found: Cannot POST /mcp/KEY", + "HTTP 404 Not Found", + ), + ("Cannot POST /mcp/KEY", ""), + ] { + let err = redacted(async { + Err::<(), Error>(StreamableHttpError::UnexpectedServerResponse( + message.to_string().into(), + )) + }) + .await + .unwrap_err(); + let StreamableHttpError::UnexpectedServerResponse(got) = err else { + panic!("unexpected variant: {err}"); + }; + assert_eq!(got, expected); + } + } +} diff --git a/backend/windmill-worker/src/ai/utils.rs b/backend/windmill-worker/src/ai/utils.rs index f0a0b5c20f..44513c806e 100644 --- a/backend/windmill-worker/src/ai/utils.rs +++ b/backend/windmill-worker/src/ai/utils.rs @@ -615,6 +615,17 @@ pub async fn load_mcp_tools( None }; + // A URL can carry the server's key, so it goes through the same + // permissioned path as the token. + if let Some(var_path) = mcp_resource.url.strip_prefix("$var:") { + mcp_resource.url = client.get_variable_value(var_path).await.map_err(|e| { + Error::internal_err(format!( + "Failed to resolve variable {} for the url of MCP resource {}: {}", + var_path, resource_name, e + )) + })?; + } + // Header secrets go through the same permissioned path as the token. for (name, value) in mcp_resource.headers.iter_mut().flatten() { if let Some(var_path) = value.strip_prefix("$var:") {