scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-10-10 03:30:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

Merge pull request #571 from fabro-sh/move-secrets-mcp-servers-sqlite

Browse source 
Move secrets and MCP server storage to SQLite
This commit is contained in:
Scott Werner 2026-07-22 13:29:26 -04:00 • committed by GitHub
commit 7792883440
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
26 changed files with 1811 additions and 730 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
Cargo.lock generated
View file

@ -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]]

70
docs/plans/2026-07-11-mcp-servers-sqlite-plan.md Normal file
View file

@ -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-<timestamp>.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.

15
docs/public/agents/mcp.mdx
View file

@ -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

1
lib/crates/fabro-db/Cargo.toml
View file

@ -14,6 +14,7 @@ workspace = true
[dependencies]
anyhow.workspace = true
chrono.workspace = true
sqlx.workspace = true
tokio.workspace = true
tracing.workspace = true

73
lib/crates/fabro-db/migrations/2026071102_mcp_servers.sql Normal file
View file

@ -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
)
)
);

79
lib/crates/fabro-db/src/legacy.rs Normal file
View file

@ -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
/// `<name>.imported-<timestamp>.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<PathBuf, LegacyBackupError> {
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<Utc>) -> 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-"))
);
}
}

8
lib/crates/fabro-db/src/lib.rs
View file

@ -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<DateTime<Utc>, chrono::ParseError> {
DateTime::parse_from_rfc3339(value).map(|timestamp| timestamp.with_timezone(&Utc))
}
static MIGRATOR: Migrator = sqlx::migrate!("./migrations");
#[derive(Clone)]

157
lib/crates/fabro-db/tests/sqlite.rs
View file

@ -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<i64>,
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()?;

27
lib/crates/fabro-environment/src/store.rs
View file

@ -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<PathBuf, EnvironmentStoreError> {
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<Utc>) -> 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<EnvironmentId, EnvironmentStoreError> {
@ -786,12 +773,6 @@ fn id_from_path(path: &Path) -> Result<EnvironmentId, EnvironmentStoreError> {
})
}
fn is_toml_file(path: &Path) -> bool {
path.extension()
.and_then(|extension| extension.to_str())
.is_some_and(|extension| extension == "toml")
}
fn encode_json<T: serde::Serialize>(
field: &'static str,
value: &T,

6
lib/crates/fabro-mcp-store/Cargo.toml
View file

@ -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"

48
lib/crates/fabro-mcp-store/src/error.rs
View file

@ -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",
}
}
}

11
lib/crates/fabro-mcp-store/src/lib.rs
View file

@ -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};

34
lib/crates/fabro-mcp-store/src/model.rs
View file

@ -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<PersistedMcpServer> 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<McpServerDefinition, McpServerStoreError> {
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<Vec<u8>, Mcp
fn parse_persisted(bytes: &[u8], path: PathBuf) -> Result<PersistedMcpServer, McpServerStoreError> {
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)
})
}

923
lib/crates/fabro-mcp-store/src/store.rs
View file

File diff suppressed because it is too large Load diff

375
lib/crates/fabro-mcp-store/tests/store.rs Normal file
View file

@ -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"));
}

93
lib/crates/fabro-server/migrations/2026052501_optional_server_env_secrets_to_vault.rs
View file

@ -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<String, String>,
) -> anyhow::Result<OptionalServerEnvSecretsMigrationReport> {
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<OptionalServerEnvSecretsMigrationReport> {
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));

11
lib/crates/fabro-server/src/migrations.rs
View file

@ -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<LegacyVau
legacy_vault_entries::migrate_legacy_vault_file(path)
}
#[cfg(test)]
pub(crate) fn migrate_optional_server_env_secrets_to_vault(
vault: &mut Vault,
server_env_path: &Path,
env_entries: &HashMap<String, String>,
) -> anyhow::Result<OptionalServerEnvSecretsMigrationReport> {
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,

37
lib/crates/fabro-server/src/server.rs
View file

@ -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<EnvironmentStore> {
fn load_store_blocking<T, F, Fut>(description: &'static str, load: F) -> anyhow::Result<T>
where
T: Send + 'static,
F: FnOnce() -> Fut + Send + 'static,
Fut: std::future::Future<Output = anyhow::Result<T>>,
{
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<Arc<AppState>> {
@ -2382,15 +2382,24 @@ pub(crate) fn build_app_state(config: AppStateConfig) -> anyhow::Result<Arc<AppS
.providers
.local
.enabled;
let environment_pool = db_pool.clone();
let environment_store = Arc::new(
load_environment_store_blocking(db_pool.clone(), local_provider_enabled)
.context("load environments")?,
load_store_blocking("environment store", move || async move {
EnvironmentStore::load(environment_pool, local_provider_enabled)
.await
.map_err(anyhow::Error::new)
})
.context("load environments")?,
);
let mcp_server_dir = mcp_server_dir_for_active_config(&active_config_path);
let mcp_server_pool = db_pool.clone();
let mcp_server_store = Arc::new(
McpServerStore::load(mcp_server_dir)
.map_err(anyhow::Error::new)
.context("load mcp servers")?,
load_store_blocking("MCP server store", move || async move {
McpServerStore::open(mcp_server_pool, mcp_server_dir)
.await
.map_err(anyhow::Error::new)
})
.context("load mcp servers")?,
);
let variables = Arc::new(VariableStore::new(db_pool.clone()));
let secret_store = Arc::new(SecretStore::new(db_pool));

11
lib/crates/fabro-server/src/server/handler/mcp_servers.rs
View file

@ -143,11 +143,18 @@ impl From<McpServerStoreError> 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",
),

49
lib/crates/fabro-server/src/server/tests.rs
View file

@ -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

188
lib/crates/fabro-server/src/startup.rs
View file

@ -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<Path>) {
}
}
#[cfg(test)]
fn load_startup_vault(vault_path: impl AsRef<Path>) -> anyhow::Result<Vault> {
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<Path>,
server_env_path: impl AsRef<Path>,
env_entries: &HashMap<String, String>,
) -> anyhow::Result<Vault> {
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(|| "<none>".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<String, String>,
@ -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");
}

154
lib/crates/fabro-server/tests/it/api/mcp_servers.rs
View file

@ -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<Body> {
@ -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<McpServerDefinition> {
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");

52
lib/crates/fabro-types/src/mcp_store.rs
View file

@ -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<String, String>) -> Vec<String> {
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 {

22
lib/crates/fabro-types/src/settings/run.rs
View file

@ -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"

36
lib/crates/fabro-variable/src/lib.rs
View file

@ -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<PathBuf, Error> {
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<Utc>) -> 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<i64, Error> {
@ -354,11 +342,9 @@ fn variable_from_row(row: &SqliteRow) -> Result<Variable, Error> {
}
fn parse_timestamp(name: &str, column: &'static str, value: &str) -> Result<DateTime<Utc>, 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,
})
}

54
lib/crates/fabro-vault/src/store.rs
View file

@ -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<Vec<SecretMetadata>, 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::<Result<HashMap<_, _>, _>>()?;
.map(|row| Ok((row.try_get::<String, _>("name")?, entry_from_row(row)?)))
.collect::<Result<HashMap<_, _>, 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<SecretEntry, SecretStoreError> {
let metadata = metadata_from_row(row)?;
let name = metadata.name;
let value = row.try_get::<String, _>("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<SecretMetadata, SecretStoreError> {
@ -500,13 +495,11 @@ fn parse_timestamp(
column: &'static str,
value: &str,
) -> Result<DateTime<Utc>, 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<PathBuf, SecretStoreError> {
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<Utc>) -> 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,
})
}