mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-10-13 00:03:00 +00:00
fix: resolve $var: references in mcp resource headers (#11679)
* fix: resolve $var: references in mcp resource headers Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix: require variables:read scope to resolve mcp header variables Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
9fa9b77926
commit
478eef7352
@@ -0,0 +1,110 @@
|
||||
//! An MCP resource header whose value is a `$var:` reference reaches the MCP
|
||||
//! server as the variable's value, and only for a caller allowed to read it.
|
||||
#![cfg(feature = "mcp")]
|
||||
|
||||
use sqlx::{Pool, Postgres};
|
||||
use tokio::io::AsyncReadExt;
|
||||
use windmill_common::client::AuthedClient;
|
||||
use windmill_test_utils::*;
|
||||
use windmill_worker::{load_mcp_tools, McpResourceConfig};
|
||||
|
||||
const SECRET_VALUE: &str = "S3CRET-MCP-HEADER-VALUE";
|
||||
|
||||
/// A secret variable and an MCP resource at `url` sending it as a header.
|
||||
async fn seed(db: &Pool<Postgres>, url: &str) -> anyhow::Result<()> {
|
||||
let mc = windmill_common::variables::build_crypt(db, "test-workspace").await?;
|
||||
sqlx::query(
|
||||
"INSERT INTO variable (workspace_id, path, value, is_secret, description, extra_perms)
|
||||
VALUES ('test-workspace', 'u/test-user/mcp_header', $1, true, '', '{}')",
|
||||
)
|
||||
.bind(windmill_common::variables::encrypt(&mc, SECRET_VALUE))
|
||||
.execute(db)
|
||||
.await?;
|
||||
sqlx::query(
|
||||
"INSERT INTO resource (workspace_id, path, value, resource_type, extra_perms, created_by)
|
||||
VALUES ('test-workspace', 'u/test-user/mcp', $1, 'mcp', '{}', 'test-user')",
|
||||
)
|
||||
.bind(serde_json::json!({
|
||||
"name": "mcp",
|
||||
"url": url,
|
||||
"headers": { "X-Api-Key": "$var:u/test-user/mcp_header", "X-Plain": "plain" }
|
||||
}))
|
||||
.execute(db)
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[sqlx::test(fixtures("base"))]
|
||||
async fn test_mcp_header_var_ref_is_resolved(db: Pool<Postgres>) -> anyhow::Result<()> {
|
||||
initialize_tracing().await;
|
||||
std::env::set_var("ALLOW_PRIVATE_MCP_SERVER_URLS", "true");
|
||||
|
||||
// Not an MCP server: it captures the first request and hangs up, so loading
|
||||
// fails at the connect step after the headers were sent.
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
|
||||
let mcp_port = listener.local_addr()?.port();
|
||||
let first_request = tokio::spawn(async move {
|
||||
let (mut socket, _) = listener.accept().await?;
|
||||
let mut buf = vec![0u8; 16 * 1024];
|
||||
let n = socket.read(&mut buf).await?;
|
||||
anyhow::Ok(String::from_utf8_lossy(&buf[..n]).to_lowercase())
|
||||
});
|
||||
seed(&db, &format!("http://127.0.0.1:{mcp_port}/mcp")).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 config = vec![McpResourceConfig {
|
||||
resource_path: "$res:u/test-user/mcp".to_string(),
|
||||
include_tools: None,
|
||||
exclude_tools: None,
|
||||
}];
|
||||
let _ = load_mcp_tools(&db, "test-workspace", config, &client).await;
|
||||
|
||||
let request = first_request.await??;
|
||||
assert!(
|
||||
request.contains(&format!("x-api-key: {}", SECRET_VALUE.to_lowercase())),
|
||||
"the header must carry the variable's value, not the reference: {request}"
|
||||
);
|
||||
assert!(request.contains("x-plain: plain"), "{request}");
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// A token scoped to the resource alone must not have the API read the
|
||||
/// variable a header points at.
|
||||
#[sqlx::test(fixtures("base"))]
|
||||
async fn test_mcp_header_var_ref_requires_variable_scope(db: Pool<Postgres>) -> anyhow::Result<()> {
|
||||
initialize_tracing().await;
|
||||
seed(&db, "https://mcp.invalid.windmill.test").await?;
|
||||
sqlx::query(
|
||||
"INSERT INTO token (token_hash, token_prefix, token, email, label, super_admin, scopes)
|
||||
VALUES (encode(sha256('RES_ONLY_TOKEN'::bytea), 'hex'), 'RES_ONLY_T', 'RES_ONLY_TOKEN',
|
||||
'test@windmill.dev', 'resources only', false, '{resources:read}')",
|
||||
)
|
||||
.execute(&db)
|
||||
.await?;
|
||||
|
||||
let server = ApiServer::start(db.clone()).await?;
|
||||
let resp = reqwest::Client::new()
|
||||
.get(format!(
|
||||
"http://localhost:{}/api/w/test-workspace/resources/mcp_tools/u/test-user/mcp",
|
||||
server.addr.port()
|
||||
))
|
||||
.header("Authorization", "Bearer RES_ONLY_TOKEN")
|
||||
.send()
|
||||
.await?;
|
||||
let status = resp.status();
|
||||
let body = resp.text().await?;
|
||||
assert_eq!(status, reqwest::StatusCode::FORBIDDEN, "{body}");
|
||||
assert!(
|
||||
body.contains("variables:read:u/test-user/mcp_header"),
|
||||
"{body}"
|
||||
);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
@@ -23,14 +23,12 @@ async fn with_deadline<T>(
|
||||
what: &str,
|
||||
fut: impl std::future::Future<Output = Result<T>>,
|
||||
) -> Result<T> {
|
||||
tokio::time::timeout(MCP_DEADLINE, fut)
|
||||
.await
|
||||
.map_err(|_| {
|
||||
Error::ExecutionErr(format!(
|
||||
"MCP server did not answer within {}s ({what})",
|
||||
MCP_DEADLINE.as_secs()
|
||||
))
|
||||
})?
|
||||
tokio::time::timeout(MCP_DEADLINE, fut).await.map_err(|_| {
|
||||
Error::ExecutionErr(format!(
|
||||
"MCP server did not answer within {}s ({what})",
|
||||
MCP_DEADLINE.as_secs()
|
||||
))
|
||||
})?
|
||||
}
|
||||
|
||||
/// Connect to the MCP server described by the `mcp` resource at `path`.
|
||||
@@ -65,8 +63,9 @@ async fn connect_mcp_client(
|
||||
let resource_value = not_found_if_none(resource_value_o, "Resource", path)?
|
||||
.ok_or_else(|| Error::BadRequest(format!("Empty resource value for {}", path)))?;
|
||||
|
||||
let mcp_resource = serde_json::from_str::<windmill_mcp::McpResource>(resource_value.0.get())
|
||||
.map_err(|e| Error::BadRequest(format!("Failed to parse MCP resource: {}", e)))?;
|
||||
let mut mcp_resource =
|
||||
serde_json::from_str::<windmill_mcp::McpResource>(resource_value.0.get())
|
||||
.map_err(|e| Error::BadRequest(format!("Failed to parse MCP resource: {}", e)))?;
|
||||
|
||||
#[cfg(feature = "oauth2")]
|
||||
{
|
||||
@@ -126,6 +125,15 @@ 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) {
|
||||
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()));
|
||||
*value = get_value_internal(&db_authed, w_id, var_path, false).await?;
|
||||
}
|
||||
}
|
||||
|
||||
windmill_mcp::McpClient::from_resource(mcp_resource, token)
|
||||
.await
|
||||
.map_err(|e| Error::ExecutionErr(format!("Failed to connect to MCP server: {}", e)))
|
||||
|
||||
@@ -70,7 +70,8 @@ impl McpClient {
|
||||
headers.insert(name, value);
|
||||
}
|
||||
_ => {
|
||||
tracing::warn!("Invalid header: {}={}", key, value);
|
||||
// The value may be a resolved secret: keep it out of the logs.
|
||||
tracing::warn!("Invalid header: {}", key);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -570,7 +570,7 @@ pub async fn load_mcp_tools(
|
||||
// is not allowed to read (resources:read:{path}). Reading through the raw
|
||||
// db pool here would bypass the authorization enforced by the regular MCP
|
||||
// tools API (get_mcp_tools) and act as a confused deputy.
|
||||
let mcp_resource = client
|
||||
let mut mcp_resource = client
|
||||
.get_resource_value::<McpResource>(path)
|
||||
.await
|
||||
.map_err(|e| {
|
||||
@@ -615,6 +615,18 @@ pub async fn load_mcp_tools(
|
||||
None
|
||||
};
|
||||
|
||||
// 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:") {
|
||||
*value = client.get_variable_value(var_path).await.map_err(|e| {
|
||||
Error::internal_err(format!(
|
||||
"Failed to resolve variable {} for header {} of MCP resource {}: {}",
|
||||
var_path, name, resource_name, e
|
||||
))
|
||||
})?;
|
||||
}
|
||||
}
|
||||
|
||||
// Create new MCP client for this execution
|
||||
tracing::debug!("Creating fresh MCP client for {}", resource_name);
|
||||
let mcp_conn = McpClient::from_resource(mcp_resource, token)
|
||||
|
||||
Reference in New Issue
Block a user