feat(mcp): add streamable http server and fine-grained authorization

This commit is contained in:
二丫讲梵 authored and GitHub committed 2026-08-31 22:56:16 +08:00
1 parent f725c50b2c
commit f075caef21
47 files changed
+4960 -509

No files matched your search

+70
View File
@@ -163,6 +163,13 @@ pub trait DbxBackend: Send + Sync {
async fn load_mcp_global_policy(&self) -> Result<McpGlobalPolicy, String>;
async fn load_connections(&self) -> Result<Vec<ConnectionConfig>, String>;
/// Return database names visible to the DBX connection itself. The MCP
/// server applies its own database-scope policy before exposing these
/// names to a client.
async fn list_databases(&self, connection: &ConnectionConfig) -> Result<Vec<String>, String> {
let _ = connection;
Err("Database metadata is not supported by this backend.".to_string())
}
async fn load_connection_group_paths(&self) -> Result<HashMap<String, Vec<String>>, String> {
Ok(HashMap::new())
}
@@ -474,6 +481,12 @@ impl WebBackend {
}
impl LocalBackend {
/// Reuse an already initialized DBX application state, for hosts that
/// embed MCP alongside their own HTTP server.
pub fn from_app_state(state: Arc<AppState>, data_dir: PathBuf) -> Self {
Self { state, data_dir }
}
pub async fn open(path: &Path) -> Result<Self, String> {
let storage = Storage::open(path).await?;
let configs = storage.load_connections().await?;
@@ -567,6 +580,16 @@ impl DbxBackend for LocalBackend {
Ok(configs)
}
async fn list_databases(&self, connection: &ConnectionConfig) -> Result<Vec<String>, String> {
// `list_databases_core` supports many database engines and therefore
// produces a very large future. Boxing it here keeps the async-trait
// implementation below Rust's type-layout recursion limit in desktop
// builds without changing its execution behaviour.
Box::pin(dbx_core::schema::list_databases_core(&self.state, &connection.id))
.await
.map(|databases| databases.into_iter().map(|database| database.name).collect())
}
async fn load_connection_group_paths(&self) -> Result<HashMap<String, Vec<String>>, String> {
Ok(self.state.storage.load_sidebar_layout().await?.map(connection_group_paths).unwrap_or_default())
}
@@ -763,6 +786,51 @@ impl DbxBackend for WebBackend {
.map_err(|error| format!("Invalid connection list response: {error}"))
}
async fn list_databases(&self, connection: &ConnectionConfig) -> Result<Vec<String>, String> {
self.ensure_connected(connection).await?;
match connection.db_type {
DatabaseType::MongoDb => self
.request(
reqwest::Method::POST,
"/api/mongo/list-databases",
Some(json!({ "connectionId": connection.id })),
)
.await?
.json::<Vec<String>>()
.await
.map_err(|error| format!("Invalid MongoDB database list response: {error}")),
DatabaseType::Redis => {
let databases = self
.request(
reqwest::Method::POST,
"/api/redis/list-databases",
Some(json!({ "connectionId": connection.id })),
)
.await?
.json::<Vec<Value>>()
.await
.map_err(|error| format!("Invalid Redis database list response: {error}"))?;
Ok(databases
.into_iter()
.filter_map(|database| {
database.get("db").and_then(Value::as_u64).map(|database| database.to_string())
})
.collect())
}
_ => self
.request(
reqwest::Method::GET,
&format!("/api/schema/databases?connection_id={}", url_encode(&connection.id)),
None,
)
.await?
.json::<Vec<dbx_core::db::DatabaseInfo>>()
.await
.map(|databases| databases.into_iter().map(|database| database.name).collect())
.map_err(|error| format!("Invalid database list response: {error}")),
}
}
async fn load_connection_group_paths(&self) -> Result<HashMap<String, Vec<String>>, String> {
let layout = self
.request(reqwest::Method::GET, "/api/layout/sidebar", None)
@@ -1869,6 +1937,8 @@ mod tests {
read_only,
allow_dangerous_sql: false,
allowed_connection_ids: None,
allowed_tool_names: None,
connection_policies: Vec::new(),
query_timeout_secs: None,
}
}
+13
View File
@@ -0,0 +1,13 @@
use axum::Json;
use serde::Serialize;
/// A deliberately non-sensitive liveness response for HTTP deployment probes.
#[derive(Serialize)]
pub struct HealthStatus {
pub status: &'static str,
pub transport: &'static str,
}
pub async fn health() -> Json<HealthStatus> {
Json(HealthStatus { status: "ok", transport: "streamable-http" })
}
+118
View File
@@ -0,0 +1,118 @@
use std::{io, sync::Arc};
use axum::{http::Method, middleware, routing::get, Router};
use rmcp::transport::{
streamable_http_server::{session::local::LocalSessionManager, tower::StreamableHttpService},
StreamableHttpServerConfig,
};
use tokio_util::sync::CancellationToken;
use tower_http::cors::{AllowOrigin, Any, CorsLayer};
use crate::{
diagnostics::health,
http_auth::{authorize_request, HttpAuth},
runtime::HttpRuntimeConfig,
DbxBackend, DbxMcpServer, McpScope,
};
/// Builds a protected Streamable HTTP MCP router for embedding in an existing
/// HTTP server. The embedded host remains responsible for choosing the public
/// listener and lifecycle; this router only owns the `/mcp` protocol route.
pub fn streamable_http_router(
backend: Arc<dyn DbxBackend>,
path: &str,
auth: HttpAuth,
allowed_hosts: Vec<String>,
web_mode: bool,
) -> Router {
build_streamable_http_router(backend, path, auth, allowed_hosts, web_mode, None)
}
fn build_streamable_http_router(
backend: Arc<dyn DbxBackend>,
path: &str,
auth: HttpAuth,
allowed_hosts: Vec<String>,
web_mode: bool,
cancellation: Option<CancellationToken>,
) -> Router {
let mut rmcp_config = StreamableHttpServerConfig::default().with_allowed_hosts(allowed_hosts);
if let Some(cancellation) = cancellation {
rmcp_config = rmcp_config.with_cancellation_token(cancellation);
}
let server_backend = backend.clone();
let scope = McpScope::from_env();
let service: StreamableHttpService<DbxMcpServer, LocalSessionManager> = StreamableHttpService::new(
move || Ok(DbxMcpServer::with_runtime_options(server_backend.clone(), scope.clone(), web_mode)),
Default::default(),
rmcp_config,
);
// The authentication middleware and CORS response must use the same
// predicate. In particular, loopback desktop mode permits localhost
// browser origins without requiring users to enumerate every development
// port, while remote mode still requires exact configured origins.
let cors_auth = auth.clone();
let router =
Router::new().nest_service(path, service).layer(middleware::from_fn_with_state(auth, authorize_request));
router.layer(
CorsLayer::new()
.allow_origin(AllowOrigin::predicate(move |origin, _| {
origin.to_str().is_ok_and(|origin| cors_auth.origin_is_allowed(origin))
}))
.allow_methods([Method::GET, Method::POST, Method::DELETE])
.allow_headers(Any),
)
}
/// Serves one stateful rmcp Streamable HTTP endpoint. Every MCP protocol
/// session receives a fresh `DbxMcpServer`, while the database backend remains
/// shared and all authorization happens before rmcp sees a request.
pub async fn serve_streamable_http(backend: Arc<dyn DbxBackend>, config: HttpRuntimeConfig) -> io::Result<()> {
let cancellation = CancellationToken::new();
let shutdown = cancellation.clone();
tokio::spawn(async move {
let _ = tokio::signal::ctrl_c().await;
shutdown.cancel();
});
serve_streamable_http_with_shutdown(backend, config, cancellation).await
}
/// Serves the HTTP transport until `cancellation` is cancelled. Embedding
/// hosts use this variant so their own lifecycle controls shutdown instead of
/// relying on a process-wide Ctrl-C handler.
pub async fn serve_streamable_http_with_shutdown(
backend: Arc<dyn DbxBackend>,
config: HttpRuntimeConfig,
cancellation: CancellationToken,
) -> io::Result<()> {
let listener = tokio::net::TcpListener::bind(config.bind_addr).await?;
serve_streamable_http_on_listener(backend, config, cancellation, listener).await
}
/// Variant for hosts that must bind synchronously before reporting the server
/// as healthy (for example, DBX Desktop settings UI).
pub async fn serve_streamable_http_on_listener(
backend: Arc<dyn DbxBackend>,
config: HttpRuntimeConfig,
cancellation: CancellationToken,
listener: tokio::net::TcpListener,
) -> io::Result<()> {
let mcp_router = build_streamable_http_router(
backend,
&config.path,
config.auth,
config.allowed_hosts,
false,
Some(cancellation.child_token()),
);
let router = Router::new().route("/healthz", get(health)).route("/readyz", get(health)).merge(mcp_router);
eprintln!("DBX MCP Streamable HTTP listening on http://{}{}", config.bind_addr, config.path);
axum::serve(listener, router)
.with_graceful_shutdown(async move {
cancellation.cancelled().await;
})
.await
}
+134
View File
@@ -0,0 +1,134 @@
use std::{collections::HashSet, sync::Arc};
use axum::{
extract::{Request, State},
http::{header, HeaderValue, StatusCode},
middleware::Next,
response::{IntoResponse, Response},
};
use url::Url;
/// Authentication and browser-origin policy for the Streamable HTTP endpoint.
/// The token is intentionally not `Debug` and is never exposed by diagnostics.
#[derive(Clone)]
pub struct HttpAuth {
token: Arc<[u8]>,
allowed_origins: HashSet<String>,
allow_loopback_origins: bool,
}
impl HttpAuth {
pub fn new(
token: String,
allowed_origins: impl IntoIterator<Item = String>,
allow_loopback_origins: bool,
) -> Result<Self, String> {
if token.trim().is_empty() || token.contains(char::is_whitespace) {
return Err("MCP HTTP bearer token must be non-empty and contain no whitespace".into());
}
let mut normalized_origins = HashSet::new();
for origin in allowed_origins {
normalized_origins.insert(normalize_origin(&origin)?);
}
Ok(Self { token: Arc::from(token.into_bytes()), allowed_origins: normalized_origins, allow_loopback_origins })
}
fn token_matches(&self, candidate: &str) -> bool {
let candidate = candidate.as_bytes();
if candidate.len() != self.token.len() {
return false;
}
// Keep comparison work independent of the first mismatching byte.
let difference = self
.token
.iter()
.zip(candidate)
.fold(0_u8, |difference, (expected, actual)| difference | (expected ^ actual));
difference == 0
}
pub fn origin_is_allowed(&self, origin: &str) -> bool {
let Ok(origin) = normalize_origin(origin) else {
return false;
};
if self.allowed_origins.contains(&origin) {
return true;
}
self.allow_loopback_origins && origin_is_loopback(&origin)
}
}
pub async fn authorize_request(State(auth): State<HttpAuth>, request: Request, next: Next) -> Response {
let method = request.method().clone();
let path = request.uri().path().to_string();
if let Some(origin) = request.headers().get(header::ORIGIN) {
let origin = match origin.to_str() {
Ok(origin) => origin,
Err(_) => {
log::warn!(target: "dbx_mcp::audit", "MCP HTTP request rejected: method={method} path={path} reason=invalid-origin");
return forbidden();
}
};
if !auth.origin_is_allowed(origin) {
log::warn!(target: "dbx_mcp::audit", "MCP HTTP request rejected: method={method} path={path} origin={origin} reason=origin-not-allowed");
return forbidden();
}
}
let Some(token) = bearer_token(request.headers().get(header::AUTHORIZATION)) else {
log::warn!(target: "dbx_mcp::audit", "MCP HTTP request rejected: method={method} path={path} reason=missing-bearer-token");
return unauthorized();
};
if !auth.token_matches(token) {
log::warn!(target: "dbx_mcp::audit", "MCP HTTP request rejected: method={method} path={path} reason=invalid-bearer-token");
return unauthorized();
}
let response = next.run(request).await;
log::info!(target: "dbx_mcp::audit", "MCP HTTP request authenticated: method={method} path={path} status={}", response.status());
response
}
fn bearer_token(value: Option<&HeaderValue>) -> Option<&str> {
let value = value?.to_str().ok()?;
let (scheme, token) = value.split_once(' ')?;
(scheme.eq_ignore_ascii_case("bearer") && !token.is_empty() && !token.contains(char::is_whitespace))
.then_some(token)
}
fn normalize_origin(origin: &str) -> Result<String, String> {
let url = Url::parse(origin).map_err(|_| format!("invalid allowed origin: {origin}"))?;
if !matches!(url.scheme(), "http" | "https")
|| url.host_str().is_none()
|| url.path() != "/"
|| url.query().is_some()
|| url.fragment().is_some()
{
return Err(format!("allowed origin must be an HTTP(S) origin without a path: {origin}"));
}
Ok(url.origin().ascii_serialization())
}
fn origin_is_loopback(origin: &str) -> bool {
let Ok(url) = Url::parse(origin) else {
return false;
};
match url.host_str() {
Some("localhost") => true,
Some(host) => host.parse::<std::net::IpAddr>().is_ok_and(|ip| ip.is_loopback()),
None => false,
}
}
fn unauthorized() -> Response {
let mut response = (StatusCode::UNAUTHORIZED, "Unauthorized").into_response();
response.headers_mut().insert(header::WWW_AUTHENTICATE, HeaderValue::from_static("Bearer"));
response
}
fn forbidden() -> Response {
(StatusCode::FORBIDDEN, "Forbidden").into_response()
}
+7
View File
@@ -1,11 +1,18 @@
pub mod backend;
pub mod diagnostics;
pub mod http;
pub mod http_auth;
pub mod paths;
pub mod runtime;
pub mod server;
pub mod session;
pub mod transport;
pub use backend::{ConnectionSummary, DbxBackend, LocalBackend, WebBackend};
pub use dbx_core::mongo_shell as mongo;
pub use http::{serve_streamable_http_on_listener, serve_streamable_http_with_shutdown, streamable_http_router};
pub use http_auth::HttpAuth;
pub use runtime::{HttpRuntimeConfig, McpTransport, RuntimeConfig};
pub use server::{DbxMcpServer, McpScope};
pub use session::McpSessionStore;
pub use transport::with_legacy_discovery_fallback;
+18 -5
View File
@@ -1,10 +1,14 @@
use std::sync::Arc;
use dbx_mcp::{with_legacy_discovery_fallback, DbxBackend, DbxMcpServer, LocalBackend, WebBackend};
use dbx_mcp::{
http::serve_streamable_http, with_legacy_discovery_fallback, DbxBackend, DbxMcpServer, LocalBackend, McpTransport,
RuntimeConfig, WebBackend,
};
use rmcp::ServiceExt;
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let runtime = RuntimeConfig::from_environment_and_args()?;
let backend: Arc<dyn DbxBackend> = if let Ok(base_url) = std::env::var("DBX_WEB_URL") {
Arc::new(
WebBackend::new(base_url, std::env::var("DBX_WEB_PASSWORD").unwrap_or_default())
@@ -14,8 +18,17 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
let db_path = dbx_mcp::paths::storage_db_path().map_err(std::io::Error::other)?;
Arc::new(LocalBackend::open(&db_path).await.map_err(std::io::Error::other)?)
};
let transport = with_legacy_discovery_fallback(rmcp::transport::stdio());
let service = DbxMcpServer::new(backend).serve(transport).await?;
service.waiting().await?;
Ok(())
match runtime.transport {
McpTransport::Stdio => {
let transport = with_legacy_discovery_fallback(rmcp::transport::stdio());
let service = DbxMcpServer::new(backend).serve(transport).await?;
service.waiting().await?;
Ok(())
}
McpTransport::StreamableHttp => {
let http = runtime.http.ok_or_else(|| std::io::Error::other("missing HTTP MCP runtime configuration"))?;
serve_streamable_http(backend, http).await?;
Ok(())
}
}
}
+174
View File
@@ -0,0 +1,174 @@
use std::{
env, fs,
net::{IpAddr, SocketAddr},
};
use crate::http_auth::HttpAuth;
const DEFAULT_HTTP_HOST: &str = "127.0.0.1";
const DEFAULT_HTTP_PORT: u16 = 5225;
const DEFAULT_HTTP_PATH: &str = "/mcp";
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum McpTransport {
Stdio,
StreamableHttp,
}
pub struct RuntimeConfig {
pub transport: McpTransport,
pub http: Option<HttpRuntimeConfig>,
}
pub struct HttpRuntimeConfig {
pub(crate) bind_addr: SocketAddr,
pub(crate) path: String,
pub(crate) auth: HttpAuth,
pub(crate) allowed_hosts: Vec<String>,
}
impl HttpRuntimeConfig {
pub fn new(bind_addr: SocketAddr, path: String, auth: HttpAuth, allowed_hosts: Vec<String>) -> Self {
Self { bind_addr, path, auth, allowed_hosts }
}
pub fn bind_addr(&self) -> SocketAddr {
self.bind_addr
}
}
impl RuntimeConfig {
/// Parses DBX MCP's small runtime surface without adding a command-line dependency.
/// `stdio` remains the default, preserving all existing npm and native launchers.
pub fn from_environment_and_args() -> Result<Self, String> {
let mut transport = env::var("DBX_MCP_TRANSPORT").ok();
let mut host = env::var("DBX_MCP_HTTP_HOST").ok();
let mut port = env::var("DBX_MCP_HTTP_PORT").ok();
let mut path = env::var("DBX_MCP_HTTP_PATH").ok();
let mut allow_remote = env_flag("DBX_MCP_HTTP_ALLOW_REMOTE")?;
let mut explicit_allow_remote = false;
let mut arguments = env::args().skip(1);
while let Some(argument) = arguments.next() {
match argument.as_str() {
"--stdio" => transport = Some("stdio".into()),
"--http" => transport = Some("streamable-http".into()),
"--http-allow-remote" => {
allow_remote = true;
explicit_allow_remote = true;
}
"--transport" => transport = Some(required_argument(&mut arguments, "--transport")?),
"--http-host" => host = Some(required_argument(&mut arguments, "--http-host")?),
"--http-port" => port = Some(required_argument(&mut arguments, "--http-port")?),
"--http-path" => path = Some(required_argument(&mut arguments, "--http-path")?),
_ => {
if let Some(value) = argument.strip_prefix("--transport=") {
transport = Some(value.into());
} else if let Some(value) = argument.strip_prefix("--http-host=") {
host = Some(value.into());
} else if let Some(value) = argument.strip_prefix("--http-port=") {
port = Some(value.into());
} else if let Some(value) = argument.strip_prefix("--http-path=") {
path = Some(value.into());
} else {
return Err(format!("unknown DBX MCP argument: {argument}"));
}
}
}
}
let transport = parse_transport(transport.as_deref().unwrap_or("stdio"))?;
if transport == McpTransport::Stdio {
return Ok(Self { transport, http: None });
}
let host = host.unwrap_or_else(|| DEFAULT_HTTP_HOST.into());
let ip = host.parse::<IpAddr>().map_err(|_| "DBX_MCP_HTTP_HOST must be an IP address".to_string())?;
let port = port
.unwrap_or_else(|| DEFAULT_HTTP_PORT.to_string())
.parse::<u16>()
.map_err(|_| "DBX_MCP_HTTP_PORT must be a valid TCP port".to_string())?;
let bind_addr = SocketAddr::new(ip, port);
let path = path.unwrap_or_else(|| DEFAULT_HTTP_PATH.into());
validate_path(&path)?;
let is_loopback = ip.is_loopback();
if !is_loopback && !(allow_remote && explicit_allow_remote) {
return Err(
"non-loopback HTTP binding requires both DBX_MCP_HTTP_ALLOW_REMOTE=1 and --http-allow-remote".into()
);
}
let allowed_hosts = comma_separated_env("DBX_MCP_HTTP_ALLOWED_HOSTS");
let allowed_origins = comma_separated_env("DBX_MCP_HTTP_ALLOWED_ORIGINS");
if !is_loopback && (allowed_hosts.is_empty() || allowed_origins.is_empty()) {
return Err(
"non-loopback HTTP binding requires DBX_MCP_HTTP_ALLOWED_HOSTS and DBX_MCP_HTTP_ALLOWED_ORIGINS".into(),
);
}
let token = http_token_from_environment()?;
let auth = HttpAuth::new(token, allowed_origins.clone(), is_loopback)?;
let allowed_hosts =
if is_loopback { vec!["localhost".into(), "127.0.0.1".into(), "::1".into()] } else { allowed_hosts };
Ok(Self { transport, http: Some(HttpRuntimeConfig { bind_addr, path, auth, allowed_hosts }) })
}
}
fn parse_transport(value: &str) -> Result<McpTransport, String> {
match value.trim().to_ascii_lowercase().as_str() {
"stdio" => Ok(McpTransport::Stdio),
"http" | "streamable-http" | "streamable_http" => Ok(McpTransport::StreamableHttp),
_ => Err("DBX_MCP_TRANSPORT must be stdio or streamable-http".into()),
}
}
fn required_argument(arguments: &mut impl Iterator<Item = String>, name: &str) -> Result<String, String> {
arguments.next().filter(|value| !value.starts_with("--")).ok_or_else(|| format!("{name} requires a value"))
}
fn env_flag(name: &str) -> Result<bool, String> {
match env::var(name) {
Ok(value) => match value.trim().to_ascii_lowercase().as_str() {
"1" | "true" | "yes" => Ok(true),
"0" | "false" | "no" | "" => Ok(false),
_ => Err(format!("{name} must be true or false")),
},
Err(env::VarError::NotPresent) => Ok(false),
Err(env::VarError::NotUnicode(_)) => Err(format!("{name} must be valid UTF-8")),
}
}
fn comma_separated_env(name: &str) -> Vec<String> {
env::var(name)
.ok()
.into_iter()
.flat_map(|value| {
value.split(',').map(str::trim).filter(|value| !value.is_empty()).map(ToOwned::to_owned).collect::<Vec<_>>()
})
.collect()
}
fn http_token_from_environment() -> Result<String, String> {
let inline_token = env::var("DBX_MCP_HTTP_TOKEN").ok();
let token_file = env::var("DBX_MCP_HTTP_TOKEN_FILE").ok();
match (inline_token, token_file) {
(Some(_), Some(_)) => Err("set only one of DBX_MCP_HTTP_TOKEN or DBX_MCP_HTTP_TOKEN_FILE".into()),
(Some(token), None) if !token.trim().is_empty() => Ok(token),
(None, Some(path)) => fs::read_to_string(&path)
.map_err(|error| format!("failed to read DBX_MCP_HTTP_TOKEN_FILE: {error}"))
.map(|token| token.trim_end_matches(['\r', '\n']).to_owned())
.and_then(|token| {
(!token.is_empty()).then_some(token).ok_or_else(|| "DBX_MCP_HTTP_TOKEN_FILE is empty".into())
}),
_ => Err("Streamable HTTP requires DBX_MCP_HTTP_TOKEN or DBX_MCP_HTTP_TOKEN_FILE".into()),
}
}
fn validate_path(path: &str) -> Result<(), String> {
if path == "/" || !path.starts_with('/') || path.ends_with('/') || path.contains('?') || path.contains('#') {
return Err("DBX_MCP_HTTP_PATH must be an absolute path such as /mcp".into());
}
Ok(())
}
+481 -54
View File
@@ -18,18 +18,25 @@ use dbx_core::{
db::redis_driver::{classify_command, parse_command_argv, RedisCommandResult, RedisCommandSafety},
models::connection::DatabaseType,
production_safety::{
is_production_database, mongo_pipeline_targets_production_database, targets_production_database,
is_production_database, mongo_pipeline_targets_production_database, sql_references_disallowed_database,
targets_production_database,
},
query_execution_sql::is_write_sql_for_database,
sql_risk::{
classify_sql_risk_for_database, is_dangerous_sql_for_database, mcp_sql_has_forbidden_database_switch, SqlRisk,
},
storage::McpGlobalPolicy,
storage::{McpDatabaseScope, McpGlobalPolicy},
};
#[derive(Debug, Deserialize, schemars::JsonSchema)]
pub struct ListConnectionsRequest {}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
pub struct ListDatabasesRequest {
#[serde(flatten)]
pub selector: ConnectionSelector,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
pub struct ConnectionSelector {
#[schemars(description = "Unique ID of the DBX connection")]
@@ -136,7 +143,9 @@ pub struct DuplicateConnectionRequest {
#[derive(Debug, Deserialize, schemars::JsonSchema)]
pub struct RemoveConnectionRequest {
pub connection_name: String,
#[schemars(description = "Name of the DBX connection when connection_id is not provided")]
#[schemars(extend("type" = "string"))]
pub connection_name: Option<String>,
#[schemars(extend("type" = "string"))]
pub connection_id: Option<String>,
}
@@ -244,6 +253,14 @@ pub struct McpScope {
struct ResolvedConnection {
connection: dbx_core::models::connection::ConnectionConfig,
policy: McpGlobalPolicy,
database_scope: DatabaseScope,
}
#[derive(Clone, Debug)]
enum DatabaseScope {
All,
Selected(Vec<String>),
None,
}
impl McpScope {
@@ -324,6 +341,9 @@ impl DbxMcpServer {
&self,
Parameters(ListConnectionsRequest {}): Parameters<ListConnectionsRequest>,
) -> CallToolResult {
if let Err(error) = self.ensure_tool_allowed("dbx_list_connections").await {
return error;
}
match self.load_scoped_connections().await {
Ok(connections) if connections.is_empty() => text("No connections configured in DBX."),
Ok(connections) => {
@@ -342,13 +362,31 @@ impl DbxMcpServer {
}
}
#[tool(name = "dbx_list_tables", description = "List tables and views for a database connection")]
async fn list_tables(&self, Parameters(request): Parameters<ListTablesRequest>) -> CallToolResult {
#[tool(
name = "dbx_list_databases",
description = "List database names available through a DBX connection. When the connection has a database allowlist, returns only the databases allowed by DBX MCP settings; use this before listing tables if the connection has no default database."
)]
async fn list_databases(&self, Parameters(request): Parameters<ListDatabasesRequest>) -> CallToolResult {
if let Err(error) = self.ensure_tool_allowed("dbx_list_databases").await {
return error;
}
let resolved = match self.resolve_connection(&request.selector).await {
Ok(resolved) => resolved,
Err(error) => return error,
};
let database = match self.resolve_database(request.database, &resolved.connection) {
self.list_databases_for_resolved(&resolved).await
}
#[tool(name = "dbx_list_tables", description = "List tables and views for a database connection")]
async fn list_tables(&self, Parameters(request): Parameters<ListTablesRequest>) -> CallToolResult {
if let Err(error) = self.ensure_tool_allowed("dbx_list_tables").await {
return error;
}
let resolved = match self.resolve_connection(&request.selector).await {
Ok(resolved) => resolved,
Err(error) => return error,
};
let database = match self.resolve_database(request.database, &resolved) {
Ok(database) => database,
Err(error) => return error,
};
@@ -378,11 +416,14 @@ impl DbxMcpServer {
#[tool(name = "dbx_describe_table", description = "Get column definitions for a table")]
async fn describe_table(&self, Parameters(request): Parameters<DescribeTableRequest>) -> CallToolResult {
if let Err(error) = self.ensure_tool_allowed("dbx_describe_table").await {
return error;
}
let resolved = match self.resolve_connection(&request.selector).await {
Ok(resolved) => resolved,
Err(error) => return error,
};
let database = match self.resolve_database(request.database, &resolved.connection) {
let database = match self.resolve_database(request.database, &resolved) {
Ok(database) => database,
Err(error) => return error,
};
@@ -402,6 +443,9 @@ impl DbxMcpServer {
description = "Execute a SQL query on a database connection (max 100 rows returned)"
)]
async fn execute_query(&self, Parameters(request): Parameters<ExecuteQueryRequest>) -> CallToolResult {
if let Err(error) = self.ensure_tool_allowed("dbx_execute_query").await {
return error;
}
let explicit_cell_window = (request.cell_char_offset.is_some() || request.cell_char_limit.is_some())
.then(|| QueryCellWindow::from_options(request.cell_char_offset, request.cell_char_limit));
let resolved = match self.resolve_connection(&request.selector).await {
@@ -415,6 +459,13 @@ impl DbxMcpServer {
"Redis connections do not accept SQL through dbx_execute_query. Use dbx_execute_redis_command.",
);
}
// Database discovery does not require a default database. In a
// narrowed MCP scope it must never reveal names outside the allowlist,
// so serve it directly from the scoped discovery path instead of
// passing SHOW DATABASES to the database driver.
if is_database_discovery_sql(&request.sql) {
return self.list_databases_for_resolved(&resolved).await;
}
// Resolve the session before the database so its connection/database
// binding is enforced on every stateful query.
let session = match request.session_id.as_deref().map(str::trim).filter(|id| !id.is_empty()) {
@@ -441,7 +492,7 @@ impl DbxMcpServer {
}
None => None,
};
let database = match self.resolve_database(request.database, connection) {
let database = match self.resolve_database(request.database, &resolved) {
Ok(database) => database,
Err(error) => return error,
};
@@ -457,7 +508,13 @@ impl DbxMcpServer {
}
}
if connection.db_type == DatabaseType::MongoDb {
let command = match validate_mongo_command(connection, &resolved.policy, &database, &request.sql) {
let command = match validate_mongo_command(
connection,
&resolved.policy,
&resolved.database_scope,
&database,
&request.sql,
) {
Ok(command) => command,
Err(error) => return error,
};
@@ -476,6 +533,12 @@ impl DbxMcpServer {
// switching is allowed — unless a hard database scope is configured,
// which a USE statement could otherwise escape.
let allow_database_switch = session.is_some() && self.scope.database.is_none();
if connection.db_type != DatabaseType::MongoDb {
if let Err(error) = ensure_sql_database_scope(&resolved.database_scope, connection, &database, &request.sql)
{
return error;
}
}
let permissions =
match validate_sql_policy(connection, &resolved.policy, &database, &request.sql, allow_database_switch) {
Ok(permissions) => permissions,
@@ -512,6 +575,9 @@ impl DbxMcpServer {
description = "Open a stateful query session pinned to a single backend connection. Returns a session ID for dbx_execute_query: USE, SET CATALOG, session variables and temporary tables persist across calls within the session. Close with dbx_close_session when done; idle sessions expire after 30 minutes."
)]
async fn open_session(&self, Parameters(request): Parameters<OpenSessionRequest>) -> CallToolResult {
if let Err(error) = self.ensure_tool_allowed("dbx_open_session").await {
return error;
}
let resolved = match self.resolve_connection(&request.selector).await {
Ok(resolved) => resolved,
Err(error) => return error,
@@ -523,7 +589,7 @@ impl DbxMcpServer {
format!("Sessions are only supported for SQL connections; \"{}\" is not one.", connection.name),
);
}
let database = match self.resolve_database(request.database, connection) {
let database = match self.resolve_database(request.database, &resolved) {
Ok(database) => database,
Err(error) => return error,
};
@@ -543,6 +609,9 @@ impl DbxMcpServer {
description = "Close a stateful query session and release its pinned backend connection"
)]
async fn close_session(&self, Parameters(request): Parameters<CloseSessionRequest>) -> CallToolResult {
if let Err(error) = self.ensure_tool_allowed("dbx_close_session").await {
return error;
}
let (session, expired) = self.sessions.begin_close(&request.session_id).await.into_parts();
self.close_backend_sessions_best_effort(expired).await;
let Some(session) = session else {
@@ -572,6 +641,9 @@ impl DbxMcpServer {
&self,
Parameters(request): Parameters<ExecuteRedisCommandRequest>,
) -> CallToolResult {
if let Err(error) = self.ensure_tool_allowed("dbx_execute_redis_command").await {
return error;
}
let resolved = match self.resolve_connection(&request.selector).await {
Ok(resolved) => resolved,
Err(error) => return error,
@@ -610,7 +682,7 @@ impl DbxMcpServer {
"MCP Redis command execution is read-only in DBX MCP settings.",
);
}
let database = match self.resolve_redis_database(request.db, connection) {
let database = match self.resolve_redis_database(request.db, &resolved) {
Ok(database) => database,
Err(error) => return error,
};
@@ -641,6 +713,9 @@ impl DbxMcpServer {
description = "Send a base64-encoded message to a Kafka, RocketMQ, or RabbitMQ topic/queue"
)]
async fn send_message(&self, Parameters(request): Parameters<SendMessageRequest>) -> CallToolResult {
if let Err(error) = self.ensure_tool_allowed("dbx_send_message").await {
return error;
}
#[cfg(not(feature = "mq-admin"))]
{
let _ = request;
@@ -702,12 +777,15 @@ impl DbxMcpServer {
#[tool(name = "dbx_get_schema_context", description = "Get compact table and column context for writing SQL")]
async fn get_schema_context(&self, Parameters(request): Parameters<SchemaContextRequest>) -> CallToolResult {
if let Err(error) = self.ensure_tool_allowed("dbx_get_schema_context").await {
return error;
}
let resolved = match self.resolve_connection(&request.selector).await {
Ok(resolved) => resolved,
Err(error) => return error,
};
let connection = &resolved.connection;
let database = match self.resolve_database(request.database, connection) {
let database = match self.resolve_database(request.database, &resolved) {
Ok(database) => database,
Err(error) => return error,
};
@@ -750,6 +828,9 @@ impl DbxMcpServer {
#[tool(name = "dbx_add_connection", description = "Add a new database connection to DBX")]
async fn add_connection(&self, Parameters(request): Parameters<AddConnectionRequest>) -> CallToolResult {
if let Err(error) = self.ensure_tool_allowed("dbx_add_connection").await {
return error;
}
let policy = match self.load_policy().await {
Ok(policy) => policy,
Err(error) => return error,
@@ -804,6 +885,9 @@ impl DbxMcpServer {
&self,
Parameters(request): Parameters<DuplicateConnectionRequest>,
) -> CallToolResult {
if let Err(error) = self.ensure_tool_allowed("dbx_duplicate_connection").await {
return error;
}
let policy = match self.load_policy().await {
Ok(policy) => policy,
Err(error) => return error,
@@ -859,6 +943,9 @@ impl DbxMcpServer {
#[tool(name = "dbx_remove_connection", description = "Remove a database connection from DBX")]
async fn remove_connection(&self, Parameters(request): Parameters<RemoveConnectionRequest>) -> CallToolResult {
if let Err(error) = self.ensure_tool_allowed("dbx_remove_connection").await {
return error;
}
let policy = match self.load_policy().await {
Ok(policy) => policy,
Err(error) => return error,
@@ -876,22 +963,38 @@ impl DbxMcpServer {
let target = if let Some(id) = request.connection_id.as_deref().map(str::trim).filter(|id| !id.is_empty()) {
connections.iter().find(|connection| connection.id == id).cloned()
} else {
let Some(name) = request.connection_name.as_deref().map(str::trim).filter(|name| !name.is_empty()) else {
return tool_error("CONNECTION_NOT_FOUND", "Either connection_id or connection_name is required.");
};
let matching = connections
.iter()
.filter(|connection| connection.name.eq_ignore_ascii_case(&request.connection_name))
.filter(|connection| connection.name.eq_ignore_ascii_case(name))
.cloned()
.collect::<Vec<_>>();
if matching.len() > 1 {
return tool_error("AMBIGUOUS_CONNECTION", ambiguous_connections(&request.connection_name, &matching));
return tool_error("AMBIGUOUS_CONNECTION", ambiguous_connections(name, &matching));
}
matching.into_iter().next()
};
let Some(target) = target else {
return tool_error(
"CONNECTION_NOT_FOUND",
format!("Connection \"{}\" not found.", request.connection_name),
request
.connection_id
.as_deref()
.filter(|id| !id.trim().is_empty())
.map(|id| format!("Connection with id \"{id}\" not found."))
.unwrap_or_else(|| {
format!("Connection \"{}\" not found.", request.connection_name.as_deref().unwrap_or_default())
}),
);
};
if !policy_allows_connection(&policy, &target) {
return tool_error(
"CONNECTION_OUT_OF_SCOPE",
format!("Connection \"{}\" is not allowed by DBX MCP settings.", target.id),
);
}
match self.backend.remove_connection_for_mcp(&target.id).await {
Ok(true) => text(format!("Connection \"{}\" (id: {}) removed.", target.name, target.id)),
Ok(false) => tool_error("CONNECTION_NOT_FOUND", format!("Connection \"{}\" not found.", target.name)),
@@ -901,12 +1004,15 @@ impl DbxMcpServer {
#[tool(name = "dbx_open_table", description = "Open a table in DBX desktop app. Requires DBX to be running.")]
async fn open_table(&self, Parameters(request): Parameters<OpenTableRequest>) -> CallToolResult {
if let Err(error) = self.ensure_tool_allowed("dbx_open_table").await {
return error;
}
let resolved = match self.resolve_connection(&request.selector).await {
Ok(resolved) => resolved,
Err(error) => return error,
};
let connection = &resolved.connection;
let database = match self.resolve_database(request.database, connection) {
let database = match self.resolve_database(request.database, &resolved) {
Ok(database) => database,
Err(error) => return error,
};
@@ -938,6 +1044,9 @@ impl DbxMcpServer {
description = "Execute a SQL query in DBX desktop app UI and show results there. Requires DBX to be running."
)]
async fn execute_and_show(&self, Parameters(request): Parameters<ExecuteAndShowRequest>) -> CallToolResult {
if let Err(error) = self.ensure_tool_allowed("dbx_execute_and_show").await {
return error;
}
let resolved = match self.resolve_connection(&request.selector).await {
Ok(resolved) => resolved,
Err(error) => return error,
@@ -946,10 +1055,16 @@ impl DbxMcpServer {
if connection.db_type == DatabaseType::Redis {
return tool_error("REDIS_COMMAND_REQUIRED", "Use dbx_execute_redis_command for Redis connections.");
}
let database = match self.resolve_database(request.database, connection) {
let database = match self.resolve_database(request.database, &resolved) {
Ok(database) => database,
Err(error) => return error,
};
if connection.db_type != DatabaseType::MongoDb {
if let Err(error) = ensure_sql_database_scope(&resolved.database_scope, connection, &database, &request.sql)
{
return error;
}
}
let permissions = if connection.db_type == DatabaseType::MongoDb {
mcp_permissions(connection, &resolved.policy)
} else {
@@ -959,7 +1074,9 @@ impl DbxMcpServer {
}
};
if connection.db_type == DatabaseType::MongoDb {
if let Err(error) = validate_mongo_command(connection, &resolved.policy, &database, &request.sql) {
if let Err(error) =
validate_mongo_command(connection, &resolved.policy, &resolved.database_scope, &database, &request.sql)
{
return error;
}
}
@@ -985,6 +1102,20 @@ impl DbxMcpServer {
}
impl DbxMcpServer {
async fn list_databases_for_resolved(&self, resolved: &ResolvedConnection) -> CallToolResult {
match &resolved.database_scope {
DatabaseScope::None => tool_error(
"DATABASE_OUT_OF_SCOPE",
"This connection is configured with no database access in DBX MCP settings.",
),
DatabaseScope::Selected(databases) => text(format_database_names(databases)),
DatabaseScope::All => match self.backend.list_databases(&resolved.connection).await {
Ok(databases) => text(format_database_names(&databases)),
Err(error) => tool_error("DATABASE_LIST_ERROR", error),
},
}
}
async fn load_scoped_connections(&self) -> Result<Vec<dbx_core::models::connection::ConnectionConfig>, String> {
let policy = self.backend.load_mcp_global_policy().await?;
let connections = self.backend.load_connections().await?;
@@ -999,15 +1130,40 @@ impl DbxMcpServer {
self.backend.load_mcp_global_policy().await.map_err(|error| backend_tool_error("MCP_POLICY_UNAVAILABLE", error))
}
async fn ensure_tool_allowed(&self, tool_name: &str) -> Result<(), CallToolResult> {
let policy = self.load_policy().await?;
if policy_allows_tool(&policy, tool_name) {
Ok(())
} else {
Err(tool_error("TOOL_OUT_OF_SCOPE", format!("Tool \"{tool_name}\" is not allowed by DBX MCP settings.")))
}
}
/// The advertised `tools/list` view: router-enabled tools narrowed to the
/// global policy's tool allowlist. When the policy cannot be loaded every
/// tool call already fails with `MCP_POLICY_UNAVAILABLE`, so fall back to
/// the plain router view rather than adding a new failure mode here.
async fn policy_filtered_tools(&self) -> Vec<rmcp::model::Tool> {
match self.load_policy().await {
Ok(policy) => self
.tool_router
.list_all()
.into_iter()
.filter(|tool| policy_allows_tool(&policy, tool.name.as_ref()))
.collect(),
Err(_) => self.tool_router.list_all(),
}
}
// CallToolResult is the rmcp wire response type; keeping it unboxed avoids conversions at every tool boundary.
#[allow(clippy::result_large_err)]
fn resolve_database(
&self,
requested: Option<String>,
connection: &dbx_core::models::connection::ConnectionConfig,
resolved: &ResolvedConnection,
) -> Result<String, CallToolResult> {
let requested = requested.map(|database| database.trim().to_string()).filter(|database| !database.is_empty());
if let Some(scoped) = self.scope.database.as_deref() {
let database = if let Some(scoped) = self.scope.database.as_deref() {
if let Some(requested) = requested.as_deref() {
if requested != scoped {
return Err(tool_error(
@@ -1016,9 +1172,12 @@ impl DbxMcpServer {
));
}
}
return Ok(scoped.to_string());
}
Ok(requested.or_else(|| connection.database.clone()).unwrap_or_default())
scoped.to_string()
} else {
requested.or_else(|| resolved.connection.database.clone()).unwrap_or_default()
};
ensure_database_in_scope(&resolved.database_scope, &database)?;
Ok(database)
}
/// Resolve the schema for scoped CLI agents. A selected schema is a hard
@@ -1045,7 +1204,7 @@ impl DbxMcpServer {
fn resolve_redis_database(
&self,
requested: Option<u32>,
connection: &dbx_core::models::connection::ConnectionConfig,
resolved: &ResolvedConnection,
) -> Result<u32, CallToolResult> {
if let Some(scoped) = self.scope.database.as_deref() {
let scoped_database = parse_redis_database(scoped).ok_or_else(|| {
@@ -1062,9 +1221,12 @@ impl DbxMcpServer {
));
}
}
ensure_database_in_scope(&resolved.database_scope, &scoped_database.to_string())?;
return Ok(scoped_database);
}
Ok(requested.or_else(|| redis_database(connection)).unwrap_or(0))
let database = requested.or_else(|| redis_database(&resolved.connection)).unwrap_or(0);
ensure_database_in_scope(&resolved.database_scope, &database.to_string())?;
Ok(database)
}
async fn resolve_connection(&self, selector: &ConnectionSelector) -> Result<ResolvedConnection, CallToolResult> {
@@ -1088,7 +1250,7 @@ impl DbxMcpServer {
format!("Connection \"{id}\" is not allowed by DBX MCP settings."),
));
}
return Ok(ResolvedConnection { connection, policy });
return Ok(resolved_connection(policy, connection));
}
if self.scope.connection_scope_enabled() {
let connection = connections
@@ -1109,7 +1271,7 @@ impl DbxMcpServer {
"The DBX AI session scope is outside the global MCP connection allowlist.",
));
}
return Ok(ResolvedConnection { connection, policy });
return Ok(resolved_connection(policy, connection));
}
let Some(name) = selector.connection_name.as_deref().map(str::trim).filter(|name| !name.is_empty()) else {
return Err(tool_error("CONNECTION_NOT_FOUND", "Either connection_id or connection_name is required."));
@@ -1129,7 +1291,7 @@ impl DbxMcpServer {
"CONNECTION_OUT_OF_SCOPE",
format!("Connection \"{name}\" is not allowed by DBX MCP settings."),
)),
[connection] => Ok(ResolvedConnection { connection: connection.clone(), policy }),
[connection] => Ok(resolved_connection(policy, connection.clone())),
_ => Err(tool_error("AMBIGUOUS_CONNECTION", ambiguous_connections(name, &allowed))),
}
}
@@ -1142,6 +1304,21 @@ impl ServerHandler for DbxMcpServer {
.with_server_info(Implementation::new("dbx", env!("CARGO_PKG_VERSION")))
.with_instructions("Use DBX connections to inspect schemas and query databases safely.")
}
/// Hide tools the global policy disallows from the advertised list, the
/// same way scope-based route disabling does. Per-call enforcement stays
/// in `ensure_tool_allowed`; this only stops clients (especially LLMs)
/// from seeing — and repeatedly retrying — capabilities that would be
/// rejected anyway. When the policy cannot be loaded, every call already
/// fails with `MCP_POLICY_UNAVAILABLE`, so the listing falls back to the
/// router view instead of adding a new failure mode here.
async fn list_tools(
&self,
_request: Option<rmcp::model::PaginatedRequestParams>,
_context: rmcp::service::RequestContext<rmcp::service::RoleServer>,
) -> Result<rmcp::model::ListToolsResult, rmcp::ErrorData> {
Ok(rmcp::model::ListToolsResult { tools: self.policy_filtered_tools().await, meta: None, next_cursor: None })
}
}
fn text(value: impl Into<String>) -> CallToolResult {
@@ -1188,6 +1365,101 @@ fn policy_allows_connection(
policy.allowed_connection_ids.as_ref().is_none_or(|allowed| allowed.iter().any(|id| id == &connection.id))
}
fn policy_allows_tool(policy: &McpGlobalPolicy, tool_name: &str) -> bool {
policy.allowed_tool_names.as_ref().is_none_or(|allowed| allowed.iter().any(|name| name == tool_name))
}
fn database_scope_for_connection(
policy: &McpGlobalPolicy,
connection: &dbx_core::models::connection::ConnectionConfig,
) -> DatabaseScope {
let Some(rule) = policy.connection_policies.iter().find(|rule| rule.connection_id == connection.id) else {
return DatabaseScope::All;
};
match rule.database_scope {
McpDatabaseScope::All => DatabaseScope::All,
McpDatabaseScope::Selected => DatabaseScope::Selected(rule.allowed_databases.clone()),
McpDatabaseScope::None => DatabaseScope::None,
}
}
fn resolved_connection(
policy: McpGlobalPolicy,
connection: dbx_core::models::connection::ConnectionConfig,
) -> ResolvedConnection {
let database_scope = database_scope_for_connection(&policy, &connection);
let policy = effective_policy_for_connection(policy, &connection);
ResolvedConnection { connection, policy, database_scope }
}
#[allow(clippy::result_large_err)]
fn ensure_database_in_scope(scope: &DatabaseScope, database: &str) -> Result<(), CallToolResult> {
match scope {
DatabaseScope::All => Ok(()),
DatabaseScope::Selected(allowed) if allowed.iter().any(|allowed| allowed == database) => Ok(()),
DatabaseScope::Selected(_) | DatabaseScope::None => Err(tool_error(
"DATABASE_OUT_OF_SCOPE",
format!("Database \"{database}\" is not allowed by DBX MCP settings for this connection."),
)),
}
}
/// A selected database scope applies to SQL object references as well as the
/// request-level `database` field. Without this guard an MCP client could pick
/// an allowed default and access another MySQL/SQL Server database through a
/// qualified name such as `other_db.users`.
#[allow(clippy::result_large_err)]
fn ensure_sql_database_scope(
scope: &DatabaseScope,
connection: &dbx_core::models::connection::ConnectionConfig,
active_database: &str,
sql: &str,
) -> Result<(), CallToolResult> {
let DatabaseScope::Selected(allowed_databases) = scope else {
return Ok(());
};
if sql_references_disallowed_database(sql, &connection.db_type, active_database, allowed_databases) {
return Err(tool_error(
"DATABASE_OUT_OF_SCOPE",
"SQL references a database that is not allowed by DBX MCP settings for this connection.",
));
}
Ok(())
}
fn is_database_discovery_sql(sql: &str) -> bool {
let normalized = sql.trim().trim_end_matches(';').trim().to_ascii_lowercase();
matches!(normalized.as_str(), "show databases" | "show schemas")
}
fn format_database_names(databases: &[String]) -> String {
if databases.is_empty() {
"No databases are available through this MCP connection.".to_string()
} else {
databases.iter().map(|database| format!("- {database}")).collect::<Vec<_>>().join("\n")
}
}
/// A connection-level rule is a restrictive overlay, never an escalation. It
/// therefore composes safely with the global policy, the connection's own
/// read-only flag, and production-database protections checked elsewhere.
fn effective_policy_for_connection(
mut policy: McpGlobalPolicy,
connection: &dbx_core::models::connection::ConnectionConfig,
) -> McpGlobalPolicy {
if let Some(rule) = policy.connection_policies.iter().find(|rule| rule.connection_id == connection.id) {
if rule.execution_mode_configured {
policy.read_only |= rule.read_only;
policy.allow_dangerous_sql &= !rule.read_only && rule.allow_dangerous_sql;
}
}
// This returned value is carried through the request only; retaining a
// complete policy document here could accidentally be reused for another
// connection by a future caller.
policy.connection_policies.clear();
policy
}
fn mcp_permissions(
connection: &dbx_core::models::connection::ConnectionConfig,
policy: &McpGlobalPolicy,
@@ -1261,6 +1533,7 @@ fn validate_sql_policy(
fn validate_mongo_command(
connection: &dbx_core::models::connection::ConnectionConfig,
policy: &McpGlobalPolicy,
database_scope: &DatabaseScope,
database: &str,
source: &str,
) -> Result<MongoCommand, CallToolResult> {
@@ -1278,6 +1551,11 @@ fn validate_mongo_command(
"MongoDB runCommand is not available through MCP; review and execute it manually in DBX.",
));
}
if let MongoCommand::Aggregate { pipeline, .. } = &command {
for target_database in mongo_aggregate_target_databases(pipeline, database)? {
ensure_database_in_scope(database_scope, &target_database)?;
}
}
let permissions = mcp_permissions(connection, policy);
let production_database = match &command {
MongoCommand::Aggregate { pipeline, .. } => {
@@ -1309,6 +1587,40 @@ fn validate_mongo_command(
Ok(command)
}
#[allow(clippy::result_large_err)]
fn mongo_aggregate_target_databases(pipeline: &str, active_database: &str) -> Result<Vec<String>, CallToolResult> {
let stages = serde_json::from_str::<serde_json::Value>(pipeline)
.ok()
.and_then(|value| value.as_array().cloned())
.ok_or_else(|| tool_error("QUERY_ERROR", "MongoDB aggregate pipeline must be a JSON array."))?;
let mut databases = Vec::new();
for stage in stages {
let Some(stage) = stage.as_object() else { continue };
for key in ["$out", "$merge"] {
let Some(target) = stage.get(key) else { continue };
let target_database = match target {
serde_json::Value::String(_) => active_database.to_string(),
serde_json::Value::Object(target) => target
.get("db")
.or_else(|| {
target.get("into").and_then(serde_json::Value::as_object).and_then(|into| into.get("db"))
})
.and_then(serde_json::Value::as_str)
.unwrap_or(active_database)
.to_string(),
_ => {
return Err(tool_error(
"QUERY_ERROR",
"MongoDB aggregate output target must be a string or object.",
))
}
};
databases.push(target_database);
}
}
Ok(databases)
}
fn non_empty_env(name: &str) -> Option<String> {
std::env::var(name).ok().map(|value| value.trim().to_string()).filter(|value| !value.is_empty())
}
@@ -1462,6 +1774,7 @@ mod tests {
struct FakeBackend {
connections: Vec<ConnectionConfig>,
policy: McpGlobalPolicy,
recorded_arguments: std::sync::Mutex<Vec<(String, serde_json::Value)>>,
closed_sessions: std::sync::Mutex<Vec<String>>,
pinned_sessions: std::sync::Mutex<HashSet<String>>,
@@ -1472,6 +1785,7 @@ mod tests {
fn default() -> Self {
Self {
connections: Vec::new(),
policy: McpGlobalPolicy::default(),
recorded_arguments: std::sync::Mutex::new(Vec::new()),
closed_sessions: std::sync::Mutex::new(Vec::new()),
pinned_sessions: std::sync::Mutex::new(HashSet::new()),
@@ -1495,6 +1809,10 @@ mod tests {
.unwrap()
}
fn resolved_connection_for_test(connection: ConnectionConfig) -> ResolvedConnection {
ResolvedConnection { connection, policy: McpGlobalPolicy::default(), database_scope: DatabaseScope::All }
}
fn result_text(result: &CallToolResult) -> &str {
result.content[0].as_text().expect("text tool result").text.as_str()
}
@@ -1510,7 +1828,7 @@ mod tests {
#[async_trait]
impl DbxBackend for FakeBackend {
async fn load_mcp_global_policy(&self) -> Result<McpGlobalPolicy, String> {
Ok(McpGlobalPolicy::default())
Ok(self.policy.clone())
}
async fn load_connections(&self) -> Result<Vec<ConnectionConfig>, String> {
@@ -1605,10 +1923,11 @@ mod tests {
let tools = server.tool_router.list_all();
let names = tools.iter().map(|tool| tool.name.as_ref()).collect::<Vec<_>>();
#[cfg(feature = "mq-admin")]
assert_eq!(tools.len(), 14);
assert_eq!(tools.len(), 15);
#[cfg(not(feature = "mq-admin"))]
assert_eq!(tools.len(), 13);
assert_eq!(tools.len(), 14);
assert!(names.contains(&"dbx_list_connections"));
assert!(names.contains(&"dbx_list_databases"));
assert!(names.contains(&"dbx_list_tables"));
assert!(names.contains(&"dbx_describe_table"));
assert!(names.contains(&"dbx_execute_query"));
@@ -1625,6 +1944,33 @@ mod tests {
assert!(names.contains(&"dbx_send_message"));
}
#[tokio::test]
async fn tools_list_hides_tools_the_policy_disallows() {
let policy = McpGlobalPolicy {
allowed_tool_names: Some(vec!["dbx_list_connections".to_string(), "dbx_execute_query".to_string()]),
..Default::default()
};
let server = DbxMcpServer::with_runtime_options(
Arc::new(FakeBackend { policy, ..Default::default() }),
McpScope::default(),
false,
);
let tools = server.policy_filtered_tools().await;
let names = tools.iter().map(|tool| tool.name.as_ref()).collect::<Vec<_>>();
assert_eq!(names, vec!["dbx_execute_query", "dbx_list_connections"]);
}
#[tokio::test]
async fn tools_list_keeps_the_full_view_without_an_allowlist() {
let server = DbxMcpServer::with_runtime_options(Arc::new(FakeBackend::default()), McpScope::default(), false);
let tools = server.policy_filtered_tools().await;
assert_eq!(tools.len(), server.tool_router.list_all().len());
}
#[test]
fn schema_context_tables_schema_is_gemini_compatible() {
let server = DbxMcpServer::with_runtime_options(Arc::new(FakeBackend::default()), McpScope::default(), false);
@@ -1699,7 +2045,7 @@ mod tests {
("dbx_open_table", &["database", "schema"]),
("dbx_execute_and_show", &["database"]),
("dbx_add_connection", &["port", "database", "driver_profile"]),
("dbx_remove_connection", &["connection_id"]),
("dbx_remove_connection", &["connection_id", "connection_name"]),
("dbx_execute_redis_command", &["db"]),
("dbx_get_schema_context", &["database", "schema", "max_tables"]),
];
@@ -1817,9 +2163,9 @@ mod tests {
);
let names = server.tool_router.list_all().into_iter().map(|tool| tool.name).collect::<Vec<_>>();
#[cfg(feature = "mq-admin")]
assert_eq!(names.len(), 9);
assert_eq!(names.len(), 10);
#[cfg(not(feature = "mq-admin"))]
assert_eq!(names.len(), 8);
assert_eq!(names.len(), 9);
assert!(!names.iter().any(|name| name == "dbx_add_connection"));
assert!(!names.iter().any(|name| name == "dbx_duplicate_connection"));
assert!(!names.iter().any(|name| name == "dbx_remove_connection"));
@@ -1858,9 +2204,10 @@ mod tests {
);
assert_eq!(server.load_scoped_connections().await.unwrap().len(), 1);
assert_eq!(server.resolve_database(None, &scoped).unwrap(), "analytics");
assert_eq!(server.resolve_database(Some("analytics".to_string()), &scoped).unwrap(), "analytics");
let error = server.resolve_database(Some("production".to_string()), &scoped).unwrap_err();
let resolved = resolved_connection_for_test(scoped.clone());
assert_eq!(server.resolve_database(None, &resolved).unwrap(), "analytics");
assert_eq!(server.resolve_database(Some("analytics".to_string()), &resolved).unwrap(), "analytics");
let error = server.resolve_database(Some("production".to_string()), &resolved).unwrap_err();
assert!(result_text(&error).contains("DATABASE_OUT_OF_SCOPE"));
let names = server.tool_router.list_all().into_iter().map(|tool| tool.name).collect::<Vec<_>>();
@@ -1868,6 +2215,78 @@ mod tests {
assert!(!names.iter().any(|name| name == "dbx_execute_and_show"));
}
#[test]
fn configured_database_allowlist_is_enforced_for_sql_and_redis_targets() {
let sql = connection("sql", "sql", "mysql", "reporting");
let selected = ResolvedConnection {
connection: sql,
policy: McpGlobalPolicy::default(),
database_scope: DatabaseScope::Selected(vec!["reporting".to_string()]),
};
let server = DbxMcpServer::with_runtime_options(Arc::new(FakeBackend::default()), McpScope::default(), false);
assert_eq!(server.resolve_database(Some("reporting".to_string()), &selected).unwrap(), "reporting");
let error = server.resolve_database(Some("production".to_string()), &selected).unwrap_err();
assert!(result_text(&error).contains("DATABASE_OUT_OF_SCOPE"));
let redis = ResolvedConnection {
connection: connection("redis", "redis", "redis", "0"),
policy: McpGlobalPolicy::default(),
database_scope: DatabaseScope::Selected(vec!["2".to_string()]),
};
assert_eq!(server.resolve_redis_database(Some(2), &redis).unwrap(), 2);
let error = server.resolve_redis_database(Some(3), &redis).unwrap_err();
assert!(result_text(&error).contains("DATABASE_OUT_OF_SCOPE"));
}
#[test]
fn configured_database_allowlist_blocks_qualified_sql_references() {
let connection = connection("sql", "sql", "mysql", "reporting");
let scope = DatabaseScope::Selected(vec!["reporting".to_string()]);
assert!(ensure_sql_database_scope(&scope, &connection, "reporting", "SELECT * FROM reporting.users").is_ok());
let error =
ensure_sql_database_scope(&scope, &connection, "reporting", "SELECT * FROM production.users").unwrap_err();
assert!(result_text(&error).contains("DATABASE_OUT_OF_SCOPE"));
}
#[tokio::test]
async fn selected_database_scope_can_be_discovered_without_a_connection_default_database() {
let connection = connection("sql", "sql", "mysql", "");
let backend = Arc::new(FakeBackend {
connections: vec![connection],
policy: McpGlobalPolicy {
allowed_connection_ids: Some(vec!["sql".to_string()]),
connection_policies: vec![dbx_core::storage::McpConnectionPolicy {
connection_id: "sql".to_string(),
read_only: false,
allow_dangerous_sql: false,
database_scope: McpDatabaseScope::Selected,
allowed_databases: vec!["aa".to_string(), "aaa".to_string(), "abc".to_string()],
execution_mode_configured: false,
}],
..Default::default()
},
..Default::default()
});
let server = DbxMcpServer::with_runtime_options(backend, McpScope::default(), false);
let result = server.list_databases(Parameters(ListDatabasesRequest { selector: selector("sql") })).await;
assert_eq!(result_text(&result), "- aa\n- aaa\n- abc");
let show_databases = server
.execute_query(Parameters(ExecuteQueryRequest {
selector: selector("sql"),
database: None,
sql: "SHOW DATABASES".to_string(),
session_id: None,
cell_char_offset: None,
cell_char_limit: None,
}))
.await;
assert_eq!(result_text(&show_databases), "- aa\n- aaa\n- abc");
assert!(is_database_discovery_sql(" SHOW DATABASES; "));
assert!(is_database_discovery_sql("show schemas"));
assert!(!is_database_discovery_sql("show tables"));
}
#[test]
fn schema_scope_is_a_hard_bound() {
let dameng = connection("dameng-1", "Dameng", "dameng", "APPDB");
@@ -1881,7 +2300,8 @@ mod tests {
false,
);
assert_eq!(server.resolve_database(None, &dameng).unwrap(), "APPDB");
let resolved = resolved_connection_for_test(dameng.clone());
assert_eq!(server.resolve_database(None, &resolved).unwrap(), "APPDB");
assert_eq!(server.resolve_schema(None).unwrap(), "REPORTING");
assert_eq!(server.resolve_schema(Some("REPORTING".to_string())).unwrap(), "REPORTING");
let error = server.resolve_schema(Some("APP_USER".to_string())).unwrap_err();
@@ -1896,8 +2316,9 @@ mod tests {
McpScope { database: Some("2".to_string()), ..Default::default() },
false,
);
assert_eq!(scoped.resolve_redis_database(None, &redis).unwrap(), 2);
let error = scoped.resolve_redis_database(Some(3), &redis).unwrap_err();
let resolved = resolved_connection_for_test(redis.clone());
assert_eq!(scoped.resolve_redis_database(None, &resolved).unwrap(), 2);
let error = scoped.resolve_redis_database(Some(3), &resolved).unwrap_err();
assert!(result_text(&error).contains("DATABASE_OUT_OF_SCOPE"));
let invalid = DbxMcpServer::with_runtime_options(
@@ -1905,7 +2326,7 @@ mod tests {
McpScope { database: Some("analytics".to_string()), ..Default::default() },
false,
);
let error = invalid.resolve_redis_database(None, &redis).unwrap_err();
let error = invalid.resolve_redis_database(None, &resolved).unwrap_err();
assert!(result_text(&error).contains("INVALID_DATABASE_SCOPE"));
}
@@ -1917,21 +2338,27 @@ mod tests {
read_only: false,
allow_dangerous_sql: true,
allowed_connection_ids: None,
query_timeout_secs: None,
..Default::default()
};
let error = validate_mongo_command(
&mongo,
&policy,
&DatabaseScope::All,
"staging",
r#"db.items.aggregate([{"$out":{"db":"production","coll":"archive"}}])"#,
)
.unwrap_err();
assert!(result_text(&error).contains("PRODUCTION_WRITE_BLOCKED"));
assert!(
validate_mongo_command(&mongo, &policy, "staging", r#"db.items.aggregate([{"$out":"archive"}])"#,).is_ok()
);
assert!(validate_mongo_command(
&mongo,
&policy,
&DatabaseScope::All,
"staging",
r#"db.items.aggregate([{"$out":"archive"}])"#,
)
.is_ok());
}
#[test]
@@ -1944,22 +2371,22 @@ mod tests {
read_only: true,
allow_dangerous_sql: false,
allowed_connection_ids: None,
query_timeout_secs: None,
..Default::default()
},
McpGlobalPolicy {
read_only: false,
allow_dangerous_sql: false,
allowed_connection_ids: None,
query_timeout_secs: None,
..Default::default()
},
McpGlobalPolicy {
read_only: false,
allow_dangerous_sql: true,
allowed_connection_ids: None,
query_timeout_secs: None,
..Default::default()
},
] {
let error = validate_mongo_command(&mongo, &policy, "staging", source).unwrap_err();
let error = validate_mongo_command(&mongo, &policy, &DatabaseScope::All, "staging", source).unwrap_err();
assert!(result_text(&error).contains("SQL_BLOCKED"));
assert!(result_text(&error).contains("runCommand"));
}
@@ -1990,7 +2417,7 @@ mod tests {
read_only: false,
allow_dangerous_sql: true,
allowed_connection_ids: None,
query_timeout_secs: None,
..Default::default()
};
let read_only_error =
validate_sql_policy(&read_only, &writable_policy, "app", "DELETE FROM sessions", false).unwrap_err();
@@ -2045,7 +2472,7 @@ mod tests {
read_only: false,
allow_dangerous_sql: false,
allowed_connection_ids: None,
query_timeout_secs: None,
..Default::default()
};
let permissions = mcp_permissions(&connection, &safe_write);
assert!(!permissions.allow_dangerous, "confirmed SQL must NOT elevate allow_dangerous for Redis/Mongo paths");
@@ -2072,7 +2499,7 @@ mod tests {
read_only: false,
allow_dangerous_sql: true,
allowed_connection_ids: None,
query_timeout_secs: None,
..Default::default()
};
let permissions =
validate_sql_policy(&connection, &full_access, "app", "CREATE TABLE metrics (id INT)", false).unwrap();
@@ -2086,7 +2513,7 @@ mod tests {
read_only: true,
allow_dangerous_sql: false,
allowed_connection_ids: None,
query_timeout_secs: None,
..Default::default()
};
let error = validate_sql_policy(&connection, &read_only_policy, "app", "CREATE TABLE metrics (id INT)", false)
.unwrap_err();
@@ -2111,7 +2538,7 @@ mod tests {
read_only: false,
allow_dangerous_sql: false,
allowed_connection_ids: None,
query_timeout_secs: None,
..Default::default()
};
let blocked = validate_sql_policy(&starrocks, &policy, "default_catalog", "USE analytics", false).unwrap_err();