fix(plugins): stabilize plugin integration

This commit is contained in:
Jinpy authored and GitHub committed 2026-09-14 21:03:47 +08:00
1 parent b70d6da227
commit 3d1b84831f
76 files changed
+3846 -431

No files matched your search

@@ -0,0 +1,29 @@
//! One-shot plugin installer for local DBX app stores:
//! cargo run -p dbx-core --example install_plugin -- <plugins-root> <dbxp> <app-version> [--rollback-plugin <id>]
use dbx_core::plugins::{PluginInstallPolicy, PluginPackageInstaller};
fn main() {
let args: Vec<String> = std::env::args().skip(1).collect();
if args.len() < 3 {
eprintln!("usage: install_plugin <plugins-root> <package.dbxp> <app-version>");
std::process::exit(2);
}
let installer = PluginPackageInstaller::new(std::path::PathBuf::from(&args[0]), &args[2]).expect("installer");
match installer.install_file(std::path::Path::new(&args[1]), PluginInstallPolicy::LocalDevelopment) {
Ok(result) => {
let response = result.response();
println!(
"installed {} v{} (previous: {:?}, signature: {:?}, sha256: {})",
response.plugin.manifest.id,
response.plugin.manifest.version,
response.previous_version,
response.signature,
response.package_sha256
);
}
Err(error) => {
eprintln!("install failed: {error}");
std::process::exit(1);
}
}
}
+12 -11
View File
@@ -13,9 +13,9 @@ use std::net::{IpAddr, Ipv4Addr};
use crate::ai::AiConfigItem;
use crate::connection_secrets::{
CASSANDRA_KEYSTORE_PASSWORD_KEY, CASSANDRA_TRUSTSTORE_PASSWORD_KEY, MQ_AUTH_API_KEY_VALUE_KEY,
MQ_AUTH_CLIENT_SECRET_KEY, MQ_AUTH_PASSWORD_KEY, MQ_AUTH_TOKEN_KEY, MQ_TOKEN_SIGNING_KEY, NACOS_AUTH_PASSWORD_KEY,
NACOS_RNACOS_CONSOLE_PASSWORD_KEY, PLUGIN_CONNECTION_SECRET_PREFIX,
plugin_connection_secret_key, CASSANDRA_KEYSTORE_PASSWORD_KEY, CASSANDRA_TRUSTSTORE_PASSWORD_KEY,
MQ_AUTH_API_KEY_VALUE_KEY, MQ_AUTH_CLIENT_SECRET_KEY, MQ_AUTH_PASSWORD_KEY, MQ_AUTH_TOKEN_KEY,
MQ_TOKEN_SIGNING_KEY, NACOS_AUTH_PASSWORD_KEY, NACOS_RNACOS_CONSOLE_PASSWORD_KEY, PLUGIN_CONNECTION_SECRET_PREFIX,
};
use crate::models::connection::{ConnectionConfig, DatabaseType, TransportLayerConfig};
use crate::saved_sql::SavedSqlLibrary;
@@ -970,6 +970,9 @@ fn scrub_connection_secrets(config: &mut ConnectionConfig) {
config.connection_string = None;
config.init_script = None;
scrub_mqtt_auth_secrets(config);
for secret in config.connection_secrets.values_mut() {
secret.clear();
}
scrub_mq_external_config_secrets(config);
scrub_nacos_auth_secrets(config);
scrub_cassandra_tls_secrets(config);
@@ -1009,14 +1012,6 @@ async fn build_sensitive_payload(
if config.save_password {
push_secret(&mut connection_secrets, &config.id, "password", &config.password);
}
for (key, secret) in &config.connection_secrets {
push_secret(
&mut connection_secrets,
&config.id,
&format!("{PLUGIN_CONNECTION_SECRET_PREFIX}{key}"),
secret,
);
}
push_secret(&mut connection_secrets, &config.id, "init_script", config.init_script.as_deref().unwrap_or(""));
for (index, layer) in config.transport_layers.iter().enumerate() {
match layer {
@@ -1060,6 +1055,9 @@ async fn build_sensitive_payload(
push_cassandra_tls_secrets(&mut connection_secrets, config);
if config.save_password {
push_nacos_external_config_secrets(&mut connection_secrets, config);
for (key, secret) in &config.connection_secrets {
push_secret(&mut connection_secrets, &config.id, &plugin_connection_secret_key(key)?, secret);
}
}
}
@@ -2219,6 +2217,7 @@ mod tests {
production_databases: vec![],
database_info: None,
};
config.connection_secrets.insert("api_token".to_string(), "plugin-secret".to_string());
scrub_connection_secrets(&mut config);
assert!(config.password.is_empty());
match &config.transport_layers[0] {
@@ -2235,8 +2234,10 @@ mod tests {
assert!(config.redis_sentinel_password.is_empty());
assert!(config.connection_string.is_none());
assert!(config.init_script.is_none());
assert_eq!(config.connection_secrets.get("api_token").map(String::as_str), None);
let public_json = serde_json::to_string(&config).unwrap();
assert!(!public_json.contains("token-value"));
assert!(!public_json.contains("plugin-secret"));
assert!(super::SECRET_KEYS.contains(&"init_script"));
}
+47 -17
View File
@@ -29,7 +29,10 @@ use crate::models::connection::{
use crate::mongo_oidc::MongoOidcBrowserOpener;
use crate::nacos::config::{NACOS_CONSOLE_SESSION_PASSWORD, NACOS_PRIMARY_SESSION_PASSWORD};
use crate::path_utils::expand_tilde;
use crate::plugins::{PluginConnectionHandle, PluginDriverSession, PluginHost, PluginRegistry, PluginRuntimeEnv};
use crate::plugins::{
PluginConnectionActionResult, PluginConnectionHandle, PluginDriverSession, PluginHost, PluginRegistry,
PluginRuntimeEnv,
};
use crate::query_cancel::RunningQueries;
use crate::session_credentials::SessionCredentialStore;
use crate::storage::{normalize_duckdb_worker_max_processes, Storage, DUCKDB_WORKER_MAX_PROCESSES_DEFAULT};
@@ -1346,8 +1349,10 @@ impl AppState {
agent_dir: PathBuf,
app_version: impl Into<String>,
) -> Self {
let data_dir = storage.data_dir().to_path_buf();
let app_version = app_version.into();
let data_dir = storage.data_dir().to_path_buf();
let plugins = PluginRegistry::new_with_app_version(plugin_dir, app_version.clone());
let plugin_host = PluginHost::new(plugins.clone());
Self {
connections: Arc::new(RwLock::new(ConnectionPoolRegistry::new())),
task_supervisor: TaskSupervisor::new(),
@@ -1360,8 +1365,8 @@ impl AppState {
proxy_tunnels: ProxyTunnelManager::new(),
http_tunnels: HttpTunnelManager::new(),
storage,
plugins: PluginRegistry::new_with_app_version(plugin_dir.clone(), app_version.clone()),
plugin_host: PluginHost::new(PluginRegistry::new_with_app_version(plugin_dir, app_version.clone())),
plugins,
plugin_host,
agent_manager: crate::agent_manager::AgentManager::new_with_base_dir_and_app_version(
agent_dir,
app_version,
@@ -2033,6 +2038,7 @@ impl AppState {
self.proxy_tunnels.stop_all_tunnels(),
self.http_tunnels.stop_all_tunnels(),
self.agent_manager.stop_daemons(),
self.plugin_host.stop_all(),
);
};
if tokio::time::timeout(deadline, shutdown).await.is_err() {
@@ -2204,7 +2210,9 @@ impl AppState {
self.reset_connection_transport_for_config(connection_id, &db_config).await;
return Err(err);
}
probe_connection_endpoint(&db_config, &host, port).await?;
if db_config.db_type != DatabaseType::Plugin {
probe_connection_endpoint(&db_config, &host, port).await?;
}
if let Err(err) = self.ensure_current_connection_attempt(connection_id, connection_attempt).await {
self.reset_connection_transport_for_config(connection_id, &db_config).await;
return Err(err);
@@ -2787,7 +2795,9 @@ impl AppState {
}
self.external_driver_pool("jdbc", &jdbc_config).await?
}
DatabaseType::Plugin => return Err("Plugin-owned connections use the plugin host".to_string()),
DatabaseType::Plugin => {
PoolKind::PluginConnection(self.plugin_host.connect_connection(&db_config, &host, port).await?)
}
#[cfg(feature = "mq-admin")]
DatabaseType::MessageQueue => {
// MQ admin connections don't hold a data query pool. We just test
@@ -3056,6 +3066,28 @@ impl AppState {
Ok(("127.0.0.1".to_string(), local_port))
}
pub async fn invoke_plugin_connection_action(
&self,
config: ConnectionConfig,
action_id: &str,
) -> Result<PluginConnectionActionResult, String> {
if config.db_type != DatabaseType::Plugin {
return Err("Connection is not owned by a plugin".to_string());
}
let config = config.canonicalized();
let transport_id = format!("{}:plugin-action:{action_id}", config.id);
let has_transport_layers = config.has_effective_transport_layers();
let connection_id = if has_transport_layers { transport_id.as_str() } else { config.id.as_str() };
let result = match self.connection_host_port(connection_id, &config).await {
Ok((host, port)) => self.plugin_host.invoke_connection_action(&config, action_id, &host, port).await,
Err(error) => Err(error),
};
if has_transport_layers {
self.reset_connection_transport_for_config(&transport_id, &config).await;
}
result
}
pub async fn connect_redis_sentinel(
&self,
connection_id: &str,
@@ -3838,13 +3870,13 @@ impl AppState {
}
}
}
PoolKind::PluginConnection(handle) => !handle.is_running(),
PoolKind::Sqlite(_)
| PoolKind::DuckDbWorker(_)
| PoolKind::ExternalDriver { .. }
| PoolKind::MessageQueue
| PoolKind::Nacos
| PoolKind::Consul(_) => false,
PoolKind::PluginConnection(handle) => !handle.is_running(),
#[cfg(feature = "mq-admin")]
PoolKind::Mqtt(_) => false,
}
@@ -5002,15 +5034,6 @@ impl AppState {
self.pool_routing_control().close_removed(removed).await;
}
pub async fn invoke_plugin_connection_action(
&self,
config: ConnectionConfig,
action_id: &str,
) -> Result<crate::plugins::PluginConnectionActionResult, String> {
let (host, port) = self.connection_host_port(&config.id, &config).await?;
self.plugin_host.invoke_connection_action(&config, action_id, &host, port).await
}
async fn drain_connection_pools(&self, connection_id: &str) -> Vec<(String, PoolKind)> {
let pool_prefix = format!("{connection_id}:");
let keys_to_remove: Vec<String> = self
@@ -5771,7 +5794,14 @@ async fn close_pool_kind(pool: PoolKind) -> Result<(), String> {
session.shutdown().await;
}
PoolKind::PluginConnection(handle) => {
handle.disconnect().await?;
if let Err(error) = handle.disconnect().await {
log::warn!(
"Failed to disconnect plugin connection '{}' ({}/{}): {error}",
handle.connection_id,
handle.plugin_id,
handle.provider_id
);
}
}
PoolKind::MessageQueue => {}
PoolKind::Nacos => {}
+64 -1
View File
@@ -29,6 +29,16 @@ pub const CASSANDRA_TRUSTSTORE_PASSWORD_KEY: &str = "cassandra.tls.truststore_pa
pub const CASSANDRA_KEYSTORE_PASSWORD_KEY: &str = "cassandra.tls.keystore_password";
pub const PLUGIN_CONNECTION_SECRET_PREFIX: &str = "plugin_connection.";
/// Storage-level secret key for one plugin connection field: namespaced under
/// [`PLUGIN_CONNECTION_SECRET_PREFIX`] so plugin keys never collide with the
/// fixed driver keys and can be scrubbed with a single prefix delete.
pub fn plugin_connection_secret_key(key: &str) -> Result<String, String> {
if key.is_empty() {
return Err("Plugin connection secret key must not be empty".to_string());
}
Ok(format!("{PLUGIN_CONNECTION_SECRET_PREFIX}{key}"))
}
pub trait ConnectionSecretStore {
fn set_secret(&self, connection_id: &str, key: &str, secret: &str) -> Result<(), String>;
fn get_secret(&self, connection_id: &str, key: &str) -> Result<Option<String>, String>;
@@ -117,6 +127,10 @@ pub fn save_connections_to_file(
persist_mq_token_signing_secret(store, config)?;
persist_mqtt_auth_secrets(store, config)?;
persist_cassandra_tls_secrets(store, config)?;
delete_secret_prefix(store, &config.id, PLUGIN_CONNECTION_SECRET_PREFIX)?;
for (key, secret) in &config.connection_secrets {
persist_secret(store, &config.id, &plugin_connection_secret_key(key)?, secret)?;
}
// New configs persist transport-layer secrets only. Remove legacy transport secret slots after the
// migrated layer values have been written so old configs do not keep two sources of truth.
@@ -187,6 +201,19 @@ pub fn load_connections_from_file(
hydrate_mq_token_signing_secret(store, config, &mut needs_rewrite)?;
hydrate_mqtt_auth_secrets(store, config, &mut needs_rewrite)?;
hydrate_cassandra_tls_secrets(store, config, &mut needs_rewrite)?;
let plugin_secret_keys = config.connection_secrets.keys().cloned().collect::<Vec<_>>();
for key in plugin_secret_keys {
let storage_key = plugin_connection_secret_key(&key)?;
let current = config.connection_secrets.get(&key).cloned().unwrap_or_default();
if current.is_empty() {
if let Some(secret) = store.get_secret(&config.id, &storage_key)? {
config.connection_secrets.insert(key, secret);
}
} else {
store.set_secret(&config.id, &storage_key, &current)?;
needs_rewrite = true;
}
}
}
if needs_rewrite {
@@ -818,11 +845,18 @@ fn sanitize_connections(configs: &[ConnectionConfig]) -> Vec<ConnectionConfig> {
scrub_mq_token_signing_secret(&mut config);
scrub_mqtt_auth_secrets(&mut config);
scrub_cassandra_tls_secrets(&mut config);
scrub_plugin_connection_secrets(&mut config);
config
})
.collect()
}
fn scrub_plugin_connection_secrets(config: &mut ConnectionConfig) {
for secret in config.connection_secrets.values_mut() {
secret.clear();
}
}
pub fn secret_account(connection_id: &str, key: &str) -> String {
format!("connection:{connection_id}:{key}")
}
@@ -833,7 +867,7 @@ mod tests {
load_connections_from_file, save_connections_to_file, ConnectionSecretStore, CASSANDRA_KEYSTORE_PASSWORD_KEY,
CASSANDRA_TRUSTSTORE_PASSWORD_KEY, CONNECTION_STRING_KEY, INIT_SCRIPT_KEY, MAIN_PASSWORD_KEY,
MQTT_AUTH_PASSWORD_KEY, MQ_AUTH_PASSWORD_KEY, MQ_AUTH_TOKEN_KEY, MQ_TOKEN_SIGNING_KEY,
REDIS_SENTINEL_PASSWORD_KEY, SSH_PASSWORD_KEY,
PLUGIN_CONNECTION_SECRET_PREFIX, REDIS_SENTINEL_PASSWORD_KEY, SSH_PASSWORD_KEY,
};
use crate::models::connection::{
ConnectionConfig, DatabaseType, HttpTunnelConfig, SshTunnelConfig, TransportLayerConfig,
@@ -1543,4 +1577,33 @@ mod tests {
assert_eq!(store.get_existing("mqtt-broker", MQTT_AUTH_PASSWORD_KEY), None);
}
#[test]
fn save_connections_moves_plugin_secrets_to_secret_store_and_restores_them() {
let path = temp_connections_file("plugin-connection-secret");
let store = MemorySecretStore::default();
let mut config = connection("plugin-connection", "", "");
config.db_type = DatabaseType::Plugin;
config.plugin_id = Some("dbx.example.hello".to_string());
config.plugin_connection_provider = Some("hello.connection".to_string());
config.plugin_connection_type = Some("hello".to_string());
config.external_config = Some(serde_json::json!({ "greeting": "Hello" }));
config.connection_secrets.insert("access_token".to_string(), "plugin-secret".to_string());
save_connections_to_file(&path, &[config], &store).unwrap();
let raw = std::fs::read_to_string(&path).unwrap();
assert!(!raw.contains("plugin-secret"));
let persisted: Vec<ConnectionConfig> = serde_json::from_str(&raw).unwrap();
assert_eq!(persisted[0].connection_secrets.get("access_token").map(String::as_str), Some(""));
assert_eq!(
store
.get_existing("plugin-connection", &format!("{PLUGIN_CONNECTION_SECRET_PREFIX}access_token"))
.as_deref(),
Some("plugin-secret")
);
let loaded = load_connections_from_file(&path, &store).unwrap();
assert_eq!(loaded[0].connection_secrets.get("access_token").map(String::as_str), Some("plugin-secret"));
}
}
+122 -7
View File
@@ -39,11 +39,12 @@ pub use manifest::{
current_plugin_target, resolve_safe_plugin_path, PluginBackendEntrypoint, PluginBackendTransport,
PluginCompatibility, PluginConnectionActionContribution, PluginConnectionActionVariant, PluginConnectionActionWhen,
PluginConnectionCapability, PluginConnectionProviderContribution, PluginContribution, PluginDriverManifest,
PluginEngines, PluginEntrypoints, PluginFilesystemCapability, PluginFilesystemProviderContribution,
PluginFormFieldBinding, PluginFormFieldDefinition, PluginFormFieldOption, PluginFormFieldType, PluginManifest,
PluginUiEntrypoint, PluginWorkbenchContribution, PLUGIN_CONNECTION_ACTION_METHOD, PLUGIN_CONNECTION_CONNECT_METHOD,
PLUGIN_CONNECTION_DISCONNECT_METHOD, PLUGIN_CONNECTION_TEST_METHOD, SUPPORTED_PLUGIN_HOST_API_VERSION,
SUPPORTED_PLUGIN_MANIFEST_VERSION, SUPPORTED_PLUGIN_PERMISSIONS, SUPPORTED_PLUGIN_PROTOCOL_VERSION,
PluginEngines, PluginEntrypoints, PluginFieldCondition, PluginFilesystemCapability,
PluginFilesystemProviderContribution, PluginFormFieldBinding, PluginFormFieldDefinition, PluginFormFieldOption,
PluginFormFieldType, PluginManifest, PluginUiEntrypoint, PluginWorkbenchContribution,
PLUGIN_CONNECTION_ACTION_METHOD, PLUGIN_CONNECTION_CONNECT_METHOD, PLUGIN_CONNECTION_DISCONNECT_METHOD,
PLUGIN_CONNECTION_TEST_METHOD, SUPPORTED_PLUGIN_HOST_API_VERSION, SUPPORTED_PLUGIN_MANIFEST_VERSION,
SUPPORTED_PLUGIN_PERMISSIONS, SUPPORTED_PLUGIN_PROTOCOL_VERSION,
};
pub use runtime::{
PluginBinaryMessage, PluginEvent, PluginHandshake, PluginHandshakeIdentity, PluginSessionState,
@@ -100,12 +101,27 @@ pub struct PluginRuntimeEnv {
vars: Vec<(String, String)>,
}
/// Environment variable that tells a plugin sidecar where to keep its
/// persistent, version-independent local data (preferences, audit logs,
/// known hosts, ...). Injected by the registry so plugins never have to fall
/// back to the OS temp dir, which macOS wipes on reboot.
pub const PLUGIN_DATA_DIR_ENV: &str = "DBX_PLUGIN_DATA_DIR";
impl PluginRuntimeEnv {
pub fn with_var(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
self.vars.push((key.into(), value.into()));
self
}
/// Sets `DBX_PLUGIN_DATA_DIR` unless the caller already provided one, so
/// an explicit override keeps winning over the registry default.
pub fn with_plugin_data_dir(self, data_dir: &Path) -> Self {
if self.get(PLUGIN_DATA_DIR_ENV).is_some() {
return self;
}
self.with_var(PLUGIN_DATA_DIR_ENV, data_dir.to_string_lossy().into_owned())
}
pub fn get(&self, key: &str) -> Option<&str> {
self.vars.iter().find_map(|(name, value)| (name == key).then_some(value.as_str()))
}
@@ -140,6 +156,14 @@ impl PluginRegistry {
&self.app_version
}
/// Persistent data directory for one plugin: `<data dir>/plugin-data/<id>`,
/// a sibling of the `plugins/` registry root so it survives version
/// upgrades and never collides with the installer-managed
/// `versions/` + `activations/` tree.
pub fn plugin_data_dir(&self, plugin_id: &str) -> PathBuf {
self.root_dir.parent().unwrap_or(&self.root_dir).join("plugin-data").join(plugin_id)
}
pub fn list_installed(&self) -> Result<Vec<InstalledPlugin>, String> {
let entries = match std::fs::read_dir(&self.root_dir) {
Ok(entries) => entries,
@@ -241,6 +265,7 @@ impl PluginRegistry {
let plugin =
self.find_driver(driver_id)?.ok_or_else(|| format!("Plugin driver '{driver_id}' is not installed"))?;
ensure_plugin_compatible(&plugin)?;
let env = env.with_plugin_data_dir(&self.plugin_data_dir(&plugin.manifest.id));
let session = PluginSidecarSession::start(plugin, self.app_version.clone(), env).await?;
let result = session.invoke_with_timeout(method, params, Some(driver_id), timeout_duration).await;
session.shutdown().await;
@@ -259,6 +284,7 @@ impl PluginRegistry {
let plugin =
self.find_driver(driver_id)?.ok_or_else(|| format!("Plugin driver '{driver_id}' is not installed"))?;
ensure_plugin_compatible(&plugin)?;
let env = env.with_plugin_data_dir(&self.plugin_data_dir(&plugin.manifest.id));
PluginDriverSession::start(plugin, driver_id.to_string(), self.app_version.clone(), env).await.map(Arc::new)
}
}
@@ -362,9 +388,98 @@ impl PluginDriverSession {
#[cfg(test)]
mod tests {
use super::{InstalledPlugin, PluginContribution, PluginManifest, PluginRegistry};
use std::path::{Path, PathBuf};
use super::{InstalledPlugin, PluginContribution, PluginManifest, PluginRegistry, PluginRuntimeEnv};
#[cfg(unix)]
use super::{PluginDriverManifest, PluginDriverSession, PluginRuntimeEnv};
use super::{PluginDriverManifest, PluginDriverSession};
#[test]
fn plugin_data_dir_lives_beside_the_plugin_registry() {
let registry = PluginRegistry::new(PathBuf::from("/data/plugins"));
assert_eq!(registry.plugin_data_dir("io.dbx.ssh"), PathBuf::from("/data/plugin-data/io.dbx.ssh"));
}
#[test]
fn plugin_data_dir_falls_back_to_registry_root_without_a_parent() {
let registry = PluginRegistry::new(PathBuf::from("/"));
assert_eq!(registry.plugin_data_dir("io.dbx.ssh"), PathBuf::from("/plugin-data/io.dbx.ssh"));
}
#[test]
fn with_plugin_data_dir_sets_the_env_var_once() {
let env = PluginRuntimeEnv::default().with_plugin_data_dir(Path::new("/data/plugin-data/io.dbx.ssh"));
assert_eq!(env.get("DBX_PLUGIN_DATA_DIR"), Some("/data/plugin-data/io.dbx.ssh"));
// An explicit caller-provided value wins over the registry default.
let explicit = PluginRuntimeEnv::default()
.with_var("DBX_PLUGIN_DATA_DIR", "/custom")
.with_plugin_data_dir(Path::new("/ignored"));
assert_eq!(explicit.get("DBX_PLUGIN_DATA_DIR"), Some("/custom"));
}
/// The registry-injected data dir must reach the sidecar process itself,
/// not just the `PluginRuntimeEnv` value object.
#[cfg(unix)]
#[tokio::test]
async fn activated_sidecar_receives_plugin_data_dir_env() {
use std::os::unix::fs::PermissionsExt;
use super::PluginHost;
let data_dir = tempfile::tempdir().unwrap();
let plugin_dir = data_dir.path().join("plugins").join("sample.sidecar");
std::fs::create_dir_all(&plugin_dir).unwrap();
let executable = plugin_dir.join("plugin.sh");
std::fs::write(
&executable,
r#"#!/bin/sh
IFS= read -r initialize
initialize_id=$(printf '%s' "$initialize" | sed -E 's/.*"id":([0-9]+).*/\1/')
printf '{"jsonrpc":"2.0","id":%s,"result":{"protocolVersion":1,"capabilities":[],"plugin":{"id":"sample.sidecar","version":"1.0.0"}}}\n' "$initialize_id"
IFS= read -r request
request_id=$(printf '%s' "$request" | sed -E 's/.*"id":([0-9]+).*/\1/')
printf '{"jsonrpc":"2.0","id":%s,"result":"%s"}\n' "$request_id" "$DBX_PLUGIN_DATA_DIR"
sleep 30
"#,
)
.unwrap();
let mut permissions = std::fs::metadata(&executable).unwrap().permissions();
permissions.set_mode(0o755);
std::fs::set_permissions(&executable, permissions).unwrap();
std::fs::write(
plugin_dir.join("manifest.json"),
serde_json::json!({
"manifest_version": 1,
"id": "sample.sidecar",
"name": "Sample Sidecar",
"version": "1.0.0",
"publisher": "dbx",
"engines": { "dbx": ">=0.1.0", "host_api": "^1.0" },
"permissions": [],
"entrypoints": {
"backend": {
"protocol_versions": [1],
"transport": "stdio-jsonl",
"executable": "plugin.sh"
}
}
})
.to_string(),
)
.unwrap();
let registry = PluginRegistry::new_with_app_version(data_dir.path().join("plugins"), "0.5.67");
let host = PluginHost::new(registry);
let reported: String =
host.invoke("sample.sidecar", "sample/dataDir", serde_json::Value::Null, None, None).await.unwrap();
assert_eq!(
PathBuf::from(reported),
data_dir.path().join("plugin-data").join("sample.sidecar"),
"sidecar must see <data dir>/plugin-data/<id> in DBX_PLUGIN_DATA_DIR"
);
host.stop_all().await;
}
#[cfg(unix)]
#[tokio::test]
+421 -3
View File
@@ -133,6 +133,7 @@ impl PluginHost {
if !plugin.compatibility.compatible {
return Err(format!("Plugin '{plugin_id}' is incompatible: {}", plugin.compatibility.errors.join("; ")));
}
let env = env.with_plugin_data_dir(&self.inner.registry.plugin_data_dir(&plugin.manifest.id));
let session = PluginSidecarSession::start(plugin, self.inner.registry.app_version().to_string(), env).await?;
self.forward_session_events(&session);
self.inner.sessions.write().await.insert(plugin_id.to_string(), session.clone());
@@ -179,6 +180,15 @@ impl PluginHost {
session.send_binary(channel, data).await
}
/// Builds the standard plugin connection lifecycle payload for a config
/// using the config endpoint as the runtime endpoint (no desktop tunnel
/// available, e.g. the MCP CLI bridge). Validates that the config maps
/// to an installed connection provider.
pub fn connection_params_standalone(&self, config: &ConnectionConfig) -> Result<serde_json::Value, String> {
let (_, provider) = self.resolve_connection_provider(config)?;
plugin_connection_params(config, &provider, &config.host, config.port)
}
pub async fn test_connection(
&self,
config: &ConnectionConfig,
@@ -390,7 +400,8 @@ fn plugin_connection_params(
"runtime": {
"host": runtime_host,
"port": runtime_port,
}
},
"operationId": uuid::Uuid::new_v4().to_string(),
}))
}
@@ -430,10 +441,22 @@ fn validate_plugin_connection_values_for_action(
}
for field in &provider.fields {
let value = plugin_connection_field_value(config, field);
if require_required_fields && field.required && plugin_field_value_is_empty(value.as_ref()) {
// Required enforcement is condition-aware (Host API 1.1): fields hidden
// by `visible_when` or only conditionally required via `required_when`
// are validated against the current stored values, mirroring the
// connection dialog's `pluginFieldConditions` semantics. This keeps
// non-dialog write paths (MCP/import) from unconditionally rejecting
// connections whose protocol-specific fields are simply not visible.
if require_required_fields
&& plugin_field_is_visible(field, config, provider)
&& plugin_field_is_required(field, config, provider)
&& plugin_field_value_is_empty(value.as_ref())
{
return Err(format!("Plugin connection field '{}' is required", field.label));
}
if let Some(value) = value {
// A stored JSON null means "unset" (older dialog builds wrote nulls for
// untouched optional fields); it must not fail the declared type check.
if let Some(value) = value.filter(|value| !value.is_null()) {
validate_plugin_field_type(field, &value)?;
if field.effective_binding() == PluginFormFieldBinding::Port
&& value.as_u64().is_none_or(|port| port == 0 || port > u16::MAX as u64)
@@ -523,6 +546,89 @@ fn plugin_field_value_is_empty(value: Option<&serde_json::Value>) -> bool {
}
}
/// Evaluates a plugin manifest field condition (`visible_when` / `required_when`,
/// Host API 1.1) against the stored connection values. Mirrors the frontend
/// `pluginFieldConditions.ts` semantics: the condition matches when the current
/// value of the referenced sibling field is listed in `one_of`; a missing
/// sibling field or an unset/empty value never matches.
fn plugin_field_condition_matches(
condition: &super::PluginFieldCondition,
config: &ConnectionConfig,
provider: &PluginConnectionProviderContribution,
) -> bool {
let Some(sibling) = provider.fields.iter().find(|field| field.key == condition.field) else {
return false;
};
let Some(value) = plugin_connection_field_value(config, sibling) else {
return false;
};
let text = match value {
serde_json::Value::Null => return false,
serde_json::Value::String(text) => text,
other => other.to_string(),
};
if text.trim().is_empty() {
return false;
}
condition.one_of.iter().any(|option| option == &text)
}
/// A field participates in the form when its `visible_when` (if any) matches.
/// Mirrors the frontend cascade (`pluginFieldConditions.ts`): once the raw
/// condition matches, the referenced sibling must itself be visible — a hidden
/// container field's stored default (e.g. `krb_credential_type: "password"`
/// while auth is simple) must not mark grandchild fields visible + required,
/// otherwise non-dialog write paths reject connections the dialog accepts.
fn plugin_field_is_visible(
field: &PluginFormFieldDefinition,
config: &ConnectionConfig,
provider: &PluginConnectionProviderContribution,
) -> bool {
let mut seen = HashSet::new();
seen.insert(field.key.clone());
plugin_field_is_visible_cached(field, config, provider, &mut seen)
}
fn plugin_field_is_visible_cached(
field: &PluginFormFieldDefinition,
config: &ConnectionConfig,
provider: &PluginConnectionProviderContribution,
seen: &mut HashSet<String>,
) -> bool {
let Some(condition) = &field.visible_when else {
return true;
};
if !plugin_field_condition_matches(condition, config, provider) {
return false;
}
// The condition matched, so the sibling exists (its value was readable);
// guard cycles anyway and fall through to visible, mirroring the frontend.
let Some(sibling) = provider.fields.iter().find(|candidate| candidate.key == condition.field) else {
return true;
};
if seen.contains(&sibling.key) {
return true;
}
seen.insert(sibling.key.clone());
plugin_field_is_visible_cached(sibling, config, provider, seen)
}
/// Effective required = static `required` OR a matching `required_when`
/// (a missing `required_when` never implies required).
fn plugin_field_is_required(
field: &PluginFormFieldDefinition,
config: &ConnectionConfig,
provider: &PluginConnectionProviderContribution,
) -> bool {
if field.required {
return true;
}
match &field.required_when {
None => false,
Some(condition) => plugin_field_condition_matches(condition, config, provider),
}
}
fn validate_plugin_field_type(field: &PluginFormFieldDefinition, value: &serde_json::Value) -> Result<(), String> {
let valid = match field.field_type {
PluginFormFieldType::Text
@@ -624,6 +730,225 @@ mod tests {
.contains("port between 1 and 65535"));
}
#[test]
fn tolerates_stored_null_config_values_for_type_check() {
let config: ConnectionConfig = serde_json::from_value(serde_json::json!({
"id": "plugin-connection",
"name": "Plugin connection",
"db_type": "plugin",
"host": "localhost",
"port": 22,
"username": "root",
"password": "",
"database": null,
"plugin_id": "sample",
"plugin_connection_provider": "sample.connection",
"plugin_connection_type": "sample",
"external_config": { "authentication": "agent", "agent_socket": null }
}))
.unwrap();
let provider: PluginConnectionProviderContribution = serde_json::from_value(serde_json::json!({
"id": "sample.connection",
"label": "Sample",
"database_type": "sample",
"fields": [
{
"key": "authentication",
"label": "Authentication",
"type": "select",
"binding": "config",
"default": "password",
"options": [
{ "value": "password", "label": "Password" },
{ "value": "agent", "label": "Agent" }
]
},
{ "key": "agent_socket", "label": "Agent socket", "type": "text", "binding": "config" }
]
}))
.unwrap();
assert!(validate_plugin_connection_values(&config, &provider).is_ok());
}
/// Mirrors the files plugin manifest shape: `bucket` is required only when
/// `protocol` selects `s3` (`required_when` paired with `visible_when`).
#[test]
fn conditional_required_field_enforced_only_when_condition_matches() {
let provider: PluginConnectionProviderContribution = serde_json::from_value(serde_json::json!({
"id": "sample.connection",
"label": "Sample",
"database_type": "sample",
"fields": [
{
"key": "protocol",
"label": "Protocol",
"type": "select",
"binding": "config",
"required": true,
"default": "fs",
"options": [
{ "value": "fs", "label": "fs" },
{ "value": "s3", "label": "s3" }
]
},
{
"key": "bucket",
"label": "Bucket",
"type": "text",
"binding": "config",
"visible_when": { "field": "protocol", "one_of": ["s3"] },
"required_when": { "field": "protocol", "one_of": ["s3"] }
}
]
}))
.unwrap();
let config_for = |external: serde_json::Value| -> ConnectionConfig {
serde_json::from_value(serde_json::json!({
"id": "plugin-connection",
"name": "Plugin connection",
"db_type": "plugin",
"host": "localhost",
"port": 22,
"username": "",
"password": "",
"database": null,
"plugin_id": "sample",
"plugin_connection_provider": "sample.connection",
"plugin_connection_type": "sample",
"external_config": external
}))
.unwrap()
};
// fs: bucket not visible → not required → connection accepted.
assert!(
validate_plugin_connection_values(&config_for(serde_json::json!({ "protocol": "fs" })), &provider).is_ok()
);
// s3: bucket visible and conditionally required → empty bucket rejected.
assert!(validate_plugin_connection_values(&config_for(serde_json::json!({ "protocol": "s3" })), &provider)
.unwrap_err()
.contains("field 'Bucket' is required"));
// s3 with a bucket value passes.
assert!(validate_plugin_connection_values(
&config_for(serde_json::json!({ "protocol": "s3", "bucket": "demo" })),
&provider
)
.is_ok());
// Whitespace-only values count as empty.
assert!(validate_plugin_connection_values(
&config_for(serde_json::json!({ "protocol": "s3", "bucket": " " })),
&provider
)
.unwrap_err()
.contains("field 'Bucket' is required"));
}
#[test]
fn statically_required_field_only_enforced_while_visible() {
let provider: PluginConnectionProviderContribution = serde_json::from_value(serde_json::json!({
"id": "sample.connection",
"label": "Sample",
"database_type": "sample",
"fields": [
{
"key": "mode",
"label": "Mode",
"type": "select",
"binding": "config",
"default": "simple",
"options": [
{ "value": "simple", "label": "Simple" },
{ "value": "custom", "label": "Custom" }
]
},
{
"key": "secret_key",
"label": "Secret key",
"type": "password",
"binding": "secret",
"required": true,
"visible_when": { "field": "mode", "one_of": ["custom"] }
}
]
}))
.unwrap();
let config_for = |external: serde_json::Value, secrets: serde_json::Value| -> ConnectionConfig {
serde_json::from_value(serde_json::json!({
"id": "plugin-connection",
"name": "Plugin connection",
"db_type": "plugin",
"host": "localhost",
"port": 22,
"username": "",
"password": "",
"database": null,
"plugin_id": "sample",
"plugin_connection_provider": "sample.connection",
"plugin_connection_type": "sample",
"external_config": external,
"connection_secrets": secrets
}))
.unwrap()
};
// Hidden static-required field must not block the connection.
assert!(validate_plugin_connection_values(
&config_for(serde_json::json!({ "mode": "simple" }), serde_json::json!({})),
&provider
)
.is_ok());
// Visible and empty → rejected.
assert!(validate_plugin_connection_values(
&config_for(serde_json::json!({ "mode": "custom" }), serde_json::json!({})),
&provider
)
.unwrap_err()
.contains("field 'Secret key' is required"));
// Visible and filled → accepted.
assert!(validate_plugin_connection_values(
&config_for(serde_json::json!({ "mode": "custom" }), serde_json::json!({ "secret_key": "k" })),
&provider
)
.is_ok());
}
#[test]
fn required_when_referencing_missing_sibling_field_is_never_required() {
let provider: PluginConnectionProviderContribution = serde_json::from_value(serde_json::json!({
"id": "sample.connection",
"label": "Sample",
"database_type": "sample",
"fields": [
{ "key": "root", "label": "Root", "type": "text", "binding": "config" },
{
"key": "orphan",
"label": "Orphan",
"type": "text",
"binding": "config",
"required_when": { "field": "ghost", "one_of": ["x"] }
}
]
}))
.unwrap();
let config: ConnectionConfig = serde_json::from_value(serde_json::json!({
"id": "plugin-connection",
"name": "Plugin connection",
"db_type": "plugin",
"host": "localhost",
"port": 22,
"username": "",
"password": "",
"database": null,
"plugin_id": "sample",
"plugin_connection_provider": "sample.connection",
"plugin_connection_type": "sample"
}))
.unwrap();
assert!(validate_plugin_connection_values(&config, &provider).is_ok());
}
#[test]
fn allows_incomplete_fields_only_for_declared_permissive_actions() {
let config: ConnectionConfig = serde_json::from_value(serde_json::json!({
@@ -699,4 +1024,97 @@ mod tests {
.unwrap_err()
.contains("invalid value type"));
}
/// Mirrors the ldap plugin manifest shape: `krb_password` conditions on
/// `krb_credential_type`, which is itself hidden unless `auth_type` is
/// `kerberos`. A stored default on the hidden container field ("password")
/// must not make `krb_password` visible + required — the connection dialog
/// (cascading visibility) accepts such a save, so validation must too.
#[test]
fn cascaded_hidden_container_default_does_not_force_grandchild_required() {
let provider: PluginConnectionProviderContribution = serde_json::from_value(serde_json::json!({
"id": "ldap.connection",
"label": "LDAP",
"database_type": "ldap",
"fields": [
{
"key": "auth_type",
"label": "Authentication",
"type": "select",
"binding": "config",
"default": "simple",
"options": [
{ "value": "simple", "label": "Simple" },
{ "value": "kerberos", "label": "Kerberos (GSSAPI)" }
]
},
{
"key": "krb_credential_type",
"label": "Kerberos credential type",
"type": "select",
"binding": "config",
"default": "password",
"visible_when": { "field": "auth_type", "one_of": ["kerberos"] },
"options": [
{ "value": "password", "label": "Password" },
{ "value": "keytab", "label": "Keytab" }
]
},
{
"key": "krb_password",
"label": "Kerberos password",
"type": "password",
"binding": "secret",
"visible_when": { "field": "krb_credential_type", "one_of": ["password"] },
"required_when": { "field": "krb_credential_type", "one_of": ["password"] }
}
]
}))
.unwrap();
let config_for = |external: serde_json::Value| -> ConnectionConfig {
serde_json::from_value(serde_json::json!({
"id": "plugin-connection",
"name": "KN-LDAP",
"db_type": "plugin",
"host": "ldap.example.com",
"port": 389,
"username": "",
"password": "",
"database": null,
"plugin_id": "io.dbx.ldap",
"plugin_connection_provider": "ldap.connection",
"plugin_connection_type": "ldap",
"external_config": external
}))
.unwrap()
};
// Simple auth with the stale stored default on the hidden container
// field: dialog saves it, host must not demand the hidden secret.
let simple = config_for(serde_json::json!({ "auth_type": "simple", "krb_credential_type": "password" }));
assert!(validate_plugin_connection_values(&simple, &provider).is_ok());
// Kerberos + password credential type: the grandchild is genuinely
// visible and required, and the empty secret must still be rejected.
let kerberos = config_for(serde_json::json!({ "auth_type": "kerberos", "krb_credential_type": "password" }));
assert!(validate_plugin_connection_values(&kerberos, &provider)
.unwrap_err()
.contains("Kerberos password' is required"));
// Cycle guard: mutual references must not hang or misclassify.
let cyclic: PluginConnectionProviderContribution = serde_json::from_value(serde_json::json!({
"id": "cyclic.connection",
"label": "Cyclic",
"database_type": "cyclic",
"fields": [
{ "key": "a", "label": "A", "type": "text", "binding": "config",
"visible_when": { "field": "b", "one_of": ["x"] } },
{ "key": "b", "label": "B", "type": "text", "binding": "config",
"visible_when": { "field": "a", "one_of": ["x"] } }
]
}))
.unwrap();
let cyclic_config = config_for(serde_json::json!({ "a": "x", "b": "x" }));
assert!(validate_plugin_connection_values(&cyclic_config, &cyclic).is_ok());
}
}
+146
View File
@@ -248,6 +248,23 @@ pub struct PluginFormFieldDefinition {
pub options: Vec<PluginFormFieldOption>,
#[serde(default)]
pub binding: Option<PluginFormFieldBinding>,
/// Plugin method returning `{ options: [{ value, label }] }`; the host
/// connection form fetches it and renders the field as a dynamic select.
/// Optional and forward/backward compatible: older hosts reject the
/// manifest, hosts without UI support keep the declared type.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub options_action: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub visible_when: Option<PluginFieldCondition>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub required_when: Option<PluginFieldCondition>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct PluginFieldCondition {
pub field: String,
pub one_of: Vec<String>,
}
impl PluginFormFieldDefinition {
@@ -964,6 +981,7 @@ fn validate_connection_actions(
fn validate_form_fields(fields: &[PluginFormFieldDefinition], contribution_index: usize, errors: &mut Vec<String>) {
let mut seen_keys = HashSet::new();
let field_keys = fields.iter().map(|field| field.key.as_str()).collect::<HashSet<_>>();
for (field_index, field) in fields.iter().enumerate() {
if !valid_identifier(&field.key) || !seen_keys.insert(&field.key) {
errors.push(format!(
@@ -976,6 +994,25 @@ fn validate_form_fields(fields: &[PluginFormFieldDefinition], contribution_index
errors,
);
for (name, condition) in
[("visible_when", field.visible_when.as_ref()), ("required_when", field.required_when.as_ref())]
{
let Some(condition) = condition else {
continue;
};
if condition.one_of.is_empty() {
errors.push(format!(
"Contribution at index {contribution_index} field {field_index} {name} one_of cannot be empty"
));
}
if !field_keys.contains(condition.field.as_str()) {
errors.push(format!(
"Contribution at index {contribution_index} field {field_index} {name} references unknown field '{}'",
condition.field
));
}
}
if matches!(field.field_type, PluginFormFieldType::Select | PluginFormFieldType::Radio) {
if field.options.is_empty() {
errors.push(format!(
@@ -1675,4 +1712,113 @@ mod tests {
assert!(errors.iter().any(|error| error.contains("invalid or duplicate id")));
assert!(errors.iter().any(|error| error.contains("timeout_ms must be between")));
}
#[test]
fn parses_form_field_conditions() {
let dir = tempfile::tempdir().unwrap();
let manifest: PluginManifest = serde_json::from_value(serde_json::json!({
"manifest_version": 1,
"id": "io.dbx.conditions",
"name": "Conditions",
"version": "1.0.0",
"publisher": "example",
"engines": { "dbx": ">=0.1.0", "host_api": "^1.0" },
"contributions": [{
"type": "connection-provider",
"id": "conditions.connection",
"database_type": "conditions",
"fields": [
{
"key": "mode",
"label": "Mode",
"type": "select",
"options": [
{ "label": "Local", "value": "local" },
{ "label": "Custom", "value": "custom" }
],
"default": "local"
},
{
"key": "private_key_path",
"label": "Private key",
"type": "text",
"visible_when": { "field": "mode", "one_of": ["custom"] },
"required_when": { "field": "mode", "one_of": ["custom"] }
}
]
}]
}))
.unwrap();
let compatibility = manifest.compatibility(dir.path(), "0.5.68");
let provider = manifest.connection_provider("conditions.connection").unwrap().unwrap();
assert!(compatibility.compatible, "{:?}", compatibility.errors);
let visible_when = provider.fields[1].visible_when.as_ref().unwrap();
assert_eq!(visible_when.field, "mode");
assert_eq!(visible_when.one_of, vec!["custom".to_string()]);
assert_eq!(provider.fields[1].required_when.as_ref().unwrap().one_of, vec!["custom".to_string()]);
}
#[test]
fn rejects_invalid_form_field_conditions() {
let dir = tempfile::tempdir().unwrap();
let malformed = serde_json::from_value::<PluginManifest>(serde_json::json!({
"manifest_version": 1,
"id": "io.dbx.conditions",
"name": "Conditions",
"version": "1.0.0",
"publisher": "example",
"engines": { "dbx": ">=0.1.0", "host_api": "^1.0" },
"contributions": [{
"type": "connection-provider",
"id": "conditions.connection",
"database_type": "conditions",
"fields": [
{ "key": "mode", "label": "Mode", "type": "text" },
{
"key": "private_key_path",
"label": "Private key",
"type": "text",
"visible_when": { "field": "missing", "one_of": [] },
"required_when": { "field": "mode", "one_of": ["custom"], "extra": true }
}
]
}]
}));
assert!(malformed.is_err());
let manifest: PluginManifest = serde_json::from_value(serde_json::json!({
"manifest_version": 1,
"id": "io.dbx.conditions",
"name": "Conditions",
"version": "1.0.0",
"publisher": "example",
"engines": { "dbx": ">=0.1.0", "host_api": "^1.0" },
"contributions": [{
"type": "connection-provider",
"id": "conditions.connection",
"database_type": "conditions",
"fields": [
{ "key": "mode", "label": "Mode", "type": "text" },
{
"key": "private_key_path",
"label": "Private key",
"type": "text",
"visible_when": { "field": "missing", "one_of": [] },
"required_when": { "field": "mode", "one_of": ["custom"] }
}
]
}]
}))
.unwrap();
let compatibility = manifest.compatibility(dir.path(), "0.5.68");
assert!(!compatibility.compatible);
assert!(compatibility.errors.iter().any(|error| error.contains("visible_when one_of cannot be empty")));
assert!(compatibility
.errors
.iter()
.any(|error| error.contains("visible_when references unknown field 'missing'")));
}
}
+9 -8
View File
@@ -4181,6 +4181,7 @@ fn pool_kind_has_transactional_path(pool: &PoolKind) -> bool {
PoolKind::MessageQueue
| PoolKind::Nacos
| PoolKind::Consul(_)
| PoolKind::PluginConnection(_)
| PoolKind::HBase(_)
| PoolKind::DuckDbWorker(_)
| PoolKind::Redis(_)
@@ -4193,8 +4194,7 @@ fn pool_kind_has_transactional_path(pool: &PoolKind) -> bool {
| PoolKind::InfluxDb(_)
| PoolKind::InfluxDb3(_)
| PoolKind::VictoriaMetrics(_)
| PoolKind::ExternalDriver { .. }
| PoolKind::PluginConnection(_) => false,
| PoolKind::ExternalDriver { .. } => false,
#[cfg(feature = "mq-admin")]
PoolKind::Mqtt(_) => false,
}
@@ -4575,9 +4575,11 @@ fn batch_transaction_path(pool: &PoolKind) -> BatchTransactionPath {
PoolKind::Sqlite(pool) => BatchTransactionPath::Sqlite(pool.clone()),
PoolKind::SqlServer(_) => BatchTransactionPath::Explicit,
PoolKind::Agent(client) => BatchTransactionPath::Agent(client.clone()),
PoolKind::MessageQueue | PoolKind::Nacos | PoolKind::Consul(_) | PoolKind::HBase(_) => {
BatchTransactionPath::Unsupported
}
PoolKind::MessageQueue
| PoolKind::Nacos
| PoolKind::PluginConnection(_)
| PoolKind::Consul(_)
| PoolKind::HBase(_) => BatchTransactionPath::Unsupported,
#[cfg(feature = "mq-admin")]
PoolKind::Mqtt(_) => BatchTransactionPath::Unsupported,
PoolKind::DuckDbWorker(_)
@@ -4595,8 +4597,7 @@ fn batch_transaction_path(pool: &PoolKind) -> BatchTransactionPath {
| PoolKind::InfluxDb(_)
| PoolKind::InfluxDb3(_)
| PoolKind::VictoriaMetrics(_)
| PoolKind::ExternalDriver { .. }
| PoolKind::PluginConnection(_) => BatchTransactionPath::Unsupported,
| PoolKind::ExternalDriver { .. } => BatchTransactionPath::Unsupported,
}
}
@@ -8809,7 +8810,7 @@ for line in sys.stdin:
std::fs::write(
&executable,
format!(
"#!/bin/sh\nwhile IFS= read -r line; do\n id=$(printf '%s' \"$line\" | sed -E 's/.*\"id\":([0-9]+).*/\\1/')\n case \"$line\" in\n *'\"method\":\"executeQueryPage\"'*)\n echo executeQueryPage >> '{}'\n case \"$line\" in\n *'left('*) printf '{{\"id\":%s,\"error\":{{\"message\":\"ERROR: invalid byte sequence for encoding \\\"UTF8\\\": 0xe2\"}}}}\\n' \"$id\" ;;\n *) printf '{{\"id\":%s,\"result\":{{\"columns\":[\"id\",\"description\",\"__DBX_LARGE_VALUE_BYTES_T_1\"],\"column_types\":[\"integer\",\"text\",\"text\"],\"rows\":[[1,\"abcdef\",\"T:3\"]],\"affected_rows\":0,\"execution_time_ms\":1,\"truncated\":false}}}}\\n' \"$id\" ;;\n esac\n ;;\n *'\"method\":\"executeQuery\"'*)\n echo executeQuery >> '{}'\n case \"$line\" in\n *'left('*) printf '{{\"id\":%s,\"error\":{{\"message\":\"ERROR: invalid byte sequence for encoding \\\"UTF8\\\": 0xe2\"}}}}\\n' \"$id\" ;;\n *) printf '{{\"id\":%s,\"result\":{{\"columns\":[\"id\",\"description\",\"__DBX_LARGE_VALUE_BYTES_T_1\"],\"column_types\":[\"integer\",\"text\",\"text\"],\"rows\":[[1,\"abcdef\",\"T:3\"]],\"affected_rows\":0,\"execution_time_ms\":1,\"truncated\":false}}}}\\n' \"$id\" ;;\n esac\n ;;\n esac\ndone\n",
"#!/bin/sh\nwhile IFS= read -r line; do\n id=$(printf '%s' \"$line\" | sed -E 's/.*\"id\":([0-9]+).*/\\1/')\n case \"$line\" in\n *'\"method\":\"executeQueryPage\"'*)\n echo executeQueryPage >> '{}'\n case \"$line\" in\n *'left('*) printf '{{\"id\":%s,\"error\":{{\"message\":\"ERROR: invalid byte sequence for encoding UTF8: 0xe2\"}}}}\\n' \"$id\" ;;\n *) printf '{{\"id\":%s,\"result\":{{\"columns\":[\"id\",\"description\",\"__DBX_LARGE_VALUE_BYTES_T_1\"],\"column_types\":[\"integer\",\"text\",\"text\"],\"rows\":[[1,\"abcdef\",\"T:3\"]],\"affected_rows\":0,\"execution_time_ms\":1,\"truncated\":false}}}}\\n' \"$id\" ;;\n esac\n ;;\n *'\"method\":\"executeQuery\"'*)\n echo executeQuery >> '{}'\n case \"$line\" in\n *'left('*) printf '{{\"id\":%s,\"error\":{{\"message\":\"ERROR: invalid byte sequence for encoding UTF8: 0xe2\"}}}}\\n' \"$id\" ;;\n *) printf '{{\"id\":%s,\"result\":{{\"columns\":[\"id\",\"description\",\"__DBX_LARGE_VALUE_BYTES_T_1\"],\"column_types\":[\"integer\",\"text\",\"text\"],\"rows\":[[1,\"abcdef\",\"T:3\"]],\"affected_rows\":0,\"execution_time_ms\":1,\"truncated\":false}}}}\\n' \"$id\" ;;\n esac\n ;;\n esac\ndone\n",
calls.display(),
calls.display()
),
+65 -8
View File
@@ -15,10 +15,11 @@ use crate::ai::{
AiRunStatus,
};
use crate::connection_secrets::{
CASSANDRA_KEYSTORE_PASSWORD_KEY, CASSANDRA_TLS_SECRET_PREFIX, CASSANDRA_TRUSTSTORE_PASSWORD_KEY,
MQ_AUTH_API_KEY_VALUE_KEY, MQ_AUTH_CLIENT_SECRET_KEY, MQ_AUTH_PASSWORD_KEY, MQ_AUTH_SECRET_PREFIX,
MQ_AUTH_TOKEN_KEY, MQ_TOKEN_SIGNING_KEY, MQ_TOKEN_SIGNING_SECRET_PREFIX, NACOS_AUTH_PASSWORD_KEY,
NACOS_AUTH_SECRET_PREFIX, NACOS_RNACOS_CONSOLE_PASSWORD_KEY, PLUGIN_CONNECTION_SECRET_PREFIX,
plugin_connection_secret_key, CASSANDRA_KEYSTORE_PASSWORD_KEY, CASSANDRA_TLS_SECRET_PREFIX,
CASSANDRA_TRUSTSTORE_PASSWORD_KEY, MQ_AUTH_API_KEY_VALUE_KEY, MQ_AUTH_CLIENT_SECRET_KEY, MQ_AUTH_PASSWORD_KEY,
MQ_AUTH_SECRET_PREFIX, MQ_AUTH_TOKEN_KEY, MQ_TOKEN_SIGNING_KEY, MQ_TOKEN_SIGNING_SECRET_PREFIX,
NACOS_AUTH_PASSWORD_KEY, NACOS_AUTH_SECRET_PREFIX, NACOS_RNACOS_CONSOLE_PASSWORD_KEY,
PLUGIN_CONNECTION_SECRET_PREFIX,
};
use crate::db::sqlite::{connect_path_create_if_missing, SqliteHandle};
use crate::history::{
@@ -1363,6 +1364,12 @@ fn scrub_nacos_auth_secrets(config: &mut ConnectionConfig) {
}
}
fn scrub_plugin_connection_secrets(config: &mut ConnectionConfig) {
for secret in config.connection_secrets.values_mut() {
secret.clear();
}
}
fn scrub_cassandra_tls_secrets(config: &mut ConnectionConfig) {
if config.db_type != DatabaseType::Cassandra {
return;
@@ -3776,16 +3783,32 @@ impl Storage {
let needs_mq_token_signing_rewrite = self.hydrate_mq_token_signing_secret(&id, &mut config).await?;
let needs_nacos_auth_rewrite = self.hydrate_nacos_auth_secret(&id, &mut config).await?;
let needs_cassandra_tls_rewrite = self.hydrate_cassandra_tls_secrets(&id, &mut config).await?;
let mut needs_plugin_secret_rewrite = false;
let plugin_secret_keys = config.connection_secrets.keys().cloned().collect::<Vec<_>>();
for key in plugin_secret_keys {
let storage_key = plugin_connection_secret_key(&key)?;
let current = config.connection_secrets.get(&key).cloned().unwrap_or_default();
if current.is_empty() {
if let Some(secret) = self.get_secret(&id, &storage_key).await? {
config.connection_secrets.insert(key, secret);
}
} else {
self.set_secret(&id, &storage_key, &current).await?;
needs_plugin_secret_rewrite = true;
}
}
let needs_external_secret_rewrite = needs_mq_auth_rewrite
|| needs_mq_token_signing_rewrite
|| needs_nacos_auth_rewrite
|| needs_cassandra_tls_rewrite;
|| needs_cassandra_tls_rewrite
|| needs_plugin_secret_rewrite;
if needs_external_secret_rewrite {
let mut sanitized = config.clone().canonicalized();
scrub_mq_auth_secrets(&mut sanitized);
scrub_mq_token_signing_secret(&mut sanitized);
scrub_nacos_auth_secrets(&mut sanitized);
scrub_cassandra_tls_secrets(&mut sanitized);
scrub_plugin_connection_secrets(&mut sanitized);
let sanitized_json = serde_json::to_string(&sanitized).map_err(|e| e.to_string())?;
let update_id = id.clone();
self.with_conn(move |conn| {
@@ -5353,8 +5376,9 @@ mod tests {
};
use crate::connection_secrets::NACOS_RNACOS_CONSOLE_PASSWORD_KEY;
use crate::connection_secrets::{
CASSANDRA_KEYSTORE_PASSWORD_KEY, CASSANDRA_TRUSTSTORE_PASSWORD_KEY, MQ_AUTH_PASSWORD_KEY, MQ_AUTH_TOKEN_KEY,
MQ_TOKEN_SIGNING_KEY, NACOS_AUTH_PASSWORD_KEY, PLUGIN_CONNECTION_SECRET_PREFIX,
plugin_connection_secret_key, CASSANDRA_KEYSTORE_PASSWORD_KEY, CASSANDRA_TRUSTSTORE_PASSWORD_KEY,
MQ_AUTH_PASSWORD_KEY, MQ_AUTH_TOKEN_KEY, MQ_TOKEN_SIGNING_KEY, NACOS_AUTH_PASSWORD_KEY,
PLUGIN_CONNECTION_SECRET_PREFIX,
};
use crate::history::{HistoryConnectionFilter, HistoryDatabaseFilter, HistoryEntry, HistorySearchRequest};
use crate::models::connection::{
@@ -6084,7 +6108,7 @@ mod tests {
}
#[tokio::test]
async fn save_connections_moves_plugin_secrets_to_secret_table_and_restores_them() {
async fn save_connections_moves_plugin_secrets_to_secret_table_and_clears_removed_values() {
let path = temp_db_path("plugin-secrets");
let storage = Storage::open(&path).await.unwrap();
let mut config = plain_connection("plugin", "");
@@ -6673,6 +6697,39 @@ mod tests {
let _ = std::fs::remove_file(path);
}
#[tokio::test]
async fn save_connections_moves_plugin_secrets_to_secret_table_and_restores_them() {
let path = temp_db_path("plugin-connection-secret");
let storage = Storage::open(&path).await.unwrap();
let mut config = mq_connection("plugin-connection", "");
config.name = "Hello plugin".to_string();
config.db_type = DatabaseType::Plugin;
config.driver_profile = Some("plugin".to_string());
config.external_config = Some(serde_json::json!({ "greeting": "Hello" }));
config.plugin_id = Some("dbx.example.hello".to_string());
config.plugin_connection_provider = Some("hello.connection".to_string());
config.plugin_connection_type = Some("hello".to_string());
config.connection_secrets.insert("access_token".to_string(), "plugin-secret".to_string());
storage.save_connections(&[config]).await.unwrap();
let raw_json = raw_connection_json(&storage, "plugin-connection").await;
assert!(!raw_json.contains("plugin-secret"));
let persisted: ConnectionConfig = serde_json::from_str(&raw_json).unwrap();
assert_eq!(persisted.connection_secrets.get("access_token"), None);
assert_eq!(
storage
.get_secret("plugin-connection", &plugin_connection_secret_key("access_token").unwrap())
.await
.unwrap()
.as_deref(),
Some("plugin-secret")
);
let loaded = storage.load_connections().await.unwrap();
assert_eq!(loaded[0].connection_secrets.get("access_token").map(String::as_str), Some("plugin-secret"));
}
#[tokio::test]
async fn metadata_save_scrubs_mq_auth_token_and_preserves_existing_secret() {
let path = temp_db_path("mq-token-metadata");
@@ -0,0 +1,38 @@
//! Verifies a real DBX app store: plugin discovery + MCP tool bridge.
//! cargo run -p dbx-mcp --example verify_plugin_store -- <data-dir> [app-version]
use dbx_mcp::LocalBackend;
#[tokio::main]
async fn main() {
let args: Vec<String> = std::env::args().skip(1).collect();
let data_dir = std::path::PathBuf::from(&args[0]);
let app_version = args.get(1).map(String::as_str).unwrap_or("0.5.96");
let backend =
LocalBackend::open_with_app_version(&data_dir.join("dbx.db"), app_version).await.expect("open local backend");
let installed = backend.state().plugins.list_installed().unwrap();
for plugin in &installed {
println!(
"plugin: {} v{} compatible={} errors={:?}",
plugin.manifest.id, plugin.manifest.version, plugin.compatibility.compatible, plugin.compatibility.errors
);
}
let providers = backend.list_plugin_tools().await.expect("list plugin tools");
for provider in &providers {
let tools = provider["tools"].as_array().map(|tools| tools.len()).unwrap_or(0);
println!("mcp tools: {} -> {} tools", provider["pluginId"], tools);
let call = backend
.call_plugin_tool(
provider["pluginId"].as_str().unwrap(),
"ssh_list_known_hosts",
None,
&serde_json::json!({}),
)
.await;
match call {
Ok(result) => println!("tool call ok: isError={}", result["isError"].as_bool().unwrap_or(true)),
Err(error) => println!("tool call failed: {error}"),
}
}
}
+66 -1
View File
@@ -582,6 +582,12 @@ impl LocalBackend {
}
pub async fn open(path: &Path) -> Result<Self, String> {
Self::open_with_app_version(path, env!("CARGO_PKG_VERSION")).await
}
/// Same as [`open`], but lets tests and embedded callers pin the app version
/// used for plugin compatibility checks instead of the compile-time version.
pub async fn open_with_app_version(path: &Path, app_version: &str) -> Result<Self, String> {
let storage = Storage::open(path).await?;
let configs = storage.load_connections().await?;
let desktop_settings = storage.load_desktop_settings().await.unwrap_or_default();
@@ -592,7 +598,7 @@ impl LocalBackend {
storage,
plugin_dir,
agent_dir,
env!("CARGO_PKG_VERSION"),
app_version,
));
let config_map: HashMap<String, ConnectionConfig> =
configs.into_iter().map(|config| (config.id.clone(), config)).collect();
@@ -604,6 +610,65 @@ impl LocalBackend {
&self.state
}
/// List the MCP tools exposed by every installed plugin.
///
/// Plugin MCP surfaces are discovered through the same `mcp/tools`
/// protocol used by the desktop bridge. Keep the response grouped by
/// plugin so callers can select the correct sidecar for a tool call.
pub async fn list_plugin_tools(&self) -> Result<Vec<Value>, String> {
let plugins = self.state.plugins.list_installed()?;
let mut providers = Vec::new();
for plugin in plugins {
if !plugin.compatibility.compatible || plugin.manifest.backend_entrypoint().is_none() {
continue;
}
let tools: Value = self
.state
.plugin_host
.invoke(&plugin.manifest.id, "mcp/tools", json!({}), None, Some(std::time::Duration::from_secs(30)))
.await?;
let tool_list = tools
.get("tools")
.cloned()
.or_else(|| tools.is_array().then_some(tools.clone()))
.unwrap_or_else(|| json!([]));
providers.push(json!({
"pluginId": plugin.manifest.id,
"tools": tool_list,
}));
}
Ok(providers)
}
/// Call one plugin MCP tool through the host-managed sidecar session.
/// When a saved connection is supplied, only its host-generated lifecycle
/// payload is sent to the plugin; credentials remain host-managed.
pub async fn call_plugin_tool(
&self,
plugin_id: &str,
tool: &str,
connection_id: Option<&str>,
arguments: &Value,
) -> Result<Value, String> {
let mut params = json!({ "tool": tool, "arguments": arguments });
if let Some(connection_id) = connection_id {
let config = self
.state
.configs
.read()
.await
.get(connection_id)
.cloned()
.ok_or_else(|| format!("Connection not found: {connection_id}"))?;
let lifecycle = self.state.plugin_host.connection_params_standalone(&config)?;
params["lifecycle"] = lifecycle;
}
self.state
.plugin_host
.invoke(plugin_id, "mcp/call", params, None, Some(std::time::Duration::from_secs(300)))
.await
}
/// Sync the latest connection list from storage into the `AppState.configs` in-memory cache:
/// upsert new/changed entries and remove connections deleted from storage. Only LocalBackend
/// needs this — WebBackend talks HTTP and holds no local AppState, and the desktop mcp_bridge
@@ -0,0 +1,66 @@
//! Full installation-pipeline test: installs the real ssh-sftp `.dbxp`
//! package through PluginPackageInstaller, then verifies the MCP bridge can
//! discover the plugin, start its sidecar, and call a tool.
//!
//! Set DBX_SSH_SFTP_PACKAGE to the packaged artifact; the test skips itself
//! otherwise.
use dbx_core::plugins::{PluginInstallPolicy, PluginPackageInstaller, PluginSignatureStatus};
use dbx_mcp::LocalBackend;
#[tokio::test]
async fn ssh_sftp_package_installs_and_serves_mcp_tools() {
let Ok(package) = std::env::var("DBX_SSH_SFTP_PACKAGE") else {
eprintln!("skipping: DBX_SSH_SFTP_PACKAGE is not set");
return;
};
let package = std::path::PathBuf::from(package);
assert!(package.is_file(), "package missing: {}", package.display());
let data_dir = tempfile::tempdir().unwrap();
let plugins_root = data_dir.path().join("plugins");
// 1. Install the .dbxp exactly like the DBX desktop app would.
let installer = PluginPackageInstaller::new(plugins_root.clone(), "0.6.0").expect("installer");
let result = installer.install_file(&package, PluginInstallPolicy::LocalDevelopment).expect("install .dbxp");
let response = result.response();
assert_eq!(response.plugin.manifest.id, "io.dbx.ssh-sftp");
assert!(!response.plugin.manifest.version.is_empty());
assert_eq!(response.signature, PluginSignatureStatus::Unsigned);
// 2. The installed layout is discoverable as a connection provider.
let database_path = data_dir.path().join("dbx.db");
let storage = dbx_core::storage::Storage::open(&database_path).await.unwrap();
drop(storage);
let backend = LocalBackend::open_with_app_version(&database_path, "0.6.0").await.unwrap();
let installed = backend.state().plugins.list_installed().unwrap();
let plugin =
installed.iter().find(|plugin| plugin.manifest.id == "io.dbx.ssh-sftp").expect("installed plugin listed");
assert!(plugin.compatibility.compatible, "plugin incompatible: {:?}", plugin.compatibility.errors);
// 3. MCP bridge: tool discovery over the real sidecar process.
let providers = backend.list_plugin_tools().await.expect("list plugin tools");
let ssh = providers
.iter()
.find(|provider| provider["pluginId"] == "io.dbx.ssh-sftp")
.expect("plugin contributes MCP tools");
let names = ssh["tools"]
.as_array()
.expect("tools array")
.iter()
.map(|tool| tool["name"].as_str().unwrap_or_default())
.collect::<Vec<_>>();
for expected in ["ssh_exec", "ssh_exec_sudo", "ssh_metrics", "sftp_list_dir", "sftp_chmod"] {
assert!(names.contains(&expected), "missing {expected} in {names:?}");
}
// 4. MCP bridge: a real tool call through the installed sidecar.
let listed = backend
.call_plugin_tool("io.dbx.ssh-sftp", "ssh_list_known_hosts", None, &serde_json::json!({}))
.await
.expect("call plugin tool");
assert!(!listed["isError"].as_bool().unwrap_or(true), "tool failed: {listed}");
let text = listed["content"][0]["text"].as_str().expect("result text");
let parsed: serde_json::Value = serde_json::from_str(text).expect("result JSON");
assert!(parsed["knownHosts"].is_array());
}