diff --git a/Cargo.lock b/Cargo.lock index c66de1be7..2b0b894cf 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2561,6 +2561,7 @@ name = "fabro-db" version = "0.302.0-nightly.1" dependencies = [ "anyhow", + "chrono", "sqlx", "tempfile", "tokio", @@ -2846,12 +2847,18 @@ dependencies = [ name = "fabro-mcp-store" version = "0.302.0-nightly.1" dependencies = [ + "chrono", + "fabro-db", "fabro-types", "serde", + "serde_json", + "sqlx", + "strum 0.28.0", "tempfile", "thiserror 2.0.18", "tokio", "toml 0.8.23", + "tracing", ] [[package]] diff --git a/docs/plans/2026-07-11-mcp-servers-sqlite-plan.md b/docs/plans/2026-07-11-mcp-servers-sqlite-plan.md new file mode 100644 index 000000000..66179f999 --- /dev/null +++ b/docs/plans/2026-07-11-mcp-servers-sqlite-plan.md @@ -0,0 +1,70 @@ +# MCP Servers to SQLite + +## Goal + +Move server-managed MCP definitions from sibling `mcps/*.toml` files into the +shared SQLite database. Preserve the REST contract, value-omitting read views, +content-hash ETags, synchronous manifest catalog reads, and all three transport +types. + +## Schema + +Use one `mcp_servers` row per definition: + +- Scalar columns: id, revision, display name, description, transport type, + protocol, URL, port, and timeouts. +- Typed JSON columns: command array, env map, and header map. +- A transport-shape constraint requires exactly the columns belonging to + `stdio`, `http`, or `sandbox`. +- ID, revision, protocol, port, timeout, and top-level JSON shape constraints + provide database-level defense in depth. +- No secondary indexes: supported queries are primary-key lookup and full + sorted listing. + +Keep `McpServerRevision` as lowercase SHA-256 hex. Create and replace derive it +from the existing canonical representation. Legacy import preserves the hash +of the original TOML bytes so an ETag remains valid across upgrade. + +## Store + +- `fabro-mcp-store` owns SQL, row mapping, typed JSON encoding, validation, + revisions, caching, and legacy import. +- Retain the synchronous in-memory catalog required by manifest resolution. +- Serialize in-process mutations, then enforce replace/delete revisions in SQL + with `WHERE id = ? AND revision = ?`. +- Update the cache only after a successful transaction. +- Revalidate every decoded row through the existing domain model. +- Encode env/header maps through sorted maps for deterministic JSON. +- Never log or debug-print transport env/header values. + +## Legacy import + +At startup, inspect `mcps/` next to the active `settings.toml`: + +1. Missing directory: no-op. +2. Parse and validate every TOML definition before mutating SQLite. +3. Insert all definitions in one transaction with + `ON CONFLICT(id) DO NOTHING`; SQLite wins. +4. Commit, then rename the directory to + `mcps.imported-.bak`. +5. If backup rename fails, leave the source directory for a retry; the next + import skips existing SQLite rows and retries the rename. + +Logs contain only paths, counts, and MCP ids. They never contain commands, +URLs, env values, or headers. + +## Tests + +- Schema accepts every transport and rejects invalid variant shapes. +- CRUD, sorted listing, reload persistence, and all transport round trips. +- Independent store instances enforce SQL revision conflicts. +- Deterministic map JSON and typed corrupted-row errors. +- Import success, SQLite precedence, stable imported revision, retry no-op, + malformed input unchanged, and directory backup. +- API persistence/restart behavior, value-omitting reads, legacy startup import, + malformed legacy startup failure, and existing auth requirements. +- Workspace build, formatter, Clippy, and relevant Nextest suites. + +## Unresolved questions + +None. diff --git a/docs/public/agents/mcp.mdx b/docs/public/agents/mcp.mdx index b3fda877c..76fb68054 100644 --- a/docs/public/agents/mcp.mdx +++ b/docs/public/agents/mcp.mdx @@ -119,6 +119,21 @@ MCP servers available to Fabro agents can be configured in two places: Each server entry specifies a transport type and optional timeouts. The server name is the TOML table key and is used in qualified tool names. +### Server-managed catalog + +Fabro servers can also manage a shared MCP catalog through the MCP servers REST API. Workflows reference a catalog definition by id instead of repeating its transport configuration: + +```toml +[run.agent.mcps.sentry] +id = "sentry" +``` + +Server-managed definitions are stored in the server's shared SQLite database. Read APIs return configured env/header names but never their values. + +When upgrading an installation that stored definitions as `mcps/*.toml` next to the active `settings.toml`, Fabro validates and imports the directory during startup. Existing SQLite rows win on id conflicts. After a successful transaction, Fabro renames the source directory to a timestamped backup such as `mcps.imported-20260711T120000000000Z.bak`. + +Transport env/header values preserve their existing plaintext-at-rest behavior in SQLite. Prefer `{{ secrets.NAME }}` interpolation over literal credentials where possible. + ## Transports ### Stdio diff --git a/lib/crates/fabro-db/Cargo.toml b/lib/crates/fabro-db/Cargo.toml index 34963bb7a..38cfed522 100644 --- a/lib/crates/fabro-db/Cargo.toml +++ b/lib/crates/fabro-db/Cargo.toml @@ -14,6 +14,7 @@ workspace = true [dependencies] anyhow.workspace = true +chrono.workspace = true sqlx.workspace = true tokio.workspace = true tracing.workspace = true diff --git a/lib/crates/fabro-db/migrations/2026071102_mcp_servers.sql b/lib/crates/fabro-db/migrations/2026071102_mcp_servers.sql new file mode 100644 index 000000000..e5e2d009e --- /dev/null +++ b/lib/crates/fabro-db/migrations/2026071102_mcp_servers.sql @@ -0,0 +1,73 @@ +CREATE TABLE mcp_servers ( + id TEXT PRIMARY KEY NOT NULL, + revision TEXT NOT NULL, + display_name TEXT NOT NULL, + description TEXT, + transport_type TEXT NOT NULL, + protocol TEXT, + command_json TEXT, + url TEXT, + port INTEGER, + env_json TEXT, + headers_json TEXT, + startup_timeout_secs INTEGER NOT NULL, + tool_timeout_secs INTEGER NOT NULL, + CHECK (length(id) BETWEEN 1 AND 63), + CHECK (substr(id, 1, 1) GLOB '[a-z0-9]'), + CHECK (id NOT GLOB '*[^a-z0-9-]*'), + CHECK (length(revision) = 64), + CHECK (revision NOT GLOB '*[^0-9a-f]*'), + CHECK (length(trim(display_name)) > 0), + CHECK (transport_type IN ('stdio', 'http', 'sandbox')), + CHECK (protocol IS NULL OR protocol IN ('streamable_http', 'sse')), + CHECK (startup_timeout_secs >= 0), + CHECK (tool_timeout_secs >= 0), + CHECK ( + ( + transport_type = 'stdio' + AND protocol IS NULL + AND command_json IS NOT NULL + AND json_valid(command_json) + AND json_type(command_json) = 'array' + AND json_array_length(command_json) > 0 + AND json_type(command_json, '$[0]') = 'text' + AND length(trim(json_extract(command_json, '$[0]'))) > 0 + AND url IS NULL + AND port IS NULL + AND env_json IS NOT NULL + AND json_valid(env_json) + AND json_type(env_json) = 'object' + AND headers_json IS NULL + ) + OR + ( + transport_type = 'http' + AND protocol IS NOT NULL + AND command_json IS NULL + AND url IS NOT NULL + AND length(trim(url)) > 0 + AND port IS NULL + AND env_json IS NULL + AND headers_json IS NOT NULL + AND json_valid(headers_json) + AND json_type(headers_json) = 'object' + ) + OR + ( + transport_type = 'sandbox' + AND protocol IS NOT NULL + AND command_json IS NOT NULL + AND json_valid(command_json) + AND json_type(command_json) = 'array' + AND json_array_length(command_json) > 0 + AND json_type(command_json, '$[0]') = 'text' + AND length(trim(json_extract(command_json, '$[0]'))) > 0 + AND url IS NULL + AND port BETWEEN 1 AND 65535 + AND env_json IS NOT NULL + AND json_valid(env_json) + AND json_type(env_json) = 'object' + AND headers_json IS NULL + ) + ) +); diff --git a/lib/crates/fabro-db/src/legacy.rs b/lib/crates/fabro-db/src/legacy.rs new file mode 100644 index 000000000..74065be40 --- /dev/null +++ b/lib/crates/fabro-db/src/legacy.rs @@ -0,0 +1,79 @@ +//! Helpers shared by the one-time imports that seed SQLite tables from +//! pre-SQLite on-disk stores. + +use std::ffi::OsString; +use std::path::{Path, PathBuf}; + +use chrono::{DateTime, Utc}; +use tokio::fs; + +/// Failure to move an imported legacy source aside, carrying the backup path +/// the caller needs for its own error variant. +#[derive(Debug)] +pub struct LegacyBackupError { + pub backup_path: PathBuf, + pub source: std::io::Error, +} + +/// Move an imported legacy file or directory aside to +/// `.imported-.bak` next to the original. `fallback_name` is +/// used when the source path has no final component. +pub async fn rename_to_legacy_backup( + source: &Path, + fallback_name: &str, +) -> Result { + let backup_path = legacy_backup_path(source, fallback_name, Utc::now()); + fs::rename(source, &backup_path) + .await + .map_err(|source| LegacyBackupError { + backup_path: backup_path.clone(), + source, + })?; + Ok(backup_path) +} + +fn legacy_backup_path(source: &Path, fallback_name: &str, imported_at: DateTime) -> PathBuf { + let timestamp = imported_at.format("%Y%m%dT%H%M%S%fZ"); + let mut file_name = source + .file_name() + .map_or_else(|| OsString::from(fallback_name), OsString::from); + file_name.push(format!(".imported-{timestamp}.bak")); + source.with_file_name(file_name) +} + +/// True when `path` has a `.toml` extension (legacy per-item store files). +pub fn is_toml_file(path: &Path) -> bool { + path.extension() + .and_then(|extension| extension.to_str()) + .is_some_and(|extension| extension == "toml") +} + +#[cfg(test)] +mod tests { + use chrono::TimeZone as _; + + use super::*; + + #[test] + fn backup_path_appends_timestamped_suffix() { + let imported_at = Utc.with_ymd_and_hms(2026, 7, 11, 1, 2, 3).unwrap(); + let backup = + legacy_backup_path(Path::new("/data/secrets.json"), "secrets.json", imported_at); + assert_eq!( + backup, + Path::new("/data/secrets.json.imported-20260711T010203000000000Z.bak") + ); + } + + #[test] + fn backup_path_uses_fallback_when_source_has_no_file_name() { + let imported_at = Utc.with_ymd_and_hms(2026, 7, 11, 1, 2, 3).unwrap(); + let backup = legacy_backup_path(Path::new("/"), "mcps", imported_at); + assert!( + backup + .file_name() + .and_then(|name| name.to_str()) + .is_some_and(|name| name.starts_with("mcps.imported-")) + ); + } +} diff --git a/lib/crates/fabro-db/src/lib.rs b/lib/crates/fabro-db/src/lib.rs index 8765c82ea..ba930523e 100644 --- a/lib/crates/fabro-db/src/lib.rs +++ b/lib/crates/fabro-db/src/lib.rs @@ -3,6 +3,7 @@ use std::path::{Path, PathBuf}; use std::time::Duration; use anyhow::Context as _; +use chrono::{DateTime, Utc}; use sqlx::migrate::{Migrate as _, Migrator}; use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions, SqliteSynchronous}; use tokio::fs; @@ -10,8 +11,15 @@ use tokio::fs; use tokio::task::spawn_blocking; use tracing::info; +pub mod legacy; + pub type DbPool = sqlx::SqlitePool; +/// Parse an RFC 3339 TEXT column value into a UTC timestamp. +pub fn parse_rfc3339_utc(value: &str) -> Result, chrono::ParseError> { + DateTime::parse_from_rfc3339(value).map(|timestamp| timestamp.with_timezone(&Utc)) +} + static MIGRATOR: Migrator = sqlx::migrate!("./migrations"); #[derive(Clone)] diff --git a/lib/crates/fabro-db/tests/sqlite.rs b/lib/crates/fabro-db/tests/sqlite.rs index 53fc0d803..60c371ffa 100644 --- a/lib/crates/fabro-db/tests/sqlite.rs +++ b/lib/crates/fabro-db/tests/sqlite.rs @@ -49,6 +49,13 @@ async fn connect_creates_parent_directory_and_migrate_is_idempotent() -> anyhow: .await?; assert_eq!(secrets_table_count, 1); + let mcp_servers_table_count: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'mcp_servers'", + ) + .fetch_one(database.pool()) + .await?; + assert_eq!(mcp_servers_table_count, 1); + let legacy_import_table_count: i64 = sqlx::query_scalar( "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'legacy_imports'", ) @@ -65,6 +72,156 @@ async fn connect_creates_parent_directory_and_migrate_is_idempotent() -> anyhow: Ok(()) } +#[tokio::test] +async fn mcp_servers_schema_rejects_invalid_transport_rows() -> anyhow::Result<()> { + let dir = tempfile::tempdir()?; + let database = fabro_db::Database::connect(dir.path().join("fabro.sqlite3")).await?; + database.migrate().await?; + + insert_mcp_server( + database.pool(), + "stdio", + "stdio", + None, + Some(r#"["server"]"#), + None, + None, + Some("{}"), + None, + ) + .await?; + insert_mcp_server( + database.pool(), + "http", + "http", + Some("streamable_http"), + None, + Some("https://example.com/mcp"), + None, + None, + Some("{}"), + ) + .await?; + insert_mcp_server( + database.pool(), + "sandbox", + "sandbox", + Some("sse"), + Some(r#"["server"]"#), + None, + Some(3000), + Some("{}"), + None, + ) + .await?; + + for result in [ + insert_mcp_server( + database.pool(), + "bad-id_", + "stdio", + None, + Some(r#"["server"]"#), + None, + None, + Some("{}"), + None, + ) + .await, + insert_mcp_server( + database.pool(), + "empty-command", + "stdio", + None, + Some("[]"), + None, + None, + Some("{}"), + None, + ) + .await, + insert_mcp_server( + database.pool(), + "http-with-env", + "http", + Some("streamable_http"), + None, + Some("https://example.com/mcp"), + None, + Some("{}"), + Some("{}"), + ) + .await, + insert_mcp_server( + database.pool(), + "sandbox-port", + "sandbox", + Some("streamable_http"), + Some(r#"["server"]"#), + None, + Some(65_536), + Some("{}"), + None, + ) + .await, + ] { + assert!(result.is_err(), "invalid MCP server row should be rejected"); + } + + Ok(()) +} + +#[expect( + clippy::too_many_arguments, + reason = "schema test helper mirrors the mutually exclusive transport columns" +)] +async fn insert_mcp_server( + pool: &fabro_db::DbPool, + id: &str, + transport_type: &str, + protocol: Option<&str>, + command_json: Option<&str>, + url: Option<&str>, + port: Option, + env_json: Option<&str>, + headers_json: Option<&str>, +) -> Result<(), sqlx::Error> { + sqlx::query( + r" + INSERT INTO mcp_servers ( + id, + revision, + display_name, + transport_type, + protocol, + command_json, + url, + port, + env_json, + headers_json, + startup_timeout_secs, + tool_timeout_secs + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ", + ) + .bind(id) + .bind("a".repeat(64)) + .bind("MCP Server") + .bind(transport_type) + .bind(protocol) + .bind(command_json) + .bind(url) + .bind(port) + .bind(env_json) + .bind(headers_json) + .bind(10_i64) + .bind(60_i64) + .execute(pool) + .await?; + Ok(()) +} + #[tokio::test] async fn environments_schema_rejects_invalid_rows() -> anyhow::Result<()> { let dir = tempfile::tempdir()?; diff --git a/lib/crates/fabro-environment/src/store.rs b/lib/crates/fabro-environment/src/store.rs index 8d465d459..798231a28 100644 --- a/lib/crates/fabro-environment/src/store.rs +++ b/lib/crates/fabro-environment/src/store.rs @@ -1,15 +1,13 @@ use std::collections::{BTreeMap, HashMap, HashSet}; -use std::ffi::OsString; use std::path::{Path, PathBuf}; use std::str::FromStr; use std::sync::Arc; -use chrono::{DateTime, Utc}; use fabro_config::{ EnvironmentDockerfileLayer, EnvironmentImageLayer, EnvironmentLayer, EnvironmentLifecycleLayer, EnvironmentNetworkLayer, EnvironmentResourcesLayer, MergeMap, StickyMap, }; -use fabro_db::DbPool; +use fabro_db::{DbPool, legacy}; use fabro_types::settings::run::{DockerfileSource, EnvironmentProvider, EnvironmentSettings}; use fabro_types::settings::{Duration, InterpString, Size}; use serde::de::DeserializeOwned; @@ -706,7 +704,7 @@ async fn legacy_environment_paths( .file_type() .await .map_err(|source| EnvironmentStoreError::io(&path, source))?; - if file_type.is_file() && is_toml_file(&path) { + if file_type.is_file() && legacy::is_toml_file(&path) { paths.push(LegacyEnvironmentPath { id: id_from_path(&path)?, path, @@ -756,20 +754,9 @@ async fn existing_environment_ids( async fn rename_imported_legacy_directory( source_dir: &Path, ) -> Result { - let backup_path = legacy_backup_path(source_dir, Utc::now()); - fs::rename(source_dir, &backup_path) + legacy::rename_to_legacy_backup(source_dir, "environments") .await - .map_err(|source| EnvironmentStoreError::io(&backup_path, source))?; - Ok(backup_path) -} - -fn legacy_backup_path(source_dir: &Path, imported_at: DateTime) -> PathBuf { - let timestamp = imported_at.format("%Y%m%dT%H%M%S%fZ"); - let mut file_name = source_dir - .file_name() - .map_or_else(|| OsString::from("environments"), OsString::from); - file_name.push(format!(".imported-{timestamp}.bak")); - source_dir.with_file_name(file_name) + .map_err(|err| EnvironmentStoreError::io(&err.backup_path, err.source)) } fn id_from_path(path: &Path) -> Result { @@ -786,12 +773,6 @@ fn id_from_path(path: &Path) -> Result { }) } -fn is_toml_file(path: &Path) -> bool { - path.extension() - .and_then(|extension| extension.to_str()) - .is_some_and(|extension| extension == "toml") -} - fn encode_json( field: &'static str, value: &T, diff --git a/lib/crates/fabro-mcp-store/Cargo.toml b/lib/crates/fabro-mcp-store/Cargo.toml index 2fe21ece3..0172d9bae 100644 --- a/lib/crates/fabro-mcp-store/Cargo.toml +++ b/lib/crates/fabro-mcp-store/Cargo.toml @@ -13,11 +13,17 @@ doctest = false workspace = true [dependencies] +chrono.workspace = true +fabro-db = { path = "../fabro-db" } fabro-types = { path = "../fabro-types" } serde.workspace = true +serde_json.workspace = true +sqlx.workspace = true +strum.workspace = true thiserror.workspace = true tokio.workspace = true toml.workspace = true +tracing.workspace = true [dev-dependencies] tempfile = "3" diff --git a/lib/crates/fabro-mcp-store/src/error.rs b/lib/crates/fabro-mcp-store/src/error.rs index f4956b18d..9949fe9b5 100644 --- a/lib/crates/fabro-mcp-store/src/error.rs +++ b/lib/crates/fabro-mcp-store/src/error.rs @@ -21,6 +21,38 @@ pub enum McpServerStoreError { #[from] source: McpServerValidationError, }, + #[error("mcp server database error")] + Db { + #[from] + source: sqlx::Error, + }, + #[error("stored mcp server {id} has invalid revision")] + StoredRevision { + id: McpServerId, + #[source] + source: fabro_types::McpServerRevisionParseError, + }, + #[error("stored mcp server {id} has invalid transport: {reason}")] + StoredTransport { id: McpServerId, reason: String }, + #[error("stored mcp server {id} has invalid {column} value {value}")] + StoredInteger { + id: McpServerId, + column: &'static str, + value: i64, + }, + #[error("encoding mcp server {field} as JSON")] + JsonEncode { + field: &'static str, + #[source] + source: serde_json::Error, + }, + #[error("decoding stored mcp server {id}.{field} JSON")] + JsonDecode { + id: McpServerId, + field: &'static str, + #[source] + source: serde_json::Error, + }, #[error("invalid mcp server filename at {path:?}")] InvalidFilename { path: PathBuf, reason: String }, #[error("failed to parse mcp server TOML at {path:?}")] @@ -46,6 +78,13 @@ pub enum McpServerStoreError { #[source] source: std::io::Error, }, + #[error("renaming legacy mcp server directory {source_path:?} to backup {backup_path:?}")] + LegacyBackup { + source_path: PathBuf, + backup_path: PathBuf, + #[source] + source: std::io::Error, + }, } impl McpServerStoreError { @@ -77,10 +116,15 @@ impl McpServerStoreError { Self::AlreadyExists { .. } => "already_exists", Self::StaleRevision { .. } => "stale_revision", Self::Validation { .. } => "validation", + Self::Db { .. } => "database", + Self::StoredRevision { .. } + | Self::StoredTransport { .. } + | Self::StoredInteger { .. } + | Self::JsonDecode { .. } => "stored_data", + Self::JsonEncode { .. } | Self::Serialize { .. } => "serialize", Self::InvalidFilename { .. } => "invalid_filename", Self::Parse { .. } | Self::InvalidUtf8 { .. } => "parse", - Self::Serialize { .. } => "serialize", - Self::Io { .. } => "io", + Self::Io { .. } | Self::LegacyBackup { .. } => "io", } } } diff --git a/lib/crates/fabro-mcp-store/src/lib.rs b/lib/crates/fabro-mcp-store/src/lib.rs index 7000a5fc5..793c84ae4 100644 --- a/lib/crates/fabro-mcp-store/src/lib.rs +++ b/lib/crates/fabro-mcp-store/src/lib.rs @@ -1,14 +1,13 @@ //! Durable storage for server-managed MCP server definitions. //! -//! Concrete [`McpServerStore`] modeled on `fabro-automation`'s -//! `AutomationStore`: per-file TOML under `{config}/mcps/{id}.toml`, in-memory -//! cache, SHA-256 revision for optimistic concurrency, and async -//! storage-agnostic methods. The domain model lives in `fabro-types`; this -//! crate owns persistence. +//! Concrete SQLite-backed [`McpServerStore`] with a synchronous catalog cache, +//! SHA-256 revisions for optimistic concurrency, and one-time import from the +//! legacy per-file TOML directory. The domain model lives in `fabro-types`; +//! this crate owns persistence. mod error; mod model; mod store; pub use error::McpServerStoreError; -pub use store::McpServerStore; +pub use store::{ImportReport, McpServerStore, import_legacy_directory_once}; diff --git a/lib/crates/fabro-mcp-store/src/model.rs b/lib/crates/fabro-mcp-store/src/model.rs index 5ae656b2b..c467abb60 100644 --- a/lib/crates/fabro-mcp-store/src/model.rs +++ b/lib/crates/fabro-mcp-store/src/model.rs @@ -3,8 +3,8 @@ //! The domain types (`McpServerDefinition`, `McpServerDraft`, //! `McpServerReplace`, `McpServerId`, `McpServerRevision`) live in //! `fabro-types` so they stay persistence-independent. This module owns the -//! store-side glue: validating, serializing to canonical TOML bytes, deriving -//! the revision, and reconstructing definitions from persisted bytes. +//! store-side glue for validating definitions, deriving revisions from +//! canonical TOML bytes, and reconstructing definitions during legacy import. use std::path::PathBuf; @@ -12,12 +12,14 @@ use fabro_types::settings::McpTransport; use fabro_types::{ McpServerDefinition, McpServerId, McpServerReplace, McpServerRevision, mcp_store, }; +use serde::de::Error as _; use serde::{Deserialize, Serialize}; +use toml::de::Error as TomlDeError; use crate::error::McpServerStoreError; -/// The on-disk body of a definition. Excludes `id`/`revision`, which are -/// derived from the filename and content hash rather than persisted. +/// The legacy TOML body of a definition. It excludes `id`/`revision`, which old +/// installations derived from the filename and content hash. #[derive(Debug, Clone, PartialEq, Deserialize)] #[serde(deny_unknown_fields)] struct PersistedMcpServer { @@ -63,11 +65,10 @@ impl From for McpServerReplace { } } -/// Build a definition + its canonical persisted bytes from a replace payload. +/// Build a definition + its canonical revision bytes from a replace payload. /// /// The revision is the SHA-256 of the freshly serialized canonical bytes, so a -/// caller can compare it to the on-disk content hash for optimistic -/// concurrency. +/// caller can use it for optimistic concurrency. pub(crate) fn definition_from_replace( id: McpServerId, replace: McpServerReplace, @@ -94,6 +95,17 @@ pub(crate) fn definition_from_persisted_path( Ok(assemble(id, revision, replace)) } +/// Reconstruct a definition from normalized durable fields, revalidating the +/// same domain invariants enforced for API writes and legacy TOML imports. +pub(crate) fn definition_from_stored_parts( + id: McpServerId, + revision: McpServerRevision, + replace: McpServerReplace, +) -> Result { + mcp_store::validate_mcp_server_fields(&replace)?; + Ok(assemble(id, revision, replace)) +} + fn assemble( id: McpServerId, revision: McpServerRevision, @@ -119,5 +131,11 @@ pub(crate) fn canonical_bytes(replace: &McpServerReplace) -> Result, Mcp fn parse_persisted(bytes: &[u8], path: PathBuf) -> Result { let content = std::str::from_utf8(bytes) .map_err(|err| McpServerStoreError::invalid_utf8(path.clone(), err))?; - toml::from_str(content).map_err(|err| McpServerStoreError::parse(path, err)) + toml::from_str(content).map_err(|err| { + // TOML parse errors retain and display the source line by default. + // MCP transport lines may contain literal credentials, so retain only + // the parser's safe reason and discard its source-text context. + let safe = TomlDeError::custom(err.message()); + McpServerStoreError::parse(path, safe) + }) } diff --git a/lib/crates/fabro-mcp-store/src/store.rs b/lib/crates/fabro-mcp-store/src/store.rs index 7ba18c2e0..321ef88d1 100644 --- a/lib/crates/fabro-mcp-store/src/store.rs +++ b/lib/crates/fabro-mcp-store/src/store.rs @@ -1,48 +1,90 @@ -use std::collections::HashMap; +use std::collections::{BTreeMap, HashMap}; use std::io::ErrorKind; use std::path::{Path, PathBuf}; +use std::str::FromStr as _; use std::sync::RwLock; -use std::time::{SystemTime, UNIX_EPOCH}; -use fabro_types::settings::run::McpServerSettings; +use fabro_db::{DbPool, legacy}; +use fabro_types::settings::run::{McpHttpProtocol, McpServerSettings, McpTransport}; use fabro_types::{ McpServerDefinition, McpServerDraft, McpServerId, McpServerReplace, McpServerRevision, - McpServerView, + McpServerValidationError, McpServerView, }; +use serde::de::DeserializeOwned; +use sqlx::query::Query; +use sqlx::sqlite::{SqliteArguments, SqliteRow}; +use sqlx::{Row as _, Sqlite, Transaction}; +use strum::{Display, EnumString, IntoStaticStr}; use tokio::fs; -use tokio::io::AsyncWriteExt as _; use tokio::sync::Mutex; +use tracing::info; use crate::error::McpServerStoreError; use crate::model; -/// Durable per-file TOML store for server-managed MCP server definitions. +/// SQLite-backed durable store for server-managed MCP server definitions. /// -/// Concrete by design (no trait): a future FS→SQL move is a one-time migration, -/// not a runtime backend choice. The method surface keeps callers storage- -/// agnostic; only [`McpServerStore::load`] knows about the filesystem. -#[derive(Debug)] +/// Reads use a synchronous in-memory catalog because manifest resolution is +/// synchronous. Mutations are serialized within this process and use +/// revision-guarded SQL so SQLite remains authoritative for concurrency. +/// Writes to the same database from other processes are not observed until +/// this store is reloaded. pub struct McpServerStore { - dir: PathBuf, + pool: DbPool, mutations: Mutex<()>, defs: RwLock>, } +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ImportReport { + pub source_path: PathBuf, + pub backup_path: PathBuf, + pub imported_rows: usize, + pub skipped_rows: usize, + pub mcp_server_ids: Vec, +} + +#[derive(Debug, Clone, Copy, Display, EnumString, IntoStaticStr)] +#[strum(serialize_all = "snake_case")] +enum TransportType { + Stdio, + Http, + Sandbox, +} + +impl TransportType { + fn as_str(self) -> &'static str { + self.into() + } +} + +impl std::fmt::Debug for McpServerStore { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("McpServerStore").finish_non_exhaustive() + } +} + impl McpServerStore { - /// Synchronously load every persisted definition in `dir`. Returns an error - /// if any file fails to parse or validate; the caller decides startup - /// failure policy. Synchronous because it runs once at construction time - /// (typically during server startup) and is invoked from non-async code. - pub fn load(dir: impl Into) -> Result { - let dir = dir.into(); - let defs = load_definitions(&dir)?; + /// Load every persisted definition and build the synchronous catalog. + pub async fn load(pool: DbPool) -> Result { + let defs = load_definitions(&pool).await?; Ok(Self { - dir, + pool, mutations: Mutex::new(()), defs: RwLock::new(defs), }) } + /// Import a legacy `mcps/*.toml` directory when present, then load the + /// SQLite-backed catalog. + pub async fn open( + pool: DbPool, + legacy_dir: impl AsRef, + ) -> Result { + import_legacy_directory_once(&pool, legacy_dir).await?; + Self::load(pool).await + } + fn read_defs( &self, ) -> std::sync::RwLockReadGuard<'_, HashMap> { @@ -76,9 +118,8 @@ impl McpServerStore { .collect() } - /// Sorted ids only, without cloning the (potentially sensitive) env/header - /// maps carried by full definitions. Used by missing-reference errors to - /// list available ids cheaply. + /// Sorted ids only, without cloning the potentially sensitive env/header + /// maps carried by full definitions. pub fn ids(&self) -> Vec { let defs = self.read_defs(); let mut ids = defs.keys().cloned().collect::>(); @@ -95,19 +136,14 @@ impl McpServerStore { draft: McpServerDraft, ) -> Result { let (id, replace) = draft.into(); + let (definition, _) = model::definition_from_replace(id.clone(), replace)?; let _mutation = self.mutations.lock().await; - if self.read_defs().contains_key(&id) { + let mut transaction = self.pool.begin().await?; + if !insert_definition_ignoring_conflict(&mut transaction, &definition).await? { return Err(McpServerStoreError::AlreadyExists { id }); } - let (definition, bytes) = model::definition_from_replace(id.clone(), replace)?; - - let path = definition_path(&self.dir, &id); - write_new(&self.dir, &path, &bytes) - .await - .map_err(|err| create_error_for(id.clone(), err))?; - - let mut defs = self.write_defs(); - defs.insert(id, definition.clone()); + transaction.commit().await?; + self.write_defs().insert(id, definition.clone()); Ok(definition) } @@ -117,16 +153,12 @@ impl McpServerStore { expected: &McpServerRevision, replace: McpServerReplace, ) -> Result { + let (definition, _) = model::definition_from_replace(id.clone(), replace)?; let _mutation = self.mutations.lock().await; - { - let defs = self.read_defs(); - check_revision(&defs, id, expected)?; - } - let (definition, bytes) = model::definition_from_replace(id.clone(), replace)?; - - write_atomic(&self.dir, &definition_path(&self.dir, id), &bytes).await?; - let mut defs = self.write_defs(); - defs.insert(id.clone(), definition.clone()); + let mut transaction = self.pool.begin().await?; + update_definition(&mut transaction, &definition, expected).await?; + transaction.commit().await?; + self.write_defs().insert(id.clone(), definition.clone()); Ok(definition) } @@ -136,39 +168,218 @@ impl McpServerStore { expected: &McpServerRevision, ) -> Result<(), McpServerStoreError> { let _mutation = self.mutations.lock().await; - { - let defs = self.read_defs(); - check_revision(&defs, id, expected)?; + let mut transaction = self.pool.begin().await?; + let result = sqlx::query("DELETE FROM mcp_servers WHERE id = ? AND revision = ?") + .bind(id.as_str()) + .bind(expected.as_str()) + .execute(&mut *transaction) + .await?; + if result.rows_affected() == 0 { + return Err(revision_mismatch_error(&mut transaction, id, expected).await?); } - - let path = definition_path(&self.dir, id); - fs::remove_file(&path) - .await - .map_err(|err| McpServerStoreError::io(path, err))?; - let mut defs = self.write_defs(); - defs.remove(id); + transaction.commit().await?; + self.write_defs().remove(id); Ok(()) } } -fn check_revision( - defs: &HashMap, +async fn load_definitions( + pool: &DbPool, +) -> Result, McpServerStoreError> { + let rows = sqlx::query( + r" + SELECT + id, + revision, + display_name, + description, + transport_type, + protocol, + command_json, + url, + port, + env_json, + headers_json, + startup_timeout_secs, + tool_timeout_secs + FROM mcp_servers + ORDER BY id + ", + ) + .fetch_all(pool) + .await?; + + rows.iter() + .map(definition_from_row) + .map(|result| result.map(|definition| (definition.id.clone(), definition))) + .collect() +} + +fn definition_from_row(row: &SqliteRow) -> Result { + let id = McpServerId::new(row.try_get::("id")?)?; + let revision_text = row.try_get::("revision")?; + let revision = McpServerRevision::from_str(&revision_text).map_err(|source| { + McpServerStoreError::StoredRevision { + id: id.clone(), + source, + } + })?; + let transport_type_text = row.try_get::("transport_type")?; + let transport_type = TransportType::from_str(&transport_type_text).map_err(|_| { + McpServerStoreError::StoredTransport { + id: id.clone(), + reason: format!("unknown transport type {transport_type_text:?}"), + } + })?; + let protocol = row.try_get::, _>("protocol")?; + let command_json = row.try_get::, _>("command_json")?; + let url = row.try_get::, _>("url")?; + let port = row.try_get::, _>("port")?; + let env_json = row.try_get::, _>("env_json")?; + let headers_json = row.try_get::, _>("headers_json")?; + let transport = match transport_type { + TransportType::Stdio => { + require_absent(&id, "protocol", protocol.as_ref())?; + require_absent(&id, "url", url.as_ref())?; + require_absent(&id, "port", port.as_ref())?; + require_absent(&id, "headers_json", headers_json.as_ref())?; + McpTransport::Stdio { + command: decode_json( + &id, + "command_json", + &require_field(&id, "command_json", command_json)?, + )?, + env: decode_string_map( + &id, + "env_json", + &require_field(&id, "env_json", env_json)?, + )?, + } + } + TransportType::Http => { + require_absent(&id, "command_json", command_json.as_ref())?; + require_absent(&id, "port", port.as_ref())?; + require_absent(&id, "env_json", env_json.as_ref())?; + McpTransport::Http { + protocol: parse_protocol(&id, require_field(&id, "protocol", protocol)?.as_str())?, + url: require_field(&id, "url", url)?, + headers: decode_string_map( + &id, + "headers_json", + &require_field(&id, "headers_json", headers_json)?, + )?, + } + } + TransportType::Sandbox => { + require_absent(&id, "url", url.as_ref())?; + require_absent(&id, "headers_json", headers_json.as_ref())?; + McpTransport::Sandbox { + protocol: parse_protocol(&id, require_field(&id, "protocol", protocol)?.as_str())?, + command: decode_json( + &id, + "command_json", + &require_field(&id, "command_json", command_json)?, + )?, + port: decode_port(&id, require_field(&id, "port", port)?)?, + env: decode_string_map( + &id, + "env_json", + &require_field(&id, "env_json", env_json)?, + )?, + } + } + }; + let replace = McpServerReplace { + display_name: row.try_get("display_name")?, + description: row.try_get("description")?, + transport, + startup_timeout_secs: decode_timeout( + &id, + "startup_timeout_secs", + row.try_get("startup_timeout_secs")?, + )?, + tool_timeout_secs: decode_timeout( + &id, + "tool_timeout_secs", + row.try_get("tool_timeout_secs")?, + )?, + }; + model::definition_from_stored_parts(id, revision, replace) +} + +fn require_field( id: &McpServerId, - expected: &McpServerRevision, + field: &'static str, + value: Option, +) -> Result { + value.ok_or_else(|| McpServerStoreError::StoredTransport { + id: id.clone(), + reason: format!("{field} is missing"), + }) +} + +fn require_absent( + id: &McpServerId, + field: &'static str, + value: Option<&T>, ) -> Result<(), McpServerStoreError> { - let current = defs - .get(id) - .ok_or_else(|| McpServerStoreError::NotFound { id: id.clone() })?; - if ¤t.revision != expected { - return Err(McpServerStoreError::StaleRevision { - id: id.clone(), - expected: expected.clone(), - actual: current.revision.clone(), + if value.is_some() { + return Err(McpServerStoreError::StoredTransport { + id: id.clone(), + reason: format!("{field} must be absent"), }); } Ok(()) } +fn parse_protocol(id: &McpServerId, value: &str) -> Result { + McpHttpProtocol::from_str(value).map_err(|_| McpServerStoreError::StoredTransport { + id: id.clone(), + reason: format!("unknown protocol {value:?}"), + }) +} + +fn decode_port(id: &McpServerId, value: i64) -> Result { + u16::try_from(value).map_err(|_| McpServerStoreError::StoredInteger { + id: id.clone(), + column: "port", + value, + }) +} + +fn decode_timeout( + id: &McpServerId, + column: &'static str, + value: i64, +) -> Result { + u64::try_from(value).map_err(|_| McpServerStoreError::StoredInteger { + id: id.clone(), + column, + value, + }) +} + +fn decode_json( + id: &McpServerId, + field: &'static str, + value: &str, +) -> Result { + serde_json::from_str(value).map_err(|source| McpServerStoreError::JsonDecode { + id: id.clone(), + field, + source, + }) +} + +fn decode_string_map( + id: &McpServerId, + field: &'static str, + value: &str, +) -> Result, McpServerStoreError> { + let ordered = decode_json::>(id, field, value)?; + Ok(ordered.into_iter().collect()) +} + fn server_settings_from_definition(definition: &McpServerDefinition) -> McpServerSettings { McpServerSettings { name: definition.id.to_string(), @@ -180,43 +391,310 @@ fn server_settings_from_definition(definition: &McpServerDefinition) -> McpServe } } -#[expect( - clippy::disallowed_methods, - reason = "MCP server directory scan runs once at startup, before the runtime needs to make progress; std::fs avoids needing a Tokio runtime for the caller." -)] -fn load_definitions( - dir: &Path, -) -> Result, McpServerStoreError> { - let entries = match std::fs::read_dir(dir) { +async fn current_revision( + transaction: &mut Transaction<'_, Sqlite>, + id: &McpServerId, +) -> Result, McpServerStoreError> { + let revision: Option = + sqlx::query_scalar("SELECT revision FROM mcp_servers WHERE id = ?") + .bind(id.as_str()) + .fetch_optional(&mut **transaction) + .await?; + revision + .map(|revision| { + McpServerRevision::from_str(&revision).map_err(|source| { + McpServerStoreError::StoredRevision { + id: id.clone(), + source, + } + }) + }) + .transpose() +} + +async fn revision_mismatch_error( + transaction: &mut Transaction<'_, Sqlite>, + id: &McpServerId, + expected: &McpServerRevision, +) -> Result { + let Some(actual) = current_revision(transaction, id).await? else { + return Ok(McpServerStoreError::NotFound { id: id.clone() }); + }; + Ok(McpServerStoreError::StaleRevision { + id: id.clone(), + expected: expected.clone(), + actual, + }) +} + +async fn insert_definition_ignoring_conflict( + transaction: &mut Transaction<'_, Sqlite>, + definition: &McpServerDefinition, +) -> Result { + let row = McpServerSqlRow::from_definition(definition)?; + let result = bind_definition(sqlx::query(INSERT_DEFINITION_SQL), row) + .execute(&mut **transaction) + .await?; + Ok(result.rows_affected() == 1) +} + +async fn update_definition( + transaction: &mut Transaction<'_, Sqlite>, + definition: &McpServerDefinition, + expected: &McpServerRevision, +) -> Result<(), McpServerStoreError> { + let row = McpServerSqlRow::from_definition(definition)?; + let result = bind_definition(sqlx::query(UPDATE_DEFINITION_SQL), row) + .bind(expected.as_str()) + .execute(&mut **transaction) + .await?; + if result.rows_affected() == 0 { + return Err(revision_mismatch_error(transaction, &definition.id, expected).await?); + } + Ok(()) +} + +/// Binds the shared placeholder order of [`INSERT_DEFINITION_SQL`] and +/// [`UPDATE_DEFINITION_SQL`]: the twelve value columns first, then `id`. +fn bind_definition( + query: Query<'_, Sqlite, SqliteArguments>, + row: McpServerSqlRow, +) -> Query<'_, Sqlite, SqliteArguments> { + query + .bind(row.revision) + .bind(row.display_name) + .bind(row.description) + .bind(row.transport_type) + .bind(row.protocol) + .bind(row.command_json) + .bind(row.url) + .bind(row.port) + .bind(row.env_json) + .bind(row.headers_json) + .bind(row.startup_timeout_secs) + .bind(row.tool_timeout_secs) + .bind(row.id) +} + +struct McpServerSqlRow { + id: String, + revision: String, + display_name: String, + description: Option, + transport_type: &'static str, + protocol: Option<&'static str>, + command_json: Option, + url: Option, + port: Option, + env_json: Option, + headers_json: Option, + startup_timeout_secs: i64, + tool_timeout_secs: i64, +} + +impl McpServerSqlRow { + fn from_definition(definition: &McpServerDefinition) -> Result { + let (transport_type, protocol, command_json, url, port, env_json, headers_json) = + match &definition.transport { + McpTransport::Stdio { command, env } => ( + TransportType::Stdio, + None, + Some(encode_json("command_json", command)?), + None, + None, + Some(encode_string_map("env_json", env)?), + None, + ), + McpTransport::Http { + protocol, + url, + headers, + } => ( + TransportType::Http, + Some(protocol.as_str()), + None, + Some(url.clone()), + None, + None, + Some(encode_string_map("headers_json", headers)?), + ), + McpTransport::Sandbox { + protocol, + command, + port, + env, + } => ( + TransportType::Sandbox, + Some(protocol.as_str()), + Some(encode_json("command_json", command)?), + None, + Some(i64::from(*port)), + Some(encode_string_map("env_json", env)?), + None, + ), + }; + Ok(Self { + id: definition.id.to_string(), + revision: definition.revision.to_string(), + display_name: definition.display_name.clone(), + description: definition.description.clone(), + transport_type: transport_type.as_str(), + protocol, + command_json, + url, + port, + env_json, + headers_json, + startup_timeout_secs: encode_timeout( + "startup_timeout_secs", + definition.startup_timeout_secs, + )?, + tool_timeout_secs: encode_timeout("tool_timeout_secs", definition.tool_timeout_secs)?, + }) + } +} + +fn encode_timeout(field: &'static str, value: u64) -> Result { + i64::try_from(value) + .map_err(|_| McpServerValidationError::TimeoutOutOfRange { field, value }.into()) +} + +fn encode_json( + field: &'static str, + value: &T, +) -> Result { + serde_json::to_string(value).map_err(|source| McpServerStoreError::JsonEncode { field, source }) +} + +fn encode_string_map( + field: &'static str, + map: &HashMap, +) -> Result { + let ordered = map + .iter() + .map(|(key, value)| (key.as_str(), value.as_str())) + .collect::>(); + encode_json(field, &ordered) +} + +const INSERT_DEFINITION_SQL: &str = r" +INSERT INTO mcp_servers ( + revision, + display_name, + description, + transport_type, + protocol, + command_json, + url, + port, + env_json, + headers_json, + startup_timeout_secs, + tool_timeout_secs, + id +) +VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) +ON CONFLICT(id) DO NOTHING +"; + +const UPDATE_DEFINITION_SQL: &str = r" +UPDATE mcp_servers SET + revision = ?, + display_name = ?, + description = ?, + transport_type = ?, + protocol = ?, + command_json = ?, + url = ?, + port = ?, + env_json = ?, + headers_json = ?, + startup_timeout_secs = ?, + tool_timeout_secs = ? +WHERE id = ? AND revision = ? +"; + +pub async fn import_legacy_directory_once( + pool: &DbPool, + source_dir: impl AsRef, +) -> Result, McpServerStoreError> { + let source_dir = source_dir.as_ref(); + let Some(paths) = legacy_definition_paths(source_dir).await? else { + return Ok(None); + }; + let definitions = read_legacy_definitions(paths).await?; + + let mut transaction = pool.begin().await?; + let mut imported_ids = Vec::new(); + let mut skipped_rows = 0; + for definition in &definitions { + if insert_definition_ignoring_conflict(&mut transaction, definition).await? { + imported_ids.push(definition.id.to_string()); + } else { + skipped_rows += 1; + } + } + transaction.commit().await?; + + let backup_path = rename_imported_legacy_directory(source_dir).await?; + let report = ImportReport { + source_path: source_dir.to_path_buf(), + backup_path, + imported_rows: imported_ids.len(), + skipped_rows, + mcp_server_ids: imported_ids, + }; + info!( + source_path = %report.source_path.display(), + backup_path = %report.backup_path.display(), + imported_rows = report.imported_rows, + skipped_rows = report.skipped_rows, + mcp_server_ids = ?report.mcp_server_ids, + "Imported legacy MCP server directory into SQLite" + ); + Ok(Some(report)) +} + +async fn legacy_definition_paths( + source_dir: &Path, +) -> Result>, McpServerStoreError> { + let mut entries = match fs::read_dir(source_dir).await { Ok(entries) => entries, - Err(err) if err.kind() == ErrorKind::NotFound => return Ok(HashMap::new()), - Err(err) => return Err(McpServerStoreError::io(dir, err)), + Err(source) if source.kind() == ErrorKind::NotFound => return Ok(None), + Err(source) => return Err(McpServerStoreError::io(source_dir, source)), }; - let mut defs = HashMap::new(); - for entry in entries { - let entry = entry.map_err(|err| McpServerStoreError::io(dir, err))?; + let mut paths = Vec::new(); + while let Some(entry) = entries + .next_entry() + .await + .map_err(|source| McpServerStoreError::io(source_dir, source))? + { let path = entry.path(); let file_type = entry .file_type() - .map_err(|err| McpServerStoreError::io(&path, err))?; - if !file_type.is_file() || !is_toml_file(&path) { - continue; + .await + .map_err(|source| McpServerStoreError::io(&path, source))?; + if file_type.is_file() && legacy::is_toml_file(&path) { + paths.push((id_from_path(&path)?, path)); } - let definition = load_definition_file(&path)?; - defs.insert(definition.id.clone(), definition); } - Ok(defs) + paths.sort_by(|left, right| left.1.cmp(&right.1)); + Ok(Some(paths)) } -#[expect( - clippy::disallowed_methods, - reason = "Sync sibling of `load_definitions`; only invoked from the synchronous startup load path." -)] -fn load_definition_file(path: &Path) -> Result { - let id = id_from_path(path)?; - let bytes = std::fs::read(path).map_err(|err| McpServerStoreError::io(path, err))?; - model::definition_from_persisted_path(id, &bytes, path) +async fn read_legacy_definitions( + paths: Vec<(McpServerId, PathBuf)>, +) -> Result, McpServerStoreError> { + let mut definitions = Vec::with_capacity(paths.len()); + for (id, path) in paths { + let bytes = fs::read(&path) + .await + .map_err(|source| McpServerStoreError::io(&path, source))?; + definitions.push(model::definition_from_persisted_path(id, &bytes, path)?); + } + definitions.sort_by(|left, right| left.id.cmp(&right.id)); + Ok(definitions) } fn id_from_path(path: &Path) -> Result { @@ -233,265 +711,14 @@ fn id_from_path(path: &Path) -> Result { }) } -fn is_toml_file(path: &Path) -> bool { - path.extension() - .and_then(|extension| extension.to_str()) - .is_some_and(|extension| extension == "toml") -} - -async fn write_atomic(dir: &Path, path: &Path, bytes: &[u8]) -> Result<(), McpServerStoreError> { - let temp_path = write_temp_file(dir, path, bytes).await?; - if let Err(err) = fs::rename(&temp_path, path).await { - cleanup_temp(&temp_path).await; - return Err(McpServerStoreError::io(path, err)); - } - - Ok(()) -} - -async fn write_new(dir: &Path, path: &Path, bytes: &[u8]) -> Result<(), McpServerStoreError> { - let temp_path = write_temp_file(dir, path, bytes).await?; - if let Err(err) = fs::hard_link(&temp_path, path).await { - cleanup_temp(&temp_path).await; - return Err(McpServerStoreError::io(path, err)); - } - cleanup_temp(&temp_path).await; - Ok(()) -} - -async fn write_temp_file( - dir: &Path, - path: &Path, - bytes: &[u8], +async fn rename_imported_legacy_directory( + source_dir: &Path, ) -> Result { - fs::create_dir_all(dir) + legacy::rename_to_legacy_backup(source_dir, "mcps") .await - .map_err(|err| McpServerStoreError::io(dir, err))?; - let temp_path = temp_path_for(path); - let mut file = fs::OpenOptions::new() - .write(true) - .create_new(true) - .open(&temp_path) - .await - .map_err(|err| McpServerStoreError::io(&temp_path, err))?; - - if let Err(err) = file.write_all(bytes).await { - cleanup_temp(&temp_path).await; - return Err(McpServerStoreError::io(&temp_path, err)); - } - if let Err(err) = file.sync_all().await { - cleanup_temp(&temp_path).await; - return Err(McpServerStoreError::io(&temp_path, err)); - } - drop(file); - - Ok(temp_path) -} - -async fn cleanup_temp(path: &Path) { - let _ = fs::remove_file(path).await; -} - -fn create_error_for(id: McpServerId, err: McpServerStoreError) -> McpServerStoreError { - match err { - McpServerStoreError::Io { source, .. } if source.kind() == ErrorKind::AlreadyExists => { - McpServerStoreError::AlreadyExists { id } - } - err => err, - } -} - -fn temp_path_for(path: &Path) -> PathBuf { - let parent = path.parent().unwrap_or_else(|| Path::new(".")); - let file_name = path - .file_name() - .and_then(|name| name.to_str()) - .unwrap_or("mcp-server.toml"); - let now = SystemTime::now() - .duration_since(UNIX_EPOCH) - .map_or(0, |duration| duration.as_nanos()); - parent.join(format!(".{file_name}.{}.{}.tmp", std::process::id(), now)) -} - -fn definition_path(dir: &Path, id: &McpServerId) -> PathBuf { - dir.join(format!("{id}.toml")) -} - -#[cfg(test)] -mod tests { - use std::collections::HashMap; - - use fabro_types::settings::McpTransport; - use fabro_types::settings::run::McpHttpProtocol; - use fabro_types::{McpServerDraft, McpServerId, McpServerReplace, McpServerRevision}; - use tokio::fs; - - use crate::error::McpServerStoreError; - use crate::store::McpServerStore; - - fn http_transport(url: &str) -> McpTransport { - McpTransport::Http { - protocol: McpHttpProtocol::default(), - url: url.to_string(), - headers: HashMap::new(), - } - } - - fn draft(id: &str, display_name: &str) -> McpServerDraft { - McpServerDraft { - id: McpServerId::new(id).unwrap(), - display_name: display_name.to_string(), - description: None, - transport: http_transport("https://example.com/mcp"), - startup_timeout_secs: 10, - tool_timeout_secs: 60, - } - } - - fn replacement(display_name: &str) -> McpServerReplace { - McpServerReplace { - display_name: display_name.to_string(), - description: Some("updated".to_string()), - transport: http_transport("https://example.com/mcp/v2"), - startup_timeout_secs: 15, - tool_timeout_secs: 90, - } - } - - #[tokio::test] - async fn missing_directory_loads_empty_store() { - let dir = tempfile::tempdir().unwrap(); - let store = McpServerStore::load(dir.path().join("mcps")).unwrap(); - - assert!(store.list().is_empty()); - assert!(store.ids().is_empty()); - } - - #[tokio::test] - async fn load_ignores_non_toml_files_and_keeps_valid_definitions() { - let dir = tempfile::tempdir().unwrap(); - let mcp_dir = dir.path().join("mcps"); - fs::create_dir_all(&mcp_dir).await.unwrap(); - fs::write(mcp_dir.join("notes.txt"), "ignore") - .await - .unwrap(); - fs::write( - mcp_dir.join("sentry.toml"), - r#" -display_name = "Sentry" -startup_timeout_secs = 10 -tool_timeout_secs = 60 - -[transport] -type = "http" -url = "https://sentry.example.com/mcp" - -[transport.headers] -"#, - ) - .await - .unwrap(); - - let store = McpServerStore::load(&mcp_dir).unwrap(); - let defs = store.list(); - - assert_eq!(defs.len(), 1); - assert_eq!(defs[0].id.as_str(), "sentry"); - assert_eq!(defs[0].display_name, "Sentry"); - } - - #[tokio::test] - async fn load_fails_on_malformed_toml() { - let dir = tempfile::tempdir().unwrap(); - let mcp_dir = dir.path().join("mcps"); - fs::create_dir_all(&mcp_dir).await.unwrap(); - fs::write(mcp_dir.join("broken.toml"), "not valid toml =") - .await - .unwrap(); - - let err = McpServerStore::load(&mcp_dir).unwrap_err(); - assert!(matches!(err, McpServerStoreError::Parse { .. })); - } - - #[tokio::test] - async fn load_fails_on_invalid_filename_id() { - let dir = tempfile::tempdir().unwrap(); - let mcp_dir = dir.path().join("mcps"); - fs::create_dir_all(&mcp_dir).await.unwrap(); - fs::write(mcp_dir.join("Bad Name.toml"), "display_name = \"Bad\"") - .await - .unwrap(); - - let err = McpServerStore::load(&mcp_dir).unwrap_err(); - assert!(matches!(err, McpServerStoreError::InvalidFilename { .. })); - } - - #[tokio::test] - async fn create_get_list_replace_and_delete_round_trip_files_and_revisions() { - let dir = tempfile::tempdir().unwrap(); - let mcp_dir = dir.path().join("mcps"); - let store = McpServerStore::load(&mcp_dir).unwrap(); - - let created = store.create(draft("sentry", "Sentry")).await.unwrap(); - let path = mcp_dir.join("sentry.toml"); - let persisted = fs::read_to_string(&path).await.unwrap(); - assert!(persisted.contains("display_name = \"Sentry\"")); - assert!(!top_level_lines(&persisted).any(|line| line.starts_with("id = "))); - assert!(!top_level_lines(&persisted).any(|line| line.starts_with("revision = "))); - assert_eq!( - created.revision, - McpServerRevision::from_bytes(persisted.as_bytes()) - ); - - assert_eq!(store.get(&created.id).unwrap(), created); - let listed = store.list(); - assert_eq!(listed.len(), 1); - assert_eq!(store.ids(), vec![created.id.clone()]); - - let replaced = store - .replace(&created.id, &created.revision, replacement("Sentry v2")) - .await - .unwrap(); - assert_ne!(replaced.revision, created.revision); - assert_eq!(replaced.display_name, "Sentry v2"); - assert_eq!(store.get(&created.id).unwrap().revision, replaced.revision); - - store.delete(&created.id, &replaced.revision).await.unwrap(); - assert!(store.get(&created.id).is_none()); - assert!(!path.exists()); - } - - #[tokio::test] - async fn replace_with_stale_revision_is_rejected() { - let dir = tempfile::tempdir().unwrap(); - let store = McpServerStore::load(dir.path().join("mcps")).unwrap(); - let created = store.create(draft("sentry", "Sentry")).await.unwrap(); - - let stale = McpServerRevision::from_bytes(b"stale"); - let err = store - .replace(&created.id, &stale, replacement("Updated")) - .await - .unwrap_err(); - assert!(matches!(err, McpServerStoreError::StaleRevision { .. })); - - // The on-disk and in-memory definition is unchanged after a rejected replace. - assert_eq!(store.get(&created.id).unwrap(), created); - } - - #[tokio::test] - async fn duplicate_create_is_rejected() { - let dir = tempfile::tempdir().unwrap(); - let store = McpServerStore::load(dir.path().join("mcps")).unwrap(); - store.create(draft("sentry", "Sentry")).await.unwrap(); - - let err = store - .create(draft("sentry", "Duplicate")) - .await - .unwrap_err(); - assert!(matches!(err, McpServerStoreError::AlreadyExists { .. })); - } - - fn top_level_lines(toml: &str) -> impl Iterator { - toml.lines().take_while(|line| !line.starts_with('[')) - } + .map_err(|err| McpServerStoreError::LegacyBackup { + source_path: source_dir.to_path_buf(), + backup_path: err.backup_path, + source: err.source, + }) } diff --git a/lib/crates/fabro-mcp-store/tests/store.rs b/lib/crates/fabro-mcp-store/tests/store.rs new file mode 100644 index 000000000..f39cd1ea9 --- /dev/null +++ b/lib/crates/fabro-mcp-store/tests/store.rs @@ -0,0 +1,375 @@ +#![expect( + clippy::unwrap_used, + reason = "SQLite MCP store integration tests use panic-on-failure fixture setup" +)] + +use std::collections::HashMap; +use std::error::Error as _; + +use fabro_db::Database; +use fabro_mcp_store::{McpServerStore, McpServerStoreError, import_legacy_directory_once}; +use fabro_types::settings::run::{McpHttpProtocol, McpTransport}; +use fabro_types::{McpServerDraft, McpServerId, McpServerReplace, McpServerRevision}; +use tokio::fs; + +async fn test_database() -> (tempfile::TempDir, Database) { + let dir = tempfile::tempdir().unwrap(); + let database = Database::connect(dir.path().join("fabro.sqlite3")) + .await + .unwrap(); + database.migrate().await.unwrap(); + (dir, database) +} + +fn http_transport(url: &str) -> McpTransport { + McpTransport::Http { + protocol: McpHttpProtocol::default(), + url: url.to_string(), + headers: HashMap::new(), + } +} + +fn draft(id: &str, display_name: &str) -> McpServerDraft { + McpServerDraft { + id: McpServerId::new(id).unwrap(), + display_name: display_name.to_string(), + description: None, + transport: http_transport("https://example.com/mcp"), + startup_timeout_secs: 10, + tool_timeout_secs: 60, + } +} + +fn replacement(display_name: &str) -> McpServerReplace { + McpServerReplace { + display_name: display_name.to_string(), + description: Some("updated".to_string()), + transport: http_transport("https://example.com/mcp/v2"), + startup_timeout_secs: 15, + tool_timeout_secs: 90, + } +} + +fn legacy_toml(display_name: &str, url: &str) -> String { + format!( + r#"display_name = "{display_name}" +startup_timeout_secs = 10 +tool_timeout_secs = 60 + +[transport] +type = "http" +url = "{url}" + +[transport.headers] +"# + ) +} + +#[tokio::test] +async fn empty_database_loads_empty_store() { + let (_dir, database) = test_database().await; + let store = McpServerStore::load(database.clone_pool()).await.unwrap(); + + assert!(store.list().is_empty()); + assert!(store.ids().is_empty()); +} + +#[tokio::test] +async fn create_get_list_replace_delete_and_reload_round_trip() { + let (_dir, database) = test_database().await; + let store = McpServerStore::load(database.clone_pool()).await.unwrap(); + + let created = store.create(draft("sentry", "Sentry")).await.unwrap(); + assert_eq!(store.get(&created.id).unwrap(), created); + assert_eq!(store.list(), vec![created.clone()]); + assert_eq!(store.ids(), vec![created.id.clone()]); + + let reloaded = McpServerStore::load(database.clone_pool()).await.unwrap(); + assert_eq!(reloaded.get(&created.id).unwrap(), created); + + let replaced = store + .replace(&created.id, &created.revision, replacement("Sentry v2")) + .await + .unwrap(); + assert_ne!(replaced.revision, created.revision); + assert_eq!(replaced.display_name, "Sentry v2"); + + store + .delete(&replaced.id, &replaced.revision) + .await + .unwrap(); + assert!(store.get(&replaced.id).is_none()); + assert!( + McpServerStore::load(database.clone_pool()) + .await + .unwrap() + .list() + .is_empty() + ); +} + +#[tokio::test] +async fn all_transport_variants_round_trip_with_sorted_map_json() { + let (_dir, database) = test_database().await; + let store = McpServerStore::load(database.clone_pool()).await.unwrap(); + let definitions = [ + McpServerDraft { + id: McpServerId::new("stdio").unwrap(), + display_name: "Stdio".to_string(), + description: None, + transport: McpTransport::Stdio { + command: vec!["server".to_string(), "--flag".to_string()], + env: HashMap::from([ + ("Z_KEY".to_string(), "z".to_string()), + ("A_KEY".to_string(), "a".to_string()), + ]), + }, + startup_timeout_secs: 1, + tool_timeout_secs: 2, + }, + McpServerDraft { + id: McpServerId::new("http").unwrap(), + display_name: "HTTP".to_string(), + description: None, + transport: McpTransport::Http { + protocol: McpHttpProtocol::Sse, + url: "https://example.com/mcp".to_string(), + headers: HashMap::from([("Authorization".to_string(), "secret".to_string())]), + }, + startup_timeout_secs: 3, + tool_timeout_secs: 4, + }, + McpServerDraft { + id: McpServerId::new("sandbox").unwrap(), + display_name: "Sandbox".to_string(), + description: None, + transport: McpTransport::Sandbox { + protocol: McpHttpProtocol::StreamableHttp, + command: vec!["server".to_string()], + port: 3000, + env: HashMap::from([("TOKEN".to_string(), "secret".to_string())]), + }, + startup_timeout_secs: 5, + tool_timeout_secs: 6, + }, + ]; + + for definition in definitions.clone() { + store.create(definition).await.unwrap(); + } + let reloaded = McpServerStore::load(database.clone_pool()).await.unwrap(); + for expected in definitions { + let actual = reloaded.get(&expected.id).unwrap(); + assert_eq!(actual.display_name, expected.display_name); + assert_eq!(actual.transport, expected.transport); + } + + let env_json: String = + sqlx::query_scalar("SELECT env_json FROM mcp_servers WHERE id = 'stdio'") + .fetch_one(database.pool()) + .await + .unwrap(); + assert_eq!(env_json, r#"{"A_KEY":"a","Z_KEY":"z"}"#); +} + +#[tokio::test] +async fn duplicate_create_is_rejected() { + let (_dir, database) = test_database().await; + let first = McpServerStore::load(database.clone_pool()).await.unwrap(); + let second = McpServerStore::load(database.clone_pool()).await.unwrap(); + first.create(draft("sentry", "Sentry")).await.unwrap(); + + let err = second + .create(draft("sentry", "Duplicate")) + .await + .unwrap_err(); + assert!(matches!(err, McpServerStoreError::AlreadyExists { .. })); +} + +#[tokio::test] +async fn revision_guard_rejects_stale_independent_store() { + let (_dir, database) = test_database().await; + let creator = McpServerStore::load(database.clone_pool()).await.unwrap(); + let created = creator.create(draft("sentry", "Sentry")).await.unwrap(); + let first = McpServerStore::load(database.clone_pool()).await.unwrap(); + let second = McpServerStore::load(database.clone_pool()).await.unwrap(); + + first + .replace(&created.id, &created.revision, replacement("Winner")) + .await + .unwrap(); + let err = second + .replace(&created.id, &created.revision, replacement("Loser")) + .await + .unwrap_err(); + + assert!(matches!(err, McpServerStoreError::StaleRevision { .. })); + assert_eq!(second.get(&created.id).unwrap(), created); +} + +#[tokio::test] +async fn imports_legacy_directory_once_without_overwriting_sql() { + let (dir, database) = test_database().await; + let store = McpServerStore::load(database.clone_pool()).await.unwrap(); + store.create(draft("existing", "SQLite")).await.unwrap(); + let source = dir.path().join("mcps"); + fs::create_dir_all(&source).await.unwrap(); + let existing_bytes = legacy_toml("Legacy", "https://legacy.example.com/mcp"); + let imported_bytes = legacy_toml("Imported", "https://new.example.com/mcp"); + fs::write(source.join("existing.toml"), &existing_bytes) + .await + .unwrap(); + fs::write(source.join("new.toml"), &imported_bytes) + .await + .unwrap(); + fs::write(source.join("notes.txt"), "preserve in backup") + .await + .unwrap(); + + let report = import_legacy_directory_once(database.pool(), &source) + .await + .unwrap() + .unwrap(); + + assert_eq!(report.imported_rows, 1); + assert_eq!(report.skipped_rows, 1); + assert_eq!(report.mcp_server_ids, vec!["new"]); + assert!(!source.exists()); + assert!(report.backup_path.join("notes.txt").exists()); + let reloaded = McpServerStore::load(database.clone_pool()).await.unwrap(); + assert_eq!( + reloaded + .get(&McpServerId::new("existing").unwrap()) + .unwrap() + .display_name, + "SQLite" + ); + let imported = reloaded.get(&McpServerId::new("new").unwrap()).unwrap(); + assert_eq!(imported.display_name, "Imported"); + assert_eq!( + imported.revision, + McpServerRevision::from_bytes(imported_bytes.as_bytes()) + ); + assert!( + import_legacy_directory_once(database.pool(), &source) + .await + .unwrap() + .is_none() + ); +} + +#[tokio::test] +async fn malformed_legacy_toml_does_not_import_or_rename() { + let (dir, database) = test_database().await; + let source = dir.path().join("mcps"); + fs::create_dir_all(&source).await.unwrap(); + fs::write( + source.join("valid.toml"), + legacy_toml("Valid", "https://example.com/mcp"), + ) + .await + .unwrap(); + fs::write(source.join("broken.toml"), "not valid toml =") + .await + .unwrap(); + + let err = import_legacy_directory_once(database.pool(), &source) + .await + .unwrap_err(); + + assert!(matches!(err, McpServerStoreError::Parse { .. })); + assert!(source.exists()); + assert!( + McpServerStore::load(database.clone_pool()) + .await + .unwrap() + .list() + .is_empty() + ); +} + +#[tokio::test] +async fn legacy_parse_error_chain_does_not_expose_transport_values() { + let (dir, database) = test_database().await; + let source = dir.path().join("mcps"); + fs::create_dir_all(&source).await.unwrap(); + fs::write( + source.join("broken.toml"), + r#" +display_name = "Broken" +startup_timeout_secs = 10 +tool_timeout_secs = 60 + +[transport] +type = "http" +url = "https://example.com/mcp" + +[transport.headers] +Authorization = "do-not-print" trailing-invalid-content +"#, + ) + .await + .unwrap(); + + let err = import_legacy_directory_once(database.pool(), &source) + .await + .unwrap_err(); + let mut rendered = err.to_string(); + let mut source = err.source(); + while let Some(err) = source { + rendered.push_str(&err.to_string()); + source = err.source(); + } + + assert!(!rendered.contains("do-not-print")); +} + +#[tokio::test] +async fn corrupted_stored_row_returns_typed_error() { + let (_dir, database) = test_database().await; + let mut connection = database.pool().acquire().await.unwrap(); + sqlx::query("PRAGMA ignore_check_constraints = ON") + .execute(&mut *connection) + .await + .unwrap(); + sqlx::query( + r" + INSERT INTO mcp_servers ( + id, revision, display_name, transport_type, command_json, env_json, + startup_timeout_secs, tool_timeout_secs + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ", + ) + .bind("broken") + .bind("not-a-revision") + .bind("Broken") + .bind("stdio") + .bind(r#"["server"]"#) + .bind("{}") + .bind(10_i64) + .bind(60_i64) + .execute(&mut *connection) + .await + .unwrap(); + + let err = McpServerStore::load(database.clone_pool()) + .await + .unwrap_err(); + assert!(matches!(err, McpServerStoreError::StoredRevision { .. })); +} + +#[tokio::test] +async fn store_debug_does_not_expose_transport_values() { + let (_dir, database) = test_database().await; + let store = McpServerStore::load(database.clone_pool()).await.unwrap(); + let mut secret_draft = draft("secret", "Secret"); + secret_draft.transport = McpTransport::Http { + protocol: McpHttpProtocol::default(), + url: "https://example.com/mcp".to_string(), + headers: HashMap::from([("Authorization".to_string(), "do-not-print".to_string())]), + }; + store.create(secret_draft).await.unwrap(); + + let debug = format!("{store:?}"); + assert!(!debug.contains("do-not-print")); +} diff --git a/lib/crates/fabro-server/migrations/2026052501_optional_server_env_secrets_to_vault.rs b/lib/crates/fabro-server/migrations/2026052501_optional_server_env_secrets_to_vault.rs index 322ded40d..ab2f03650 100644 --- a/lib/crates/fabro-server/migrations/2026052501_optional_server_env_secrets_to_vault.rs +++ b/lib/crates/fabro-server/migrations/2026052501_optional_server_env_secrets_to_vault.rs @@ -15,8 +15,6 @@ use std::path::{Path, PathBuf}; use anyhow::Context as _; use fabro_config::envfile::{self, EnvFileRemoval}; use fabro_static::{EnvVars, optional_vault_secrets}; -#[cfg(test)] -use fabro_vault::Vault; use fabro_vault::{SecretStore, SecretStoreWrite, SecretType}; pub(crate) const REMOVAL_DEADLINE: &str = "2026-08-18"; @@ -36,94 +34,6 @@ impl OptionalServerEnvSecretsMigrationReport { } } -#[cfg(test)] -pub(crate) fn migrate( - vault: &mut Vault, - server_env_path: &Path, - env_entries: &HashMap, -) -> anyhow::Result { - let server_env_entries = envfile::read_env_file(server_env_path) - .with_context(|| format!("read server env file {}", server_env_path.display()))?; - let mut vault_writes = Vec::new(); - let mut env_removals = Vec::new(); - let mut warnings = Vec::new(); - let mut preserved_env_entries = 0; - - for &name in optional_vault_secrets() { - let process_value = env_entries.get(name); - let file_value = server_env_entries.get(name); - - if let Some(vault_value) = vault.get(name) { - if let Some(file_value) = file_value { - if file_value == vault_value { - env_removals.push(env_removal(name)); - } else { - preserved_env_entries += 1; - warnings.push(format!( - "Preserved {name} in server.env because the vault already contains a different value" - )); - } - } - continue; - } - - match (process_value, file_value) { - (Some(value), Some(file_value)) => { - vault_writes.push((name, value.clone(), secret_type_for(name))); - if value == file_value { - env_removals.push(env_removal(name)); - } else { - preserved_env_entries += 1; - warnings.push(format!( - "Preserved {name} in server.env because process env takes precedence and the file value differs" - )); - } - } - (Some(value), None) => { - vault_writes.push((name, value.clone(), secret_type_for(name))); - } - (None, Some(value)) => { - vault_writes.push((name, value.clone(), secret_type_for(name))); - env_removals.push(env_removal(name)); - } - (None, None) => {} - } - } - - let mut report = OptionalServerEnvSecretsMigrationReport { - migrated_secrets: vault_writes.len(), - removed_env_entries: 0, - preserved_env_entries, - backup_path: None, - warnings, - }; - if vault_writes.is_empty() && env_removals.is_empty() { - return Ok(report); - } - - for (name, value, secret_type) in vault_writes { - vault - .set(name, &value, secret_type, None) - .with_context(|| format!("write migrated secret {name} to vault"))?; - } - - if !env_removals.is_empty() { - let backup_path = backup_server_env_file(server_env_path)?; - let update_report = - envfile::update_env_file_with_report(server_env_path, env_removals, Vec::new()) - .with_context(|| { - format!( - "remove migrated optional secrets from {}", - server_env_path.display() - ) - })?; - report.removed_env_entries = update_report.removed_keys.len(); - report.backup_path = Some(backup_path); - } - - Ok(report) -} - pub(crate) async fn migrate_to_store( store: &SecretStore, server_env_path: &Path, @@ -131,6 +41,7 @@ pub(crate) async fn migrate_to_store( ) -> anyhow::Result { let server_env_entries = envfile::read_env_file(server_env_path) .with_context(|| format!("read server env file {}", server_env_path.display()))?; + let stored = store.snapshot().await?; let mut writes = Vec::new(); let mut env_removals = Vec::new(); let mut warnings = Vec::new(); @@ -139,7 +50,7 @@ pub(crate) async fn migrate_to_store( for &name in optional_vault_secrets() { let process_value = env_entries.get(name); let file_value = server_env_entries.get(name); - if let Some(entry) = store.get(name).await? { + if let Some(entry) = stored.get_entry(name) { if let Some(file_value) = file_value { if file_value == &entry.value { env_removals.push(env_removal(name)); diff --git a/lib/crates/fabro-server/src/migrations.rs b/lib/crates/fabro-server/src/migrations.rs index 986619184..133dc885a 100644 --- a/lib/crates/fabro-server/src/migrations.rs +++ b/lib/crates/fabro-server/src/migrations.rs @@ -2,8 +2,6 @@ use std::collections::HashMap; use std::path::Path; use fabro_vault::SecretStore; -#[cfg(test)] -use fabro_vault::Vault; #[path = "../migrations/2026051801_legacy_vault_entries.rs"] mod legacy_vault_entries; @@ -21,15 +19,6 @@ pub(crate) fn migrate_legacy_vault_file(path: &Path) -> anyhow::Result, -) -> anyhow::Result { - optional_server_env_secrets_to_vault::migrate(vault, server_env_path, env_entries) -} - pub(crate) async fn migrate_optional_server_env_secrets_to_store( store: &SecretStore, server_env_path: &Path, diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index cfba70bd7..7ed6514af 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -2328,21 +2328,21 @@ fn mcp_server_dir_for_active_config(active_config_path: &std::path::Path) -> Pat reason = "synchronous app-state assembly may run inside an async runtime; a short-lived OS \ thread avoids nested Tokio runtimes" )] -fn load_environment_store_blocking( - pool: DbPool, - local_enabled: bool, -) -> anyhow::Result { +fn load_store_blocking(description: &'static str, load: F) -> anyhow::Result +where + T: Send + 'static, + F: FnOnce() -> Fut + Send + 'static, + Fut: std::future::Future>, +{ std::thread::spawn(move || { let runtime = TokioRuntimeBuilder::new_current_thread() .enable_all() .build() - .context("build environment store runtime")?; - runtime - .block_on(EnvironmentStore::load(pool, local_enabled)) - .map_err(anyhow::Error::new) + .with_context(|| format!("build {description} runtime"))?; + runtime.block_on(load()) }) .join() - .expect("environment store load thread should not panic") + .expect("store load thread should not panic") } pub(crate) fn build_app_state(config: AppStateConfig) -> anyhow::Result> { @@ -2382,15 +2382,24 @@ pub(crate) fn build_app_state(config: AppStateConfig) -> anyhow::Result for ApiError { } // Remaining variants are persistence/parse faults that indicate an // internal problem rather than a client one. - McpServerStoreError::InvalidFilename { .. } + McpServerStoreError::Db { .. } + | McpServerStoreError::StoredRevision { .. } + | McpServerStoreError::StoredTransport { .. } + | McpServerStoreError::StoredInteger { .. } + | McpServerStoreError::JsonEncode { .. } + | McpServerStoreError::JsonDecode { .. } + | McpServerStoreError::InvalidFilename { .. } | McpServerStoreError::Parse { .. } | McpServerStoreError::InvalidUtf8 { .. } | McpServerStoreError::Serialize { .. } - | McpServerStoreError::Io { .. } => Self::new( + | McpServerStoreError::Io { .. } + | McpServerStoreError::LegacyBackup { .. } => Self::new( StatusCode::INTERNAL_SERVER_ERROR, "mcp server store operation failed", ), diff --git a/lib/crates/fabro-server/src/server/tests.rs b/lib/crates/fabro-server/src/server/tests.rs index 9a863ed50..f7fed5606 100644 --- a/lib/crates/fabro-server/src/server/tests.rs +++ b/lib/crates/fabro-server/src/server/tests.rs @@ -112,8 +112,43 @@ fn test_environment_store( }) .join() .expect("environment store setup thread should not panic"); - let store = load_environment_store_blocking(pool, local_enabled) - .expect("test environment store should load"); + let store = load_store_blocking("environment store", move || async move { + EnvironmentStore::load(pool, local_enabled) + .await + .map_err(anyhow::Error::new) + }) + .expect("test environment store should load"); + (temp, store) +} + +fn test_mcp_server_store() -> (tempfile::TempDir, McpServerStore) { + let temp = tempfile::tempdir().expect("MCP server store tempdir should be created"); + let db_path = temp.path().join("fabro.sqlite3"); + let pool = std::thread::spawn(move || { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("MCP server store setup runtime should build"); + runtime.block_on(async move { + let database = fabro_db::Database::connect(db_path) + .await + .expect("test MCP server database should connect"); + database + .migrate() + .await + .expect("test MCP server database should migrate"); + database.clone_pool() + }) + }) + .join() + .expect("MCP server store setup thread should not panic"); + let mcps_dir = temp.path().join("mcps"); + let store = load_store_blocking("MCP server store", move || async move { + McpServerStore::open(pool, mcps_dir) + .await + .map_err(anyhow::Error::new) + }) + .expect("test MCP server store should load"); (temp, store) } @@ -1275,10 +1310,9 @@ id = "missing" #[test] fn system_sandbox_provider_uses_manifest_defaults() { - let (temp, environment_store) = + let (_environment_temp, environment_store) = test_environment_store(Some(EnvironmentProvider::Daytona), true); - let mcp_server_store = - McpServerStore::load(temp.path().join("mcps")).expect("mcp server store should load"); + let (_mcp_temp, mcp_server_store) = test_mcp_server_store(); let source = r#" _version = 1 @@ -1296,9 +1330,8 @@ id = "default" #[test] fn system_sandbox_provider_defaults_when_manifest_run_settings_do_not_resolve() { - let (temp, environment_store) = test_environment_store(None, true); - let mcp_server_store = - McpServerStore::load(temp.path().join("mcps")).expect("mcp server store should load"); + let (_environment_temp, environment_store) = test_environment_store(None, true); + let (_mcp_temp, mcp_server_store) = test_mcp_server_store(); let source = r#" _version = 1 diff --git a/lib/crates/fabro-server/src/startup.rs b/lib/crates/fabro-server/src/startup.rs index e0ef58fbc..78fef39cb 100644 --- a/lib/crates/fabro-server/src/startup.rs +++ b/lib/crates/fabro-server/src/startup.rs @@ -1,8 +1,6 @@ use std::collections::HashMap; use std::path::Path; -#[cfg(test)] -use anyhow::Context as _; use fabro_static::EnvVars; use fabro_types::settings::ServerNamespace; use fabro_vault::Vault; @@ -54,54 +52,6 @@ pub fn migrate_startup_vault(vault_path: impl AsRef) { } } -#[cfg(test)] -fn load_startup_vault(vault_path: impl AsRef) -> anyhow::Result { - let vault_path = vault_path.as_ref(); - migrate_startup_vault(vault_path); - Vault::load(vault_path.to_path_buf()) - .with_context(|| format!("load vault {}", vault_path.display())) -} - -#[cfg(test)] -pub(crate) fn prepare_startup_vault( - vault_path: impl AsRef, - server_env_path: impl AsRef, - env_entries: &HashMap, -) -> anyhow::Result { - let mut vault = load_startup_vault(vault_path)?; - let report = migrations::migrate_optional_server_env_secrets_to_vault( - &mut vault, - server_env_path.as_ref(), - env_entries, - ) - .context("migrate optional server env secrets into vault")?; - - for warning in &report.warnings { - warn!( - warning = %warning, - removal_deadline = migrations::OPTIONAL_SERVER_ENV_SECRETS_REMOVAL_DEADLINE, - "Optional server env secrets migration warning" - ); - } - - if report.changed() { - let backup_path = report - .backup_path - .as_ref() - .map_or_else(|| "".to_string(), |path| path.display().to_string()); - warn!( - migrated_secrets = report.migrated_secrets, - removed_env_entries = report.removed_env_entries, - preserved_env_entries = report.preserved_env_entries, - backup_path = %backup_path, - removal_deadline = migrations::OPTIONAL_SERVER_ENV_SECRETS_REMOVAL_DEADLINE, - "Migrated optional server env secrets into vault" - ); - } - - Ok(vault) -} - pub fn validate_startup( env_path: &Path, env_entries: HashMap, @@ -123,9 +73,10 @@ mod tests { use fabro_config::{ServerSettingsBuilder, envfile}; use fabro_static::EnvVars; use fabro_types::settings::ServerNamespace; - use fabro_vault::{SecretType, Vault}; + use fabro_vault::{SecretStore, SecretType, Vault}; - use super::{prepare_startup_vault, validate_startup}; + use super::validate_startup; + use crate::migrations; fn resolved_settings(auth_methods: &[&str]) -> ServerNamespace { ServerSettingsBuilder::from_toml(&format!( @@ -159,8 +110,12 @@ client_id = "Iv1.test" dir.path().join("server.env") } - fn vault_path(dir: &tempfile::TempDir) -> PathBuf { - dir.path().join("secrets.json") + async fn test_secret_store(dir: &tempfile::TempDir) -> SecretStore { + let database = fabro_db::Database::connect(dir.path().join("fabro.db")) + .await + .unwrap(); + database.migrate().await.unwrap(); + SecretStore::new(database.clone_pool()) } #[expect( @@ -280,8 +235,8 @@ client_id = "Iv1.test" .expect("github client secret in vault should satisfy startup"); } - #[test] - fn prepare_startup_vault_migrates_server_env_optional_secrets_to_vault() { + #[tokio::test] + async fn migrate_optional_secrets_moves_server_env_secrets_to_store() { let dir = tempfile::tempdir().unwrap(); let server_env_path = env_path(&dir); envfile::write_env_file( @@ -303,33 +258,36 @@ client_id = "Iv1.test" ]), ) .unwrap(); + let store = test_secret_store(&dir).await; - let vault = prepare_startup_vault(vault_path(&dir), &server_env_path, &HashMap::new()) - .expect("legacy optional secrets should migrate"); + migrations::migrate_optional_server_env_secrets_to_store( + &store, + &server_env_path, + &HashMap::new(), + ) + .await + .expect("legacy optional secrets should migrate"); - assert_eq!( - vault.get(EnvVars::GITHUB_APP_CLIENT_SECRET), - Some("legacy-client-secret") - ); - assert_eq!( - vault - .get_entry(EnvVars::GITHUB_APP_CLIENT_SECRET) - .unwrap() - .secret_type, - SecretType::Token - ); - assert_eq!( - vault.get(EnvVars::GITHUB_APP_PRIVATE_KEY), - Some("legacy-private-key") - ); - assert_eq!( - vault - .get_entry(EnvVars::GITHUB_APP_PRIVATE_KEY) - .unwrap() - .secret_type, - SecretType::File - ); - assert_eq!(vault.get(EnvVars::OPENAI_API_KEY), Some("sk-legacy")); + let client_secret = store + .get(EnvVars::GITHUB_APP_CLIENT_SECRET) + .await + .unwrap() + .expect("client secret should be stored"); + assert_eq!(client_secret.value, "legacy-client-secret"); + assert_eq!(client_secret.secret_type, SecretType::Token); + let private_key = store + .get(EnvVars::GITHUB_APP_PRIVATE_KEY) + .await + .unwrap() + .expect("private key should be stored"); + assert_eq!(private_key.value, "legacy-private-key"); + assert_eq!(private_key.secret_type, SecretType::File); + let openai_key = store + .get(EnvVars::OPENAI_API_KEY) + .await + .unwrap() + .expect("openai key should be stored"); + assert_eq!(openai_key.value, "sk-legacy"); let server_env = envfile::read_env_file(&server_env_path).unwrap(); assert!(server_env.contains_key(EnvVars::SESSION_SECRET)); @@ -339,8 +297,8 @@ client_id = "Iv1.test" assert_eq!(migration_backups(dir.path()).len(), 1); } - #[test] - fn prepare_startup_vault_prefers_process_env_and_preserves_conflicting_server_env() { + #[tokio::test] + async fn migrate_optional_secrets_prefers_process_env_and_preserves_conflicting_server_env() { let dir = tempfile::tempdir().unwrap(); let server_env_path = env_path(&dir); envfile::write_env_file( @@ -355,14 +313,22 @@ client_id = "Iv1.test" EnvVars::GITHUB_APP_CLIENT_SECRET.to_string(), "process-client-secret".to_string(), )]); + let store = test_secret_store(&dir).await; - let vault = prepare_startup_vault(vault_path(&dir), &server_env_path, &env_entries) - .expect("process env secret should migrate"); + migrations::migrate_optional_server_env_secrets_to_store( + &store, + &server_env_path, + &env_entries, + ) + .await + .expect("process env secret should migrate"); - assert_eq!( - vault.get(EnvVars::GITHUB_APP_CLIENT_SECRET), - Some("process-client-secret") - ); + let client_secret = store + .get(EnvVars::GITHUB_APP_CLIENT_SECRET) + .await + .unwrap() + .expect("client secret should be stored"); + assert_eq!(client_secret.value, "process-client-secret"); let server_env = envfile::read_env_file(&server_env_path).unwrap(); assert_eq!( server_env @@ -373,8 +339,9 @@ client_id = "Iv1.test" assert!(migration_backups(dir.path()).is_empty()); } - #[test] - fn prepare_startup_vault_keeps_existing_vault_secret_and_removes_matching_server_env() { + #[tokio::test] + async fn migrate_optional_secrets_keeps_existing_stored_secret_and_removes_matching_server_env() + { let dir = tempfile::tempdir().unwrap(); let server_env_path = env_path(&dir); envfile::write_env_file( @@ -385,30 +352,38 @@ client_id = "Iv1.test" )]), ) .unwrap(); - let mut vault = Vault::load(vault_path(&dir)).unwrap(); - vault + let store = test_secret_store(&dir).await; + store .set( EnvVars::GITHUB_APP_CLIENT_SECRET, "vault-client-secret", SecretType::Token, None, ) + .await .unwrap(); - let vault = prepare_startup_vault(vault_path(&dir), &server_env_path, &HashMap::new()) - .expect("redundant server env secret should be cleaned up"); + migrations::migrate_optional_server_env_secrets_to_store( + &store, + &server_env_path, + &HashMap::new(), + ) + .await + .expect("redundant server env secret should be cleaned up"); - assert_eq!( - vault.get(EnvVars::GITHUB_APP_CLIENT_SECRET), - Some("vault-client-secret") - ); + let client_secret = store + .get(EnvVars::GITHUB_APP_CLIENT_SECRET) + .await + .unwrap() + .expect("client secret should be stored"); + assert_eq!(client_secret.value, "vault-client-secret"); let server_env = envfile::read_env_file(&server_env_path).unwrap(); assert!(!server_env.contains_key(EnvVars::GITHUB_APP_CLIENT_SECRET)); assert_eq!(migration_backups(dir.path()).len(), 1); } - #[test] - fn prepare_startup_vault_migrated_github_client_secret_satisfies_startup() { + #[tokio::test] + async fn migrate_optional_secrets_migrated_github_client_secret_satisfies_startup() { let dir = tempfile::tempdir().unwrap(); let server_env_path = env_path(&dir); envfile::write_env_file( @@ -426,10 +401,17 @@ client_id = "Iv1.test" ) .unwrap(); let settings = resolved_settings(&["github"]); + let store = test_secret_store(&dir).await; - let vault = prepare_startup_vault(vault_path(&dir), &server_env_path, &HashMap::new()) - .expect("legacy github client secret should migrate"); + migrations::migrate_optional_server_env_secrets_to_store( + &store, + &server_env_path, + &HashMap::new(), + ) + .await + .expect("legacy github client secret should migrate"); + let vault = store.snapshot().await.unwrap().into_vault(); validate_startup(&server_env_path, HashMap::new(), &settings, &vault) .expect("migrated github client secret should satisfy startup"); } diff --git a/lib/crates/fabro-server/tests/it/api/mcp_servers.rs b/lib/crates/fabro-server/tests/it/api/mcp_servers.rs index 50e78b2a3..5c71d671e 100644 --- a/lib/crates/fabro-server/tests/it/api/mcp_servers.rs +++ b/lib/crates/fabro-server/tests/it/api/mcp_servers.rs @@ -2,8 +2,12 @@ use std::path::{Path, PathBuf}; use axum::body::Body; use axum::http::{Method, Request, StatusCode, header}; +use fabro_db::Database; +use fabro_mcp_store::McpServerStore; use fabro_server::server::build_router; use fabro_server::test_support::{TestAppStateBuilder, build_test_router, test_auth_mode}; +use fabro_types::settings::McpTransport; +use fabro_types::{McpServerDefinition, McpServerId}; use serde_json::{Value, json}; use tower::ServiceExt; @@ -47,11 +51,13 @@ fn replacement_body(display_name: &str) -> Value { fn mcp_server_app() -> (axum::Router, tempfile::TempDir, PathBuf) { let temp_dir = tempfile::tempdir().expect("mcp server test tempdir should be created"); let active_config_path = temp_dir.path().join("settings.toml"); - let mcp_dir = temp_dir.path().join("mcps"); + let vault_path = temp_dir.path().join("secrets.json"); + let db_path = temp_dir.path().join("db").join("fabro.sqlite3"); let state = TestAppStateBuilder::new() .active_config_path(active_config_path) + .vault_path(vault_path) .build(); - (build_test_router(state), temp_dir, mcp_dir) + (build_test_router(state), temp_dir, db_path) } fn json_request(method: Method, path: &str, body: &Value) -> Request { @@ -114,11 +120,14 @@ fn revision_from(body: &Value) -> &str { .expect("mcp server response should include a revision") } -async fn persisted_mcp_server_toml(mcp_dir: &Path, id: &str) -> toml::Value { - let persisted = tokio::fs::read_to_string(mcp_dir.join(format!("{id}.toml"))) +async fn persisted_mcp_server(db_path: &Path, id: &str) -> Option { + let database = Database::connect(db_path) .await - .expect("persisted mcp server TOML should be readable"); - toml::from_str(&persisted).expect("persisted mcp server TOML should parse") + .expect("persisted MCP server database should connect"); + let store = McpServerStore::load(database.clone_pool()) + .await + .expect("persisted MCP server store should load"); + store.get(&McpServerId::new(id).expect("fixture MCP server id should be valid")) } #[tokio::test] @@ -141,8 +150,8 @@ async fn empty_mcp_server_list_returns_total_zero() { } #[tokio::test] -async fn create_mcp_server_returns_etag_and_persists_sibling_toml_file() { - let (app, _temp_dir, mcp_dir) = mcp_server_app(); +async fn create_mcp_server_returns_etag_and_persists_sqlite_row() { + let (app, _temp_dir, db_path) = mcp_server_app(); let response = app .clone() @@ -167,7 +176,7 @@ async fn create_mcp_server_returns_etag_and_persists_sibling_toml_file() { assert_eq!(body["id"], "sentry"); assert_eq!(body["display_name"], "Sentry"); assert_eq!(etag, format!("\"{}\"", revision_from(&body))); - assert!(mcp_dir.join("sentry.toml").exists()); + assert!(db_path.exists()); // The response is the value-omitting view: header *names* are returned, but // the stored header value is not. @@ -177,27 +186,20 @@ async fn create_mcp_server_returns_etag_and_persists_sibling_toml_file() { "response must not echo transport header values" ); - let persisted = persisted_mcp_server_toml(&mcp_dir, "sentry").await; - assert_eq!( - persisted.get("display_name").and_then(toml::Value::as_str), - Some("Sentry") - ); - assert!(persisted.get("id").is_none()); - assert!(persisted.get("revision").is_none()); - // The value the view omits is still persisted on disk for the runtime to use. - assert_eq!( - persisted - .get("transport") - .and_then(|transport| transport.get("headers")) - .and_then(|headers| headers.get("X-Org")) - .and_then(toml::Value::as_str), - Some("fabro") - ); + let persisted = persisted_mcp_server(&db_path, "sentry") + .await + .expect("MCP server should be persisted"); + assert_eq!(persisted.display_name, "Sentry"); + assert_eq!(persisted.revision.as_str(), revision_from(&body)); + let McpTransport::Http { headers, .. } = persisted.transport else { + panic!("persisted MCP server should use HTTP transport") + }; + assert_eq!(headers.get("X-Org").map(String::as_str), Some("fabro")); } #[tokio::test] -async fn mcp_server_round_trips_through_create_get_and_toml() { - let (app, _temp_dir, mcp_dir) = mcp_server_app(); +async fn mcp_server_round_trips_through_create_get_and_sqlite() { + let (app, _temp_dir, db_path) = mcp_server_app(); let created = create_mcp_server(&app, "sentry", "Sentry").await; assert_eq!(created["transport"]["type"], "http"); @@ -211,14 +213,10 @@ async fn mcp_server_round_trips_through_create_get_and_toml() { let retrieved = response_json(response, StatusCode::OK, "GET /api/v1/mcp-servers/sentry").await; assert_eq!(retrieved, created); - let persisted = persisted_mcp_server_toml(&mcp_dir, "sentry").await; - assert_eq!( - persisted - .get("transport") - .and_then(|transport| transport.get("type")) - .and_then(toml::Value::as_str), - Some("http") - ); + let persisted = persisted_mcp_server(&db_path, "sentry") + .await + .expect("MCP server should be persisted"); + assert!(matches!(persisted.transport, McpTransport::Http { .. })); } #[tokio::test] @@ -432,8 +430,8 @@ async fn replace_and_delete_mcp_server_require_if_match() { } #[tokio::test] -async fn delete_mcp_server_removes_file_and_resource() { - let (app, _temp_dir, mcp_dir) = mcp_server_app(); +async fn delete_mcp_server_removes_sqlite_row_and_resource() { + let (app, _temp_dir, db_path) = mcp_server_app(); let created = create_mcp_server(&app, "sentry", "Sentry").await; let revision = revision_from(&created); @@ -454,7 +452,7 @@ async fn delete_mcp_server_removes_file_and_resource() { ) .await; - assert!(!mcp_dir.join("sentry.toml").exists()); + assert!(persisted_mcp_server(&db_path, "sentry").await.is_none()); let response = app .oneshot(empty_request(Method::GET, "/mcp-servers/sentry")) .await @@ -593,7 +591,7 @@ id = "sentry" } #[tokio::test] -async fn mcp_server_store_malformed_persisted_toml_fails_startup() { +async fn malformed_legacy_mcp_server_toml_fails_startup() { let temp_dir = tempfile::tempdir().expect("mcp server test tempdir should be created"); let mcp_dir = temp_dir.path().join("mcps"); tokio::fs::create_dir_all(&mcp_dir) @@ -610,6 +608,84 @@ async fn mcp_server_store_malformed_persisted_toml_fails_startup() { assert!(result.is_err()); } +#[tokio::test] +async fn legacy_mcp_server_toml_imports_at_startup_and_renames_directory() { + let temp_dir = tempfile::tempdir().expect("mcp server test tempdir should be created"); + let active_config_path = temp_dir.path().join("settings.toml"); + let vault_path = temp_dir.path().join("secrets.json"); + let db_path = temp_dir.path().join("db").join("fabro.sqlite3"); + let mcp_dir = temp_dir.path().join("mcps"); + tokio::fs::create_dir_all(&mcp_dir) + .await + .expect("legacy MCP server directory should be created"); + tokio::fs::write( + mcp_dir.join("sentry.toml"), + r#" +display_name = "Sentry" +startup_timeout_secs = 10 +tool_timeout_secs = 60 + +[transport] +type = "http" +url = "https://example.com/mcp" + +[transport.headers] +Authorization = "secret-value" +"#, + ) + .await + .expect("legacy MCP server fixture should be written"); + + let state = TestAppStateBuilder::new() + .active_config_path(active_config_path) + .vault_path(vault_path) + .build(); + let app = build_test_router(state); + + let response = app + .oneshot(empty_request(Method::GET, "/mcp-servers/sentry")) + .await + .expect("imported MCP server should respond"); + let body = response_json( + response, + StatusCode::OK, + "GET imported /api/v1/mcp-servers/sentry", + ) + .await; + assert_eq!(body["display_name"], "Sentry"); + assert_eq!(body["transport"]["header_keys"], json!(["Authorization"])); + assert!(!mcp_dir.exists()); + let mut entries = tokio::fs::read_dir(temp_dir.path()) + .await + .expect("test directory should be readable"); + let mut backup_found = false; + while let Some(entry) = entries + .next_entry() + .await + .expect("test directory entry should be readable") + { + if entry + .file_name() + .to_string_lossy() + .starts_with("mcps.imported-") + { + backup_found = true; + break; + } + } + assert!(backup_found); + let persisted = persisted_mcp_server(&db_path, "sentry") + .await + .expect("imported MCP server should be in SQLite"); + let McpTransport::Http { headers, .. } = persisted.transport else { + panic!("imported MCP server should use HTTP transport") + }; + assert_eq!( + headers.get("Authorization").map(String::as_str), + Some("secret-value") + ); +} + #[tokio::test] async fn mcp_servers_routes_require_authenticated_user() { let temp_dir = tempfile::tempdir().expect("mcp server test tempdir should be created"); diff --git a/lib/crates/fabro-types/src/mcp_store.rs b/lib/crates/fabro-types/src/mcp_store.rs index 18cedbe42..5a7d64535 100644 --- a/lib/crates/fabro-types/src/mcp_store.rs +++ b/lib/crates/fabro-types/src/mcp_store.rs @@ -3,8 +3,8 @@ //! These types describe MCP server definitions that are stored once on a Fabro //! server and later referenced by name from workflow configs. They are //! persistence-independent: the durable storage lives in the `fabro-mcp-store` -//! crate, which derives `id` (filename stem) and `revision` (content hash) and -//! never persists them inside the TOML body. +//! crate, which persists `id` and a content-hash `revision` alongside the +//! normalized definition fields. //! //! Transport is the existing [`McpTransport`](crate::settings::McpTransport) //! reused verbatim, so a stored definition uses the same `stdio`/`http`/ @@ -28,11 +28,9 @@ use crate::settings::run::McpHttpProtocol; /// A server-managed MCP server definition. /// -/// `id` and `revision` are derived (filename stem + content hash of the -/// persisted TOML bytes) and are not stored in the persisted TOML body. This is -/// the internal/persistence model and carries full transport values; it is not -/// serialized to clients. Read APIs return [`McpServerView`] instead, which -/// omits env/header values. +/// This is the internal/persistence model and carries full transport values; it +/// is not serialized to clients. Read APIs return [`McpServerView`] instead, +/// which omits env/header values. #[derive(Debug, Clone, PartialEq)] pub struct McpServerDefinition { pub id: McpServerId, @@ -44,8 +42,8 @@ pub struct McpServerDefinition { pub tool_timeout_secs: u64, } -/// Fields supplied when creating a new definition. Carries an `id` (the create -/// call assigns the filename) but no `revision` (the store derives it). +/// Fields supplied when creating a new definition. Carries an `id` but no +/// `revision` (the store derives it). #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct McpServerDraft { @@ -236,6 +234,7 @@ fn sorted_keys_ref(map: &HashMap) -> Vec { pub enum McpServerValidationError { InvalidMcpServerId { value: String }, EmptyName, + TimeoutOutOfRange { field: &'static str, value: u64 }, InvalidTransport { reason: String }, } @@ -249,6 +248,12 @@ impl fmt::Display for McpServerValidationError { ) } Self::EmptyName => f.write_str("mcp server display name must not be empty"), + Self::TimeoutOutOfRange { field, value } => { + write!( + f, + "mcp server {field} value {value} exceeds the signed 64-bit range" + ) + } Self::InvalidTransport { reason } => { write!(f, "mcp server transport is invalid: {reason}") } @@ -258,8 +263,7 @@ impl fmt::Display for McpServerValidationError { impl std::error::Error for McpServerValidationError {} -/// An MCP server id: lowercase, matches `^[a-z0-9][a-z0-9-]{0,62}$`, and equals -/// the persisted file's stem. +/// An MCP server id: lowercase and matches `^[a-z0-9][a-z0-9-]{0,62}$`. #[derive(Clone, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)] pub struct McpServerId(String); @@ -312,8 +316,9 @@ impl<'de> Deserialize<'de> for McpServerId { } } -/// A revision: the lowercase SHA-256 hex of a definition's canonical persisted -/// TOML bytes. Used as an ETag for optimistic concurrency. +/// A revision: lowercase SHA-256 hex of a definition's canonical bytes. Used as +/// an ETag for optimistic concurrency. Legacy imports preserve the hash of the +/// original TOML bytes. #[derive(Clone, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)] pub struct McpServerRevision(String); @@ -406,9 +411,17 @@ pub fn validate_mcp_server_fields( if replace.display_name.trim().is_empty() { return Err(McpServerValidationError::EmptyName); } + validate_timeout("startup_timeout_secs", replace.startup_timeout_secs)?; + validate_timeout("tool_timeout_secs", replace.tool_timeout_secs)?; validate_transport(&replace.transport) } +fn validate_timeout(field: &'static str, value: u64) -> Result<(), McpServerValidationError> { + i64::try_from(value) + .map(|_| ()) + .map_err(|_| McpServerValidationError::TimeoutOutOfRange { field, value }) +} + fn validate_transport(transport: &McpTransport) -> Result<(), McpServerValidationError> { match transport { McpTransport::Stdio { command, .. } | McpTransport::Sandbox { command, .. } => { @@ -515,6 +528,19 @@ mod tests { assert!(validate_mcp_server_fields(&replace).is_err()); } + #[test] + fn validation_rejects_timeout_outside_sqlite_integer_range() { + let replace = McpServerReplace { + display_name: "Sentry".to_string(), + description: None, + transport: http_transport(), + startup_timeout_secs: u64::MAX, + tool_timeout_secs: 60, + }; + + assert!(validate_mcp_server_fields(&replace).is_err()); + } + #[test] fn validation_accepts_well_formed_definition() { let replace = McpServerReplace { diff --git a/lib/crates/fabro-types/src/settings/run.rs b/lib/crates/fabro-types/src/settings/run.rs index 51410c44d..a582ddee9 100644 --- a/lib/crates/fabro-types/src/settings/run.rs +++ b/lib/crates/fabro-types/src/settings/run.rs @@ -2180,14 +2180,34 @@ pub enum McpTransport { }, } -#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +#[derive( + Debug, + Clone, + Copy, + Default, + PartialEq, + Eq, + strum::Display, + strum::EnumString, + strum::IntoStaticStr, + Serialize, + Deserialize, +)] #[serde(rename_all = "snake_case")] +#[strum(serialize_all = "snake_case")] pub enum McpHttpProtocol { #[default] StreamableHttp, Sse, } +impl McpHttpProtocol { + #[must_use] + pub fn as_str(self) -> &'static str { + self.into() + } +} + #[expect( clippy::trivially_copy_pass_by_ref, reason = "serde skip_serializing_if helpers receive borrowed field values" diff --git a/lib/crates/fabro-variable/src/lib.rs b/lib/crates/fabro-variable/src/lib.rs index e5437e0f8..1148f62cc 100644 --- a/lib/crates/fabro-variable/src/lib.rs +++ b/lib/crates/fabro-variable/src/lib.rs @@ -1,9 +1,8 @@ use std::collections::HashMap; -use std::ffi::OsString; use std::path::{Path, PathBuf}; use chrono::{DateTime, Utc}; -use fabro_db::DbPool; +use fabro_db::{DbPool, legacy}; use fabro_types::{Variable, is_env_style_name}; use sqlx::Row as _; use sqlx::sqlite::SqliteRow; @@ -313,24 +312,13 @@ fn parse_legacy_entries( } async fn rename_imported_legacy_file(source_path: &Path) -> Result { - let backup_path = legacy_backup_path(source_path, Utc::now()); - fs::rename(source_path, &backup_path) + legacy::rename_to_legacy_backup(source_path, "variables.json") .await - .map_err(|source| Error::LegacyBackup { + .map_err(|err| Error::LegacyBackup { source_path: source_path.to_path_buf(), - backup_path: backup_path.clone(), - source, - })?; - Ok(backup_path) -} - -fn legacy_backup_path(source_path: &Path, imported_at: DateTime) -> PathBuf { - let timestamp = imported_at.format("%Y%m%dT%H%M%S%fZ"); - let mut file_name = source_path - .file_name() - .map_or_else(|| OsString::from("variables.json"), OsString::from); - file_name.push(format!(".imported-{timestamp}.bak")); - source_path.with_file_name(file_name) + backup_path: err.backup_path, + source: err.source, + }) } fn row_count(count: usize) -> Result { @@ -354,11 +342,9 @@ fn variable_from_row(row: &SqliteRow) -> Result { } fn parse_timestamp(name: &str, column: &'static str, value: &str) -> Result, Error> { - DateTime::parse_from_rfc3339(value) - .map(|timestamp| timestamp.with_timezone(&Utc)) - .map_err(|source| Error::Timestamp { - name: name.to_string(), - column, - source, - }) + fabro_db::parse_rfc3339_utc(value).map_err(|source| Error::Timestamp { + name: name.to_string(), + column, + source, + }) } diff --git a/lib/crates/fabro-vault/src/store.rs b/lib/crates/fabro-vault/src/store.rs index d2a237e4e..8bd7df938 100644 --- a/lib/crates/fabro-vault/src/store.rs +++ b/lib/crates/fabro-vault/src/store.rs @@ -1,10 +1,9 @@ use std::collections::HashMap; -use std::ffi::OsString; use std::path::{Path, PathBuf}; use std::str::FromStr as _; use chrono::{DateTime, Utc}; -use fabro_db::{Database, DbPool}; +use fabro_db::{Database, DbPool, legacy}; use fabro_types::{OAuthCredential, SecretMetadata, SecretType}; use sqlx::sqlite::SqliteRow; use sqlx::{Row as _, Sqlite, Transaction}; @@ -177,10 +176,7 @@ impl SecretStore { .bind(name) .fetch_optional(&self.pool) .await?; - row.as_ref() - .map(entry_from_row) - .transpose() - .map(|entry| entry.map(|(_, entry)| entry)) + row.as_ref().map(entry_from_row).transpose() } pub async fn list(&self) -> Result, SecretStoreError> { @@ -288,7 +284,7 @@ impl SecretStore { .await?; if let Some(row) = row { - return entry_from_row(&row).map(|(_, entry)| entry); + return entry_from_row(&row); } match self.get(name).await? { Some(entry) => Err(SecretStoreError::StaleRevision { @@ -309,8 +305,8 @@ impl SecretStore { .await?; let entries = rows .iter() - .map(entry_from_row) - .collect::, _>>()?; + .map(|row| Ok((row.try_get::("name")?, entry_from_row(row)?))) + .collect::, SecretStoreError>>()?; Ok(SecretSnapshot(Vault::from_entries(entries))) } } @@ -444,7 +440,7 @@ async fn insert_legacy_entry( Ok(result.rows_affected() == 1) } -fn entry_from_row(row: &SqliteRow) -> Result<(String, SecretEntry), SecretStoreError> { +fn entry_from_row(row: &SqliteRow) -> Result { let metadata = metadata_from_row(row)?; let name = metadata.name; let value = row.try_get::("value")?; @@ -458,15 +454,14 @@ fn entry_from_row(row: &SqliteRow) -> Result<(String, SecretEntry), SecretStoreE if revision <= 0 { return Err(SecretStoreError::StoredRevision { name, revision }); } - let entry = SecretEntry { + Ok(SecretEntry { value, secret_type: metadata.secret_type, description: metadata.description, created_at: metadata.created_at, updated_at: metadata.updated_at, revision, - }; - Ok((name, entry)) + }) } fn metadata_from_row(row: &SqliteRow) -> Result { @@ -500,13 +495,11 @@ fn parse_timestamp( column: &'static str, value: &str, ) -> Result, SecretStoreError> { - DateTime::parse_from_rfc3339(value) - .map(|timestamp| timestamp.with_timezone(&Utc)) - .map_err(|source| SecretStoreError::Timestamp { - name: name.to_string(), - column, - source, - }) + fabro_db::parse_rfc3339_utc(value).map_err(|source| SecretStoreError::Timestamp { + name: name.to_string(), + column, + source, + }) } fn validate_oauth_json(value: &str) -> Result<(), serde_json::Error> { @@ -536,22 +529,11 @@ fn validate_stored_name(name: &str, secret_type: SecretType) -> Result<(), Secre } async fn rename_imported_legacy_file(source_path: &Path) -> Result { - let backup_path = legacy_backup_path(source_path, Utc::now()); - fs::rename(source_path, &backup_path) + legacy::rename_to_legacy_backup(source_path, "secrets.json") .await - .map_err(|source| SecretStoreError::LegacyBackup { + .map_err(|err| SecretStoreError::LegacyBackup { source_path: source_path.to_path_buf(), - backup_path: backup_path.clone(), - source, - })?; - Ok(backup_path) -} - -fn legacy_backup_path(source_path: &Path, imported_at: DateTime) -> PathBuf { - let timestamp = imported_at.format("%Y%m%dT%H%M%S%fZ"); - let mut file_name = source_path - .file_name() - .map_or_else(|| OsString::from("secrets.json"), OsString::from); - file_name.push(format!(".imported-{timestamp}.bak")); - source_path.with_file_name(file_name) + backup_path: err.backup_path, + source: err.source, + }) }