diff --git a/Cargo.lock b/Cargo.lock index c796c7439..78fef39a1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -140,9 +140,12 @@ dependencies = [ "agent-tunnel", "anyhow", "async-trait", + "camino", + "gateway-db", "libsql", "tempfile", "tokio 1.52.3", + "tracing", "uuid", ] @@ -1853,6 +1856,7 @@ dependencies = [ "etherparse", "expect-test", "futures", + "gateway-db", "hex", "hostname 0.4.2", "http-body-util", @@ -2666,6 +2670,18 @@ dependencies = [ "slab", ] +[[package]] +name = "gateway-db" +version = "0.0.0" +dependencies = [ + "anyhow", + "camino", + "libsql", + "tempfile", + "tokio 1.52.3", + "windows-sys 0.61.2", +] + [[package]] name = "generate-openapi" version = "0.0.0" @@ -7656,6 +7672,7 @@ dependencies = [ "expect-test", "fastrand", "futures-util", + "gateway-db", "ipnetwork", "jmux-proto", "libsql", diff --git a/crates/agent-tunnel-libsql/Cargo.toml b/crates/agent-tunnel-libsql/Cargo.toml index 627a04e64..a3612b6ad 100644 --- a/crates/agent-tunnel-libsql/Cargo.toml +++ b/crates/agent-tunnel-libsql/Cargo.toml @@ -12,8 +12,11 @@ workspace = true agent-tunnel.path = "../agent-tunnel" anyhow = "1" async-trait = "0.1" +camino = "1.1" +gateway-db.path = "../gateway-db" libsql = { version = "0.9", default-features = false, features = ["core", "serde"] } tokio = { version = "1.52", features = ["sync"] } +tracing = "0.1" uuid = "1.23" [dev-dependencies] diff --git a/crates/agent-tunnel-libsql/migrations/01_agent_authorization.sql b/crates/agent-tunnel-libsql/migrations/01_agent_tunnel.sql similarity index 79% rename from crates/agent-tunnel-libsql/migrations/01_agent_authorization.sql rename to crates/agent-tunnel-libsql/migrations/01_agent_tunnel.sql index eaa5fb462..bd3159927 100644 --- a/crates/agent-tunnel-libsql/migrations/01_agent_authorization.sql +++ b/crates/agent-tunnel-libsql/migrations/01_agent_tunnel.sql @@ -1,9 +1,9 @@ -CREATE TABLE metadata ( +CREATE TABLE agent_tunnel_metadata ( key TEXT PRIMARY KEY, value BLOB NOT NULL ); -CREATE TABLE accepted_agents ( +CREATE TABLE agent_tunnel_accepted_agents ( agent_id TEXT PRIMARY KEY, name TEXT NOT NULL COLLATE NOCASE UNIQUE, client_spki_sha256 BLOB NOT NULL, @@ -13,7 +13,7 @@ CREATE TABLE accepted_agents ( CHECK (length(client_spki_sha256) = 32) ); -CREATE TABLE enrollment_attempts ( +CREATE TABLE agent_tunnel_enrollment_attempts ( jti TEXT PRIMARY KEY, agent_id TEXT NOT NULL, request_sha256 BLOB NOT NULL, @@ -23,7 +23,7 @@ CREATE TABLE enrollment_attempts ( CHECK (deleted IN (0, 1)) ); -CREATE TABLE deleted_agent_keys ( +CREATE TABLE agent_tunnel_deleted_agent_keys ( client_spki_sha256 BLOB PRIMARY KEY, agent_id TEXT NOT NULL, CHECK (length(client_spki_sha256) = 32) diff --git a/crates/agent-tunnel-libsql/src/lib.rs b/crates/agent-tunnel-libsql/src/lib.rs index e23d9d3e7..56ea02620 100644 --- a/crates/agent-tunnel-libsql/src/lib.rs +++ b/crates/agent-tunnel-libsql/src/lib.rs @@ -1,3 +1,14 @@ +//! The agent tunnel's tables in `gateway.db`: accepted Agents, enrollment attempts and the CA they belong to. +//! +//! The tables are named `agent_tunnel_*`, and this crate alone creates and migrates them. Their schema version is the +//! `schema_version` row of `agent_tunnel_metadata`, not the file's `PRAGMA user_version`, because other features keep +//! their own tables in the same file. + +#[macro_use] +extern crate tracing; + +mod upgrade; + use agent_tunnel::authorization::{ AcceptedAgent, AgentAuthorizationStore, EnrollmentAttempt, EnrollmentConflict, EnrollmentOutcome, SpkiSha256, }; @@ -7,119 +18,49 @@ use libsql::{Connection, TransactionBehavior, params}; use tokio::sync::Mutex; use uuid::Uuid; -const MIGRATIONS: &[&str] = &[include_str!("../migrations/01_agent_authorization.sql")]; +pub use self::upgrade::import_2026_3_0_database; + +/// Every schema change ever made to the `agent_tunnel_*` tables, oldest first. Only ever append to it. +const MIGRATIONS: &[&str] = &[include_str!("../migrations/01_agent_tunnel.sql")]; + const CA_SPKI_SHA256_KEY: &str = "ca_spki_sha256"; +const SCHEMA_VERSION_KEY: &str = "schema_version"; +/// Agent authorization kept in the `agent_tunnel_*` tables of `gateway.db`. pub struct LibSqlAgentAuthorizationStore { conn: Mutex, } impl LibSqlAgentAuthorizationStore { - pub async fn open(path: &str, ca_spki_sha256: SpkiSha256) -> anyhow::Result { - let database = libsql::Builder::new_local(path) - .build() - .await - .context("build Agent authorization database")?; - let conn = database.connect().context("open Agent authorization database")?; + /// Opens the store on a `gateway.db` connection of its own, from `GatewayDb::connect`. + /// + /// Opening creates the `agent_tunnel_*` tables or brings them up to date, then checks them and fails if they + /// don't match. The stored Agents only make sense with the CA that signed their certificates, so the first open + /// ties the store to `ca_spki_sha256`, and later opens refuse any other CA. + pub async fn open(conn: Connection, ca_spki_sha256: SpkiSha256) -> anyhow::Result { + migrate(&conn).await?; + let store = Self { conn: Mutex::new(conn) }; - store.apply_pragmas().await?; - store.migrate_and_bind_ca(ca_spki_sha256).await?; store.validate_integrity_and_schema().await?; + store.bind_ca(ca_spki_sha256).await?; Ok(store) } - async fn apply_pragmas(&self) -> anyhow::Result<()> { - const PRAGMAS: &str = " - PRAGMA journal_mode = WAL; - PRAGMA synchronous = FULL; - PRAGMA busy_timeout = 15000; - PRAGMA foreign_keys = ON; - PRAGMA temp_store = MEMORY; - "; - - self.conn - .lock() - .await - .execute_batch(PRAGMAS) - .await - .context("apply Agent authorization database PRAGMAs")?; - Ok(()) - } - - async fn migrate_and_bind_ca(&self, ca_spki_sha256: SpkiSha256) -> anyhow::Result<()> { - let conn = self.conn.lock().await; - let user_version = Self::schema_version(&conn).await?; - - if MIGRATIONS.len() < user_version { - bail!( - "Agent authorization schema version {user_version} is newer than supported version {}", - MIGRATIONS.len() - ); - } - - if user_version == 0 { - let tx = conn - .transaction_with_behavior(TransactionBehavior::Immediate) - .await - .context("begin Agent authorization initialization")?; - for (migration_id, migration) in MIGRATIONS.iter().enumerate() { - tx.execute_batch(migration) - .await - .with_context(|| format!("apply Agent authorization migration {}", migration_id + 1))?; - if migration_id == 0 { - tx.execute( - "INSERT INTO metadata (key, value) VALUES (?1, ?2)", - params![CA_SPKI_SHA256_KEY, ca_spki_sha256.to_vec()], - ) - .await - .context("bind Agent authorization database to CA")?; - } - tx.execute_batch(&format!("PRAGMA user_version = {}", migration_id + 1)) - .await - .with_context(|| format!("record Agent authorization migration {}", migration_id + 1))?; - } - tx.commit().await.context("commit Agent authorization initialization")?; - return Ok(()); - } - - Self::validate_ca_binding(&conn, ca_spki_sha256).await?; - for (migration_id, migration) in MIGRATIONS.iter().enumerate().skip(user_version) { - let tx = conn - .transaction_with_behavior(TransactionBehavior::Immediate) - .await - .with_context(|| format!("begin Agent authorization migration {}", migration_id + 1))?; - tx.execute_batch(migration) - .await - .with_context(|| format!("apply Agent authorization migration {}", migration_id + 1))?; - tx.execute_batch(&format!("PRAGMA user_version = {}", migration_id + 1)) - .await - .with_context(|| format!("record Agent authorization migration {}", migration_id + 1))?; - tx.commit() - .await - .with_context(|| format!("commit Agent authorization migration {}", migration_id + 1))?; + /// Returns the CA the stored Agents belong to, or `None` if the agent tunnel never ran. + /// + /// Once Agents are tied to a CA, Gateway must load that CA and never generate a new one. + pub async fn bound_ca(conn: &Connection) -> anyhow::Result> { + if !metadata_table_exists(conn).await? { + return Ok(None); } - Ok(()) - } - - async fn schema_version(conn: &Connection) -> anyhow::Result { - let row = conn - .query("PRAGMA user_version", ()) - .await - .context("query Agent authorization schema version")? - .next() - .await - .context("read Agent authorization schema version")? - .context("Agent authorization schema version query returned no row")?; - let user_version = row.get::(0).context("decode Agent authorization schema version")?; - usize::try_from(user_version).context("Agent authorization schema version is too large") - } - - async fn validate_ca_binding(conn: &Connection, ca_spki_sha256: SpkiSha256) -> anyhow::Result<()> { let stored = conn - .query("SELECT value FROM metadata WHERE key = ?1", params![CA_SPKI_SHA256_KEY]) + .query( + "SELECT value FROM agent_tunnel_metadata WHERE key = ?1", + params![CA_SPKI_SHA256_KEY], + ) .await .context("query Agent authorization CA identity")? .next() @@ -129,11 +70,53 @@ impl LibSqlAgentAuthorizationStore { .transpose() .context("decode Agent authorization CA identity")?; - let stored = stored.context("Agent authorization database is missing CA metadata")?; - if stored.as_slice() != ca_spki_sha256 { - bail!("Agent authorization database belongs to a different CA"); + stored + .map(|stored| { + SpkiSha256::try_from(stored) + .map_err(|_| anyhow::anyhow!("Agent authorization CA identity has an invalid length")) + }) + .transpose() + } + + async fn bind_ca(&self, ca_spki_sha256: SpkiSha256) -> anyhow::Result<()> { + let conn = self.conn.lock().await; + let tx = conn + .transaction_with_behavior(TransactionBehavior::Immediate) + .await + .context("begin Agent authorization CA binding")?; + + match Self::bound_ca(&tx).await? { + Some(stored) if stored == ca_spki_sha256 => return Ok(()), + Some(_) => bail!("Agent authorization database belongs to a different CA"), + None => {} } + let has_agent_data = tx + .query( + "SELECT 1 FROM agent_tunnel_accepted_agents + UNION ALL SELECT 1 FROM agent_tunnel_deleted_agent_keys + UNION ALL SELECT 1 FROM agent_tunnel_enrollment_attempts + LIMIT 1", + (), + ) + .await + .context("query Agent data before binding a CA")? + .next() + .await + .context("read Agent data before binding a CA")? + .is_some(); + if has_agent_data { + bail!("Agent authorization database has Agent data but is missing CA metadata"); + } + + tx.execute( + "INSERT INTO agent_tunnel_metadata (key, value) VALUES (?1, ?2)", + params![CA_SPKI_SHA256_KEY, ca_spki_sha256.to_vec()], + ) + .await + .context("bind Agent authorization database to CA")?; + tx.commit().await.context("commit Agent authorization CA binding")?; + Ok(()) } @@ -162,18 +145,21 @@ impl LibSqlAgentAuthorizationStore { } const TABLE_PROBES: &[(&str, &str)] = &[ - ("metadata", "SELECT key, value FROM metadata LIMIT 0"), ( - "accepted_agents", - "SELECT agent_id, name, client_spki_sha256, enrollment_jti FROM accepted_agents LIMIT 0", + "agent_tunnel_metadata", + "SELECT key, value FROM agent_tunnel_metadata LIMIT 0", ), ( - "enrollment_attempts", - "SELECT jti, agent_id, request_sha256, expires_at, deleted FROM enrollment_attempts LIMIT 0", + "agent_tunnel_accepted_agents", + "SELECT agent_id, name, client_spki_sha256, enrollment_jti FROM agent_tunnel_accepted_agents LIMIT 0", ), ( - "deleted_agent_keys", - "SELECT agent_id, client_spki_sha256 FROM deleted_agent_keys LIMIT 0", + "agent_tunnel_enrollment_attempts", + "SELECT jti, agent_id, request_sha256, expires_at, deleted FROM agent_tunnel_enrollment_attempts LIMIT 0", + ), + ( + "agent_tunnel_deleted_agent_keys", + "SELECT agent_id, client_spki_sha256 FROM agent_tunnel_deleted_agent_keys LIMIT 0", ), ]; for (table, probe) in TABLE_PROBES { @@ -182,12 +168,12 @@ impl LibSqlAgentAuthorizationStore { .with_context(|| format!("validate Agent authorization table {table}"))?; } - Self::validate_unique_index(&conn, "metadata", &["key"]).await?; - Self::validate_unique_index(&conn, "accepted_agents", &["agent_id"]).await?; - Self::validate_unique_index(&conn, "accepted_agents", &["name"]).await?; - Self::validate_unique_index(&conn, "accepted_agents", &["enrollment_jti"]).await?; - Self::validate_unique_index(&conn, "enrollment_attempts", &["jti"]).await?; - Self::validate_unique_index(&conn, "deleted_agent_keys", &["client_spki_sha256"]).await?; + Self::validate_unique_index(&conn, "agent_tunnel_metadata", &["key"]).await?; + Self::validate_unique_index(&conn, "agent_tunnel_accepted_agents", &["agent_id"]).await?; + Self::validate_unique_index(&conn, "agent_tunnel_accepted_agents", &["name"]).await?; + Self::validate_unique_index(&conn, "agent_tunnel_accepted_agents", &["enrollment_jti"]).await?; + Self::validate_unique_index(&conn, "agent_tunnel_enrollment_attempts", &["jti"]).await?; + Self::validate_unique_index(&conn, "agent_tunnel_deleted_agent_keys", &["client_spki_sha256"]).await?; Ok(()) } @@ -268,6 +254,90 @@ impl LibSqlAgentAuthorizationStore { } } +/// Creates the `agent_tunnel_*` tables or brings them up to date. +async fn migrate(conn: &Connection) -> anyhow::Result<()> { + loop { + // Another connection may migrate at the same time, so the version is read under the write lock. + let tx = conn + .transaction_with_behavior(TransactionBehavior::Immediate) + .await + .context("begin agent tunnel migration")?; + let version = schema_version(&tx).await?; + + let Some(migration) = MIGRATIONS.get(version) else { + tx.rollback().await.context("end agent tunnel migration")?; + + if MIGRATIONS.len() < version { + bail!( + "agent tunnel schema version {version} is newer than supported version {}", + MIGRATIONS.len() + ); + } + + return Ok(()); + }; + + let version = version + 1; + tx.execute_batch(migration) + .await + .with_context(|| format!("apply agent tunnel migration {version}"))?; + record_schema_version(&tx, i64::try_from(version)?).await?; + tx.commit() + .await + .with_context(|| format!("commit agent tunnel migration {version}"))?; + } +} + +/// Records the version of the `agent_tunnel_*` tables, inside the transaction that changes them. +async fn record_schema_version(tx: &Connection, version: i64) -> anyhow::Result<()> { + tx.execute( + "INSERT OR REPLACE INTO agent_tunnel_metadata (key, value) VALUES (?1, ?2)", + params![SCHEMA_VERSION_KEY, version], + ) + .await + .with_context(|| format!("record agent tunnel schema version {version}"))?; + + Ok(()) +} + +/// Reads the version of the `agent_tunnel_*` tables; 0 when there are none yet. +async fn schema_version(conn: &Connection) -> anyhow::Result { + if !metadata_table_exists(conn).await? { + return Ok(0); + } + + let version = conn + .query( + "SELECT value FROM agent_tunnel_metadata WHERE key = ?1", + params![SCHEMA_VERSION_KEY], + ) + .await + .context("query agent tunnel schema version")? + .next() + .await + .context("read agent tunnel schema version")? + .context("agent tunnel tables have no schema version")? + .get::(0) + .context("decode agent tunnel schema version")?; + + usize::try_from(version).context("agent tunnel schema version is out of range") +} + +async fn metadata_table_exists(conn: &Connection) -> anyhow::Result { + let row = conn + .query( + "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'agent_tunnel_metadata'", + (), + ) + .await + .context("look for the agent tunnel tables")? + .next() + .await + .context("read the agent tunnel tables")?; + + Ok(row.is_some()) +} + #[async_trait] impl AgentAuthorizationStore for LibSqlAgentAuthorizationStore { async fn enroll(&self, attempt: EnrollmentAttempt) -> anyhow::Result { @@ -278,7 +348,7 @@ impl AgentAuthorizationStore for LibSqlAgentAuthorizationStore { }; let conn = self.conn.lock().await; conn.execute( - "DELETE FROM enrollment_attempts + "DELETE FROM agent_tunnel_enrollment_attempts WHERE expires_at < unixepoch() - 86400", (), ) @@ -288,7 +358,7 @@ impl AgentAuthorizationStore for LibSqlAgentAuthorizationStore { let existing_attempt = conn .query( "SELECT agent_id, request_sha256, deleted - FROM enrollment_attempts + FROM agent_tunnel_enrollment_attempts WHERE jti = ?1", params![attempt.token_id.to_string()], ) @@ -315,7 +385,7 @@ impl AgentAuthorizationStore for LibSqlAgentAuthorizationStore { let existing_agent = conn .query( "SELECT agent_id, name, client_spki_sha256 - FROM accepted_agents + FROM agent_tunnel_accepted_agents WHERE enrollment_jti = ?1", params![attempt.token_id.to_string()], ) @@ -336,7 +406,7 @@ impl AgentAuthorizationStore for LibSqlAgentAuthorizationStore { let agent_id_exists = conn .query( - "SELECT 1 FROM accepted_agents WHERE agent_id = ?1", + "SELECT 1 FROM agent_tunnel_accepted_agents WHERE agent_id = ?1", params![attempt.agent_id.to_string()], ) .await @@ -351,7 +421,7 @@ impl AgentAuthorizationStore for LibSqlAgentAuthorizationStore { let agent_name_exists = conn .query( - "SELECT 1 FROM accepted_agents WHERE name = ?1", + "SELECT 1 FROM agent_tunnel_accepted_agents WHERE name = ?1", params![attempt.name.clone()], ) .await @@ -367,7 +437,7 @@ impl AgentAuthorizationStore for LibSqlAgentAuthorizationStore { let deleted_key_exists = conn .query( "SELECT 1 - FROM deleted_agent_keys + FROM agent_tunnel_deleted_agent_keys WHERE client_spki_sha256 = ?1", params![attempt.client_spki_sha256.to_vec()], ) @@ -387,7 +457,7 @@ impl AgentAuthorizationStore for LibSqlAgentAuthorizationStore { .context("begin Agent enrollment transaction")?; tx.execute( - "INSERT INTO enrollment_attempts (jti, agent_id, request_sha256, expires_at) + "INSERT INTO agent_tunnel_enrollment_attempts (jti, agent_id, request_sha256, expires_at) VALUES (?1, ?2, ?3, ?4)", params![ attempt.token_id.to_string(), @@ -399,7 +469,7 @@ impl AgentAuthorizationStore for LibSqlAgentAuthorizationStore { .await .context("persist Agent enrollment attempt")?; tx.execute( - "INSERT INTO accepted_agents (agent_id, name, client_spki_sha256, enrollment_jti) + "INSERT INTO agent_tunnel_accepted_agents (agent_id, name, client_spki_sha256, enrollment_jti) VALUES (?1, ?2, ?3, ?4)", params![ attempt.agent_id.to_string(), @@ -420,7 +490,7 @@ impl AgentAuthorizationStore for LibSqlAgentAuthorizationStore { let row = conn .query( "SELECT agent_id, name, client_spki_sha256 - FROM accepted_agents + FROM agent_tunnel_accepted_agents WHERE agent_id = ?1 AND client_spki_sha256 = ?2", params![agent_id.to_string(), client_spki_sha256.to_vec()], ) @@ -438,7 +508,7 @@ impl AgentAuthorizationStore for LibSqlAgentAuthorizationStore { let row = conn .query( "SELECT agent_id, name, client_spki_sha256 - FROM accepted_agents + FROM agent_tunnel_accepted_agents WHERE agent_id = ?1", params![agent_id.to_string()], ) @@ -456,7 +526,7 @@ impl AgentAuthorizationStore for LibSqlAgentAuthorizationStore { let mut rows = conn .query( "SELECT agent_id, name, client_spki_sha256 - FROM accepted_agents + FROM agent_tunnel_accepted_agents ORDER BY name COLLATE NOCASE, agent_id", (), ) @@ -480,7 +550,7 @@ impl AgentAuthorizationStore for LibSqlAgentAuthorizationStore { let row = tx .query( "SELECT agent_id, name, client_spki_sha256, enrollment_jti - FROM accepted_agents + FROM agent_tunnel_accepted_agents WHERE agent_id = ?1", params![agent_id.to_string()], ) @@ -498,20 +568,20 @@ impl AgentAuthorizationStore for LibSqlAgentAuthorizationStore { let enrollment_jti = row.get::(3).context("decode deleted Agent enrollment JTI")?; tx.execute( - "INSERT OR IGNORE INTO deleted_agent_keys (client_spki_sha256, agent_id) + "INSERT OR IGNORE INTO agent_tunnel_deleted_agent_keys (client_spki_sha256, agent_id) VALUES (?1, ?2)", params![accepted.client_spki_sha256.to_vec(), agent_id.to_string()], ) .await .context("remember deleted Agent key")?; tx.execute( - "DELETE FROM accepted_agents WHERE agent_id = ?1", + "DELETE FROM agent_tunnel_accepted_agents WHERE agent_id = ?1", params![agent_id.to_string()], ) .await .context("delete accepted Agent")?; tx.execute( - "UPDATE enrollment_attempts SET deleted = 1 WHERE jti = ?1", + "UPDATE agent_tunnel_enrollment_attempts SET deleted = 1 WHERE jti = ?1", params![enrollment_jti], ) .await @@ -526,6 +596,14 @@ impl AgentAuthorizationStore for LibSqlAgentAuthorizationStore { mod tests { use super::*; + async fn connect(path: &str) -> anyhow::Result { + gateway_db::GatewayDb::open_path(path).await?.connect().await + } + + async fn open_store(path: &str, ca_spki_sha256: SpkiSha256) -> anyhow::Result { + LibSqlAgentAuthorizationStore::open(connect(path).await?, ca_spki_sha256).await + } + fn enrollment(agent_id: Uuid) -> EnrollmentAttempt { EnrollmentAttempt { token_id: Uuid::new_v4(), @@ -540,19 +618,19 @@ mod tests { #[tokio::test] async fn accepted_agent_survives_reopen() { let temp_dir = tempfile::tempdir().expect("create temporary directory"); - let database_path = temp_dir.path().join("agent_tunnel.db"); + let database_path = temp_dir.path().join("gateway.db"); let database_path = database_path.to_str().expect("temporary database path is UTF-8"); let ca_spki_sha256 = [0xCA; 32]; let agent_id = Uuid::new_v4(); - let store = LibSqlAgentAuthorizationStore::open(database_path, ca_spki_sha256) + let store = open_store(database_path, ca_spki_sha256) .await .expect("open Agent authorization store"); let outcome = store.enroll(enrollment(agent_id)).await.expect("enroll Agent"); assert!(matches!(outcome, EnrollmentOutcome::Created(_))); drop(store); - let reopened = LibSqlAgentAuthorizationStore::open(database_path, ca_spki_sha256) + let reopened = open_store(database_path, ca_spki_sha256) .await .expect("reopen Agent authorization store"); let accepted = reopened @@ -567,7 +645,7 @@ mod tests { #[tokio::test] async fn identical_enrollment_retry_is_idempotent() { - let store = LibSqlAgentAuthorizationStore::open(":memory:", [0xCA; 32]) + let store = open_store(":memory:", [0xCA; 32]) .await .expect("open Agent authorization store"); let attempt = enrollment(Uuid::new_v4()); @@ -581,7 +659,7 @@ mod tests { #[tokio::test] async fn delete_prevents_original_enrollment_retry() { - let store = LibSqlAgentAuthorizationStore::open(":memory:", [0xCA; 32]) + let store = open_store(":memory:", [0xCA; 32]) .await .expect("open Agent authorization store"); let attempt = enrollment(Uuid::new_v4()); @@ -607,7 +685,7 @@ mod tests { #[tokio::test] async fn accepted_agent_id_and_name_are_unique() { - let store = LibSqlAgentAuthorizationStore::open(":memory:", [0xCA; 32]) + let store = open_store(":memory:", [0xCA; 32]) .await .expect("open Agent authorization store"); let first = enrollment(Uuid::new_v4()); @@ -626,7 +704,7 @@ mod tests { #[tokio::test] async fn list_and_get_include_offline_accepted_agents() { - let store = LibSqlAgentAuthorizationStore::open(":memory:", [0xCA; 32]) + let store = open_store(":memory:", [0xCA; 32]) .await .expect("open Agent authorization store"); let first = enrollment(Uuid::new_v4()); @@ -654,7 +732,7 @@ mod tests { #[tokio::test] async fn enrollment_token_rejects_different_material() { - let store = LibSqlAgentAuthorizationStore::open(":memory:", [0xCA; 32]) + let store = open_store(":memory:", [0xCA; 32]) .await .expect("open Agent authorization store"); let first = enrollment(Uuid::new_v4()); @@ -669,7 +747,7 @@ mod tests { #[tokio::test] async fn new_token_and_key_can_reenroll_deleted_agent() { - let store = LibSqlAgentAuthorizationStore::open(":memory:", [0xCA; 32]) + let store = open_store(":memory:", [0xCA; 32]) .await .expect("open Agent authorization store"); let first = enrollment(Uuid::new_v4()); @@ -708,9 +786,9 @@ mod tests { #[tokio::test] async fn new_token_cannot_reenroll_deleted_agent_key() { let temp_dir = tempfile::tempdir().expect("create temporary directory"); - let database_path = temp_dir.path().join("agent_tunnel.db"); + let database_path = temp_dir.path().join("gateway.db"); let database_path = database_path.to_str().expect("temporary database path is UTF-8"); - let store = LibSqlAgentAuthorizationStore::open(database_path, [0xCA; 32]) + let store = open_store(database_path, [0xCA; 32]) .await .expect("open Agent authorization store"); let first = enrollment(Uuid::new_v4()); @@ -722,7 +800,7 @@ mod tests { .expect("Agent was accepted"); drop(store); - let reopened = LibSqlAgentAuthorizationStore::open(database_path, [0xCA; 32]) + let reopened = open_store(database_path, [0xCA; 32]) .await .expect("reopen Agent authorization store"); let replay = enrollment(first.agent_id); @@ -733,7 +811,7 @@ mod tests { #[tokio::test] async fn new_agent_id_cannot_reuse_deleted_agent_key() { - let store = LibSqlAgentAuthorizationStore::open(":memory:", [0xCA; 32]) + let store = open_store(":memory:", [0xCA; 32]) .await .expect("open Agent authorization store"); let first = enrollment(Uuid::new_v4()); @@ -754,14 +832,14 @@ mod tests { #[tokio::test] async fn database_rejects_another_ca() { let temp_dir = tempfile::tempdir().expect("create temporary directory"); - let database_path = temp_dir.path().join("agent_tunnel.db"); + let database_path = temp_dir.path().join("gateway.db"); let database_path = database_path.to_str().expect("temporary database path is UTF-8"); - let store = LibSqlAgentAuthorizationStore::open(database_path, [0xCA; 32]) + let store = open_store(database_path, [0xCA; 32]) .await .expect("open Agent authorization store"); drop(store); - let error = LibSqlAgentAuthorizationStore::open(database_path, [0xBB; 32]) + let error = open_store(database_path, [0xBB; 32]) .await .err() .expect("database must reject another CA"); @@ -769,67 +847,113 @@ mod tests { } #[tokio::test] - async fn database_rejects_a_newer_schema() { + async fn first_open_binds_the_ca() { let temp_dir = tempfile::tempdir().expect("create temporary directory"); - let database_path = temp_dir.path().join("agent_tunnel.db"); + let database_path = temp_dir.path().join("gateway.db"); let database_path = database_path.to_str().expect("temporary database path is UTF-8"); - let store = LibSqlAgentAuthorizationStore::open(database_path, [0xCA; 32]) + + let conn = connect(database_path).await.expect("open gateway database"); + let bound = LibSqlAgentAuthorizationStore::bound_ca(&conn) + .await + .expect("read bound CA"); + assert_eq!(bound, None); + + let store = LibSqlAgentAuthorizationStore::open(conn, [0xCA; 32]) .await .expect("open Agent authorization store"); drop(store); - let database = libsql::Builder::new_local(database_path) - .build() + let conn = connect(database_path).await.expect("reopen gateway database"); + let bound = LibSqlAgentAuthorizationStore::bound_ca(&conn) .await - .expect("open Agent authorization database"); - let conn = database.connect().expect("connect to Agent authorization database"); - conn.execute_batch("PRAGMA user_version = 99") + .expect("read bound CA"); + assert_eq!(bound, Some([0xCA; 32])); + } + + #[tokio::test] + async fn database_with_agents_but_no_ca_is_rejected() { + let temp_dir = tempfile::tempdir().expect("create temporary directory"); + let database_path = temp_dir.path().join("gateway.db"); + let database_path = database_path.to_str().expect("temporary database path is UTF-8"); + let store = open_store(database_path, [0xCA; 32]) .await - .expect("set unsupported schema version"); - drop(conn); - drop(database); + .expect("open Agent authorization store"); + store.enroll(enrollment(Uuid::new_v4())).await.expect("enroll Agent"); + drop(store); - let error = LibSqlAgentAuthorizationStore::open(database_path, [0xCA; 32]) + let conn = connect(database_path).await.expect("reopen gateway database"); + conn.execute( + "DELETE FROM agent_tunnel_metadata WHERE key = ?1", + params![CA_SPKI_SHA256_KEY], + ) + .await + .expect("delete CA metadata"); + + let error = LibSqlAgentAuthorizationStore::open(conn, [0xBB; 32]) .await .err() - .expect("database must reject a newer schema"); - assert!(error.to_string().contains("newer than supported")); + .expect("Agents without a CA must not be bound to a new CA"); + assert!(error.to_string().contains("missing CA")); } #[tokio::test] - async fn initialized_database_rejects_missing_ca_metadata() { - let temp_dir = tempfile::tempdir().expect("create temporary directory"); - let database_path = temp_dir.path().join("agent_tunnel.db"); - let database_path = database_path.to_str().expect("temporary database path is UTF-8"); - let store = LibSqlAgentAuthorizationStore::open(database_path, [0xCA; 32]) + async fn new_tables_record_their_schema_version_in_metadata() { + let conn = connect(":memory:").await.expect("open gateway database"); + + let store = LibSqlAgentAuthorizationStore::open(conn, [0xCA; 32]) .await .expect("open Agent authorization store"); - drop(store); + let conn = store.conn.lock().await; - let database = libsql::Builder::new_local(database_path) - .build() + assert_eq!( + schema_version(&conn).await.expect("read schema version"), + MIGRATIONS.len() + ); + let user_version = conn + .query("PRAGMA user_version", ()) .await - .expect("open Agent authorization database"); - let conn = database.connect().expect("connect to Agent authorization database"); - conn.execute("DELETE FROM metadata WHERE key = ?1", params![CA_SPKI_SHA256_KEY]) + .expect("query user_version") + .next() .await - .expect("delete CA metadata"); - drop(conn); - drop(database); + .expect("read user_version") + .expect("user_version row") + .get::(0) + .expect("decode user_version"); + assert_eq!(user_version, 0, "the file's user_version belongs to no feature"); + } + + #[tokio::test] + async fn newer_schema_is_rejected() { + let temp_dir = tempfile::tempdir().expect("create temporary directory"); + let database_path = temp_dir.path().join("gateway.db"); + let database_path = database_path.to_str().expect("temporary database path is UTF-8"); + drop( + open_store(database_path, [0xCA; 32]) + .await + .expect("open Agent authorization store"), + ); - let error = LibSqlAgentAuthorizationStore::open(database_path, [0xCA; 32]) + let conn = connect(database_path).await.expect("reopen gateway database"); + conn.execute( + "UPDATE agent_tunnel_metadata SET value = 99 WHERE key = ?1", + params![SCHEMA_VERSION_KEY], + ) + .await + .expect("set unsupported schema version"); + + let error = LibSqlAgentAuthorizationStore::open(conn, [0xCA; 32]) .await .err() - .expect("initialized database must require CA metadata"); - assert!(error.to_string().contains("missing CA")); + .expect("newer schema must be rejected"); + assert!(error.to_string().contains("newer than supported")); } #[tokio::test] async fn database_rejects_missing_required_table() { let temp_dir = tempfile::tempdir().expect("create temporary directory"); - let database_path = temp_dir.path().join("agent_tunnel.db"); + let database_path = temp_dir.path().join("gateway.db"); let database_path = database_path.to_str().expect("temporary database path is UTF-8"); - let store = LibSqlAgentAuthorizationStore::open(database_path, [0xCA; 32]) + let store = open_store(database_path, [0xCA; 32]) .await .expect("open Agent authorization store"); drop(store); @@ -839,13 +963,13 @@ mod tests { .await .expect("open Agent authorization database"); let conn = database.connect().expect("connect to Agent authorization database"); - conn.execute("DROP TABLE accepted_agents", ()) + conn.execute("DROP TABLE agent_tunnel_accepted_agents", ()) .await .expect("drop required table"); drop(conn); drop(database); - let error = LibSqlAgentAuthorizationStore::open(database_path, [0xCA; 32]) + let error = open_store(database_path, [0xCA; 32]) .await .err() .expect("database must reject missing required table"); @@ -855,9 +979,9 @@ mod tests { #[tokio::test] async fn database_rejects_missing_required_unique_index() { let temp_dir = tempfile::tempdir().expect("create temporary directory"); - let database_path = temp_dir.path().join("agent_tunnel.db"); + let database_path = temp_dir.path().join("gateway.db"); let database_path = database_path.to_str().expect("temporary database path is UTF-8"); - let store = LibSqlAgentAuthorizationStore::open(database_path, [0xCA; 32]) + let store = open_store(database_path, [0xCA; 32]) .await .expect("open Agent authorization store"); drop(store); @@ -869,8 +993,8 @@ mod tests { let conn = database.connect().expect("connect to Agent authorization database"); conn.execute_batch( " - ALTER TABLE accepted_agents RENAME TO accepted_agents_with_name_index; - CREATE TABLE accepted_agents ( + ALTER TABLE agent_tunnel_accepted_agents RENAME TO agent_tunnel_accepted_agents_with_name_index; + CREATE TABLE agent_tunnel_accepted_agents ( agent_id TEXT PRIMARY KEY, name TEXT NOT NULL COLLATE NOCASE, client_spki_sha256 BLOB NOT NULL, @@ -879,7 +1003,7 @@ mod tests { CHECK (name = trim(name)), CHECK (length(client_spki_sha256) = 32) ); - DROP TABLE accepted_agents_with_name_index; + DROP TABLE agent_tunnel_accepted_agents_with_name_index; ", ) .await @@ -887,7 +1011,7 @@ mod tests { drop(conn); drop(database); - let error = LibSqlAgentAuthorizationStore::open(database_path, [0xCA; 32]) + let error = open_store(database_path, [0xCA; 32]) .await .err() .expect("database must reject missing Agent name index"); diff --git a/crates/agent-tunnel-libsql/src/upgrade.rs b/crates/agent-tunnel-libsql/src/upgrade.rs new file mode 100644 index 000000000..6f72293cf --- /dev/null +++ b/crates/agent-tunnel-libsql/src/upgrade.rs @@ -0,0 +1,475 @@ +//! The upgrade from Gateway 2026.3.0, which kept the agent tunnel tables in an `agent_tunnel.db` of their own. + +use anyhow::{Context as _, bail}; +use camino::{Utf8Path, Utf8PathBuf}; +use libsql::{Connection, params}; + +use crate::record_schema_version; + +/// Only Gateway 2026.3.0 wrote this file. +pub(crate) const LEGACY_FILE_NAME: &str = "agent_tunnel.db"; + +/// The only schema 2026.3.0 ever had, as its `PRAGMA user_version`. +const LEGACY_SCHEMA_VERSION: i64 = 1; + +/// The conversion yields the tables of the first migration. +const CONVERTED_SCHEMA_VERSION: i64 = 1; + +/// Turns the 2026.3.0 tables into the first version of the `agent_tunnel_*` tables. +const CONVERSION: &str = " + ALTER TABLE metadata RENAME TO agent_tunnel_metadata; + ALTER TABLE accepted_agents RENAME TO agent_tunnel_accepted_agents; + ALTER TABLE enrollment_attempts RENAME TO agent_tunnel_enrollment_attempts; + ALTER TABLE deleted_agent_keys RENAME TO agent_tunnel_deleted_agent_keys; +"; + +/// Turns the `agent_tunnel.db` left by Gateway 2026.3.0 in `data_dir` into `gateway.db`. +/// +/// Call it before opening `gateway.db`: once `gateway.db` exists, the old file is ignored. It does nothing when there +/// is no old file. +/// +/// We convert a copy, flush it to disk, and only then rename it into place, so `gateway.db` either doesn't exist yet +/// or is whole and in the new layout. The rename reaches the disk before the old files are removed, so a power loss +/// never leaves only a partial copy behind. If Gateway stops halfway, the next start simply does it again. +pub async fn import_2026_3_0_database(data_dir: &Utf8Path) -> anyhow::Result<()> { + let legacy = data_dir.join(LEGACY_FILE_NAME); + let path = data_dir.join(gateway_db::FILE_NAME); + + if !exists(&legacy)? { + return Ok(()); + } + + if exists(&path)? { + warn!(%legacy, "Ignoring agent_tunnel.db because gateway.db already exists"); + return Ok(()); + } + + let legacy_conn = connect(&legacy).await?; + + match legacy_schema_version(&legacy_conn).await? { + // A first start that failed before creating the tables: there is nothing to import. + 0 => { + if has_tables(&legacy_conn).await? { + bail!("{legacy} has tables but no schema version"); + } + + drop(legacy_conn); + + if remove_with_sidecars(&legacy) { + info!(%legacy, "Removed an empty agent tunnel database"); + } + + return Ok(()); + } + LEGACY_SCHEMA_VERSION => {} + version => bail!("unsupported schema version {version} in {legacy}"), + } + + let copy = Utf8PathBuf::from(format!("{path}.tmp")); + + // A leftover journal of an interrupted attempt would be replayed into the new copy. + for file in with_sidecars(©) { + gateway_db::remove_file_if_exists(&file).with_context(|| format!("remove the incomplete copy {file}"))?; + } + + legacy_conn + .execute("VACUUM INTO ?1", params![copy.as_str()]) + .await + .with_context(|| format!("copy {legacy} to {copy}"))?; + drop(legacy_conn); + + convert(©).await?; + + gateway_db::flush_file(©).with_context(|| format!("flush {copy} to disk"))?; + gateway_db::rename_durably(©, &path).with_context(|| format!("rename {copy} to {path}"))?; + + // A failed removal is only noise: gateway.db exists now, so the old files are ignored. + let _ = remove_with_sidecars(&legacy); + + info!(from = %legacy, to = %path, "Moved the agent tunnel database into the gateway database"); + + Ok(()) +} + +async fn convert(copy: &Utf8Path) -> anyhow::Result<()> { + let conn = connect(copy).await?; + + // The changes must be in the file itself before the rename, not in a `-wal` file left behind. SQLite keeps the old + // mode without an error when it can't switch, so we check the mode it reports. + let journal_mode = conn + .query("PRAGMA journal_mode = DELETE", ()) + .await + .with_context(|| format!("use a rollback journal for {copy}"))? + .next() + .await + .with_context(|| format!("read the journal mode of {copy}"))? + .context("journal mode query returned no row")? + .get::(0) + .context("decode the journal mode")?; + if !journal_mode.eq_ignore_ascii_case("delete") { + bail!("journal mode of {copy} is {journal_mode}, not delete"); + } + + let tx = conn.transaction().await.context("begin agent tunnel conversion")?; + tx.execute_batch(CONVERSION) + .await + .context("add the agent tunnel table prefixes")?; + record_schema_version(&tx, CONVERTED_SCHEMA_VERSION).await?; + tx.execute_batch("PRAGMA user_version = 0") + .await + .context("clear the 2026.3.0 schema version")?; + tx.commit().await.context("commit agent tunnel conversion")?; + + Ok(()) +} + +async fn connect(path: &Utf8Path) -> anyhow::Result { + libsql::Builder::new_local(path.as_str()) + .build() + .await + .with_context(|| format!("build database {path}"))? + .connect() + .with_context(|| format!("connect to database {path}")) +} + +fn exists(path: &Utf8Path) -> anyhow::Result { + path.try_exists().with_context(|| format!("look for {path}")) +} + +async fn has_tables(conn: &Connection) -> anyhow::Result { + let row = conn + .query("SELECT 1 FROM sqlite_master LIMIT 1", ()) + .await + .context("query the 2026.3.0 agent tunnel tables")? + .next() + .await + .context("read the 2026.3.0 agent tunnel tables")?; + + Ok(row.is_some()) +} + +async fn legacy_schema_version(conn: &Connection) -> anyhow::Result { + conn.query("PRAGMA user_version", ()) + .await + .context("query the 2026.3.0 agent tunnel schema version")? + .next() + .await + .context("read the 2026.3.0 agent tunnel schema version")? + .context("schema version query returned no row")? + .get::(0) + .context("decode the 2026.3.0 agent tunnel schema version") +} + +fn with_sidecars(path: &Utf8Path) -> [Utf8PathBuf; 4] { + ["", "-journal", "-wal", "-shm"].map(|suffix| Utf8PathBuf::from(format!("{path}{suffix}"))) +} + +/// Removes the database at `path` with its journals; returns whether every file is gone. +fn remove_with_sidecars(path: &Utf8Path) -> bool { + let mut removed = true; + + for file in with_sidecars(path) { + if let Err(error) = gateway_db::remove_file_if_exists(&file) { + warn!(%error, path = %file, "Failed to remove an old agent tunnel database file"); + removed = false; + } + } + + removed +} + +#[cfg(test)] +mod tests { + use agent_tunnel::authorization::AgentAuthorizationStore as _; + use gateway_db::GatewayDb; + use uuid::Uuid; + + use super::*; + use crate::LibSqlAgentAuthorizationStore; + + /// The tables Gateway 2026.3.0 created in `agent_tunnel.db`. + const LEGACY_SCHEMA: &str = " + CREATE TABLE metadata (key TEXT PRIMARY KEY, value BLOB NOT NULL); + CREATE TABLE accepted_agents ( + agent_id TEXT PRIMARY KEY, + name TEXT NOT NULL COLLATE NOCASE UNIQUE, + client_spki_sha256 BLOB NOT NULL, + enrollment_jti TEXT NOT NULL UNIQUE, + CHECK (length(name) BETWEEN 1 AND 255), + CHECK (name = trim(name)), + CHECK (length(client_spki_sha256) = 32) + ); + CREATE TABLE enrollment_attempts ( + jti TEXT PRIMARY KEY, + agent_id TEXT NOT NULL, + request_sha256 BLOB NOT NULL, + expires_at INTEGER NOT NULL, + deleted INTEGER NOT NULL DEFAULT 0, + CHECK (length(request_sha256) = 32), + CHECK (deleted IN (0, 1)) + ); + CREATE TABLE deleted_agent_keys ( + client_spki_sha256 BLOB PRIMARY KEY, + agent_id TEXT NOT NULL, + CHECK (length(client_spki_sha256) = 32) + ); + "; + + fn temp_data_dir() -> (tempfile::TempDir, Utf8PathBuf) { + let temp_dir = tempfile::tempdir().expect("create temporary directory"); + let data_dir = Utf8PathBuf::from_path_buf(temp_dir.path().to_path_buf()).expect("temporary path is UTF-8"); + (temp_dir, data_dir) + } + + /// Writes the `agent_tunnel.db` that Gateway 2026.3.0 leaves behind, with one Agent, and returns its ID. + async fn write_legacy_database(data_dir: &Utf8Path) -> Uuid { + let conn = connect(&data_dir.join(LEGACY_FILE_NAME)) + .await + .expect("open 2026.3.0 database"); + conn.execute_batch("PRAGMA journal_mode = WAL") + .await + .expect("use WAL like 2026.3.0"); + fill_legacy_database(&conn).await + } + + /// Creates the 2026.3.0 tables with one Agent, and returns its ID. + async fn fill_legacy_database(conn: &Connection) -> Uuid { + let agent_id = Uuid::new_v4(); + conn.execute_batch(LEGACY_SCHEMA).await.expect("create 2026.3.0 tables"); + conn.execute( + "INSERT INTO metadata (key, value) VALUES ('ca_spki_sha256', ?1)", + params![vec![0xCAu8; 32]], + ) + .await + .expect("bind 2026.3.0 CA"); + conn.execute( + "INSERT INTO accepted_agents (agent_id, name, client_spki_sha256, enrollment_jti) VALUES (?1, ?2, ?3, ?4)", + params![ + agent_id.to_string(), + "montreal-office", + vec![0x11u8; 32], + Uuid::new_v4().to_string() + ], + ) + .await + .expect("store 2026.3.0 Agent"); + conn.execute_batch("PRAGMA user_version = 1") + .await + .expect("record 2026.3.0 schema version"); + agent_id + } + + async fn open_gateway_db(data_dir: &Utf8Path) -> Connection { + GatewayDb::open(data_dir) + .await + .expect("open gateway database") + .connect() + .await + .expect("connect to gateway database") + } + + async fn query_i64(conn: &Connection, sql: &str) -> i64 { + conn.query(sql, ()) + .await + .expect("run query") + .next() + .await + .expect("read row") + .expect("query returns a row") + .get::(0) + .expect("decode integer") + } + + #[tokio::test] + async fn agents_enrolled_on_2026_3_0_still_authorize_after_the_move() { + let (_temp_dir, data_dir) = temp_data_dir(); + let agent_id = write_legacy_database(&data_dir).await; + + import_2026_3_0_database(&data_dir) + .await + .expect("import 2026.3.0 database"); + + for file in with_sidecars(&data_dir.join(LEGACY_FILE_NAME)) { + assert!(!file.exists(), "{file} is removed"); + } + let path = data_dir.join(gateway_db::FILE_NAME); + let copy = Utf8PathBuf::from(format!("{path}.tmp")); + for file in with_sidecars(©) + .into_iter() + .chain(with_sidecars(&path).into_iter().skip(1)) + { + assert!(!file.exists(), "{file} is not left behind"); + } + // Bytes 18 and 19 of the header are 1 for a rollback journal and 2 for WAL. + let header = std::fs::read(&path).expect("read gateway.db"); + assert_eq!(header[18..20], [1, 1], "gateway.db is renamed in rollback journal mode"); + + let conn = open_gateway_db(&data_dir).await; + assert_eq!(query_i64(&conn, "PRAGMA user_version").await, 0); + assert_eq!( + query_i64( + &conn, + "SELECT value FROM agent_tunnel_metadata WHERE key = 'schema_version'" + ) + .await, + CONVERTED_SCHEMA_VERSION + ); + assert_eq!( + LibSqlAgentAuthorizationStore::bound_ca(&conn) + .await + .expect("read bound CA"), + Some([0xCA; 32]) + ); + + let store = LibSqlAgentAuthorizationStore::open(conn, [0xCA; 32]) + .await + .expect("open Agent authorization store"); + let accepted = store + .authorize(agent_id, [0x11; 32]) + .await + .expect("authorize moved Agent") + .expect("Agent is still accepted"); + assert_eq!(accepted.name, "montreal-office"); + } + + #[tokio::test] + async fn agents_still_in_the_2026_3_0_wal_are_imported() { + let (_source_dir, source) = temp_data_dir(); + let (_temp_dir, data_dir) = temp_data_dir(); + let legacy = source.join(LEGACY_FILE_NAME); + let conn = connect(&legacy).await.expect("open 2026.3.0 database"); + conn.execute_batch("PRAGMA journal_mode = WAL; PRAGMA wal_autocheckpoint = 0") + .await + .expect("keep changes in the WAL"); + let agent_id = fill_legacy_database(&conn).await; + + // Copied while open, as if 2026.3.0 had crashed before writing its WAL into the database. + for suffix in ["", "-wal"] { + std::fs::copy( + format!("{legacy}{suffix}"), + format!("{}{suffix}", data_dir.join(LEGACY_FILE_NAME)), + ) + .expect("copy the 2026.3.0 files"); + } + drop(conn); + assert!( + std::fs::metadata(data_dir.join(format!("{LEGACY_FILE_NAME}-wal"))) + .expect("WAL copied") + .len() + > 0 + ); + + import_2026_3_0_database(&data_dir) + .await + .expect("import 2026.3.0 database"); + + let store = LibSqlAgentAuthorizationStore::open(open_gateway_db(&data_dir).await, [0xCA; 32]) + .await + .expect("open Agent authorization store"); + assert!( + store + .authorize(agent_id, [0x11; 32]) + .await + .expect("authorize moved Agent") + .is_some() + ); + } + + #[tokio::test] + async fn interrupted_move_starts_over() { + let (_temp_dir, data_dir) = temp_data_dir(); + write_legacy_database(&data_dir).await; + let copy = Utf8PathBuf::from(format!("{}.tmp", data_dir.join(gateway_db::FILE_NAME))); + std::fs::write(©, b"half written").expect("leave an incomplete copy"); + std::fs::write(format!("{copy}-journal"), b"stale journal").expect("leave its journal"); + + import_2026_3_0_database(&data_dir) + .await + .expect("import 2026.3.0 database"); + + for file in with_sidecars(©) { + assert!(!file.exists(), "{file} is removed"); + } + let conn = open_gateway_db(&data_dir).await; + LibSqlAgentAuthorizationStore::open(conn, [0xCA; 32]) + .await + .expect("open Agent authorization store"); + } + + #[tokio::test] + async fn existing_gateway_database_is_kept() { + let (_temp_dir, data_dir) = temp_data_dir(); + drop(open_gateway_db(&data_dir).await); + write_legacy_database(&data_dir).await; + + import_2026_3_0_database(&data_dir) + .await + .expect("ignore 2026.3.0 database"); + + assert!(data_dir.join(LEGACY_FILE_NAME).exists()); + let conn = open_gateway_db(&data_dir).await; + assert_eq!( + LibSqlAgentAuthorizationStore::bound_ca(&conn) + .await + .expect("read bound CA"), + None + ); + } + + #[tokio::test] + async fn empty_2026_3_0_database_is_removed() { + let (_temp_dir, data_dir) = temp_data_dir(); + let conn = connect(&data_dir.join(LEGACY_FILE_NAME)) + .await + .expect("create empty 2026.3.0 database"); + conn.execute_batch("PRAGMA journal_mode = WAL") + .await + .expect("write the database header"); + drop(conn); + assert!(data_dir.join(LEGACY_FILE_NAME).exists()); + + import_2026_3_0_database(&data_dir) + .await + .expect("import empty 2026.3.0 database"); + + assert!(!data_dir.join(LEGACY_FILE_NAME).exists()); + assert!(!data_dir.join(gateway_db::FILE_NAME).exists()); + } + + #[tokio::test] + async fn version_0_with_tables_is_refused() { + let (_temp_dir, data_dir) = temp_data_dir(); + let conn = connect(&data_dir.join(LEGACY_FILE_NAME)) + .await + .expect("create 2026.3.0 database"); + conn.execute_batch(LEGACY_SCHEMA).await.expect("create 2026.3.0 tables"); + drop(conn); + + let error = import_2026_3_0_database(&data_dir) + .await + .expect_err("tables without a schema version must be refused"); + + assert!(error.to_string().contains("no schema version")); + assert!(data_dir.join(LEGACY_FILE_NAME).exists()); + } + + #[tokio::test] + async fn unknown_2026_3_0_schema_is_refused() { + let (_temp_dir, data_dir) = temp_data_dir(); + let conn = connect(&data_dir.join(LEGACY_FILE_NAME)) + .await + .expect("create 2026.3.0 database"); + conn.execute_batch("PRAGMA user_version = 7") + .await + .expect("set unknown schema version"); + drop(conn); + + let error = import_2026_3_0_database(&data_dir) + .await + .expect_err("unknown schema must be refused"); + + assert!(error.to_string().contains("unsupported schema version 7")); + assert!(data_dir.join(LEGACY_FILE_NAME).exists()); + assert!(!data_dir.join(gateway_db::FILE_NAME).exists()); + } +} diff --git a/crates/gateway-db/AGENTS.md b/crates/gateway-db/AGENTS.md new file mode 100644 index 000000000..6f0c39a62 --- /dev/null +++ b/crates/gateway-db/AGENTS.md @@ -0,0 +1,9 @@ +# gateway-db + +This crate owns the shared `gateway.db` file and nothing else. +Read `INTENT.md` in this directory first. + +- Never add a dependency on a crate that implements feature behavior, such as `agent-tunnel` or `agent-tunnel-libsql`. + A feature's tables, migrations and schema version belong in its own adapter crate, which may depend on `gateway-db`. +- Never read or write `PRAGMA user_version` here: no single crate owns every table in the file. +- After changing `Cargo.toml`, run `cargo tree -p gateway-db --edges normal --depth 1` and check that only generic crates are listed. diff --git a/crates/gateway-db/Cargo.toml b/crates/gateway-db/Cargo.toml new file mode 100644 index 000000000..1da7eb4c6 --- /dev/null +++ b/crates/gateway-db/Cargo.toml @@ -0,0 +1,21 @@ +[package] +name = "gateway-db" +version = "0.0.0" +edition = "2024" +authors = ["Devolutions Inc. "] +publish = false + +[lints] +workspace = true + +[dependencies] +anyhow = "1" +camino = "1.1" +libsql = { version = "0.9", default-features = false, features = ["core"] } + +[target.'cfg(windows)'.dependencies] +windows-sys = { version = "0.61", features = ["Win32_Storage_FileSystem"] } + +[dev-dependencies] +tempfile = "3.24" +tokio = { version = "1.52", features = ["macros", "rt-multi-thread"] } diff --git a/crates/gateway-db/INTENT.md b/crates/gateway-db/INTENT.md new file mode 100644 index 000000000..100c2c464 --- /dev/null +++ b/crates/gateway-db/INTENT.md @@ -0,0 +1,23 @@ +# Background +Gateway needs a database to store information for certain features. + +# Invariants +`gateway-db` only owns the shared `gateway.db` file: it opens the file, applies the PRAGMAs, and hands out a connection per feature. + +`gateway-db` must never depend on crates that implement feature behavior, such as `agent-tunnel`. Adapters (for example `agent-tunnel-libsql`) may depend on `gateway-db`, never the other way around. + +For each domain table, the table should have its prefix in table's name, for example, `agent_tunnel_*` for agent tunnel domain tables. The prefix is the compatibility boundary. + +`gateway-db` does not read or write `PRAGMA user_version`. Each adapter owns its migrations and tracks its own schema version within its prefixed tables, for example as a row in `agent_tunnel_metadata`. Migrations are append-only. NEVER remove or reorder a migration. + + +# Goals +All persistent state except the job queue and traffic audit should be stored in this database. + + +# Upgrade from 2026.3.0 +If for 2026.3.0, user turned on the feature of agent-tunnel, then the database will be migrated from `agent_tunnel.db` to `gateway.db`. The migration will be done automatically on Gateway start up. This is one-off, atomic operation. + +# Future +Later, `gateway-db` could take migrations from each adapter. +An adapter would pass its prefix and its list of migrations, and `gateway-db` would run them and track a version per prefix \ No newline at end of file diff --git a/crates/gateway-db/src/lib.rs b/crates/gateway-db/src/lib.rs new file mode 100644 index 000000000..8f96df4f7 --- /dev/null +++ b/crates/gateway-db/src/lib.rs @@ -0,0 +1,196 @@ +//! Gateway's own database, `gateway.db`. +//! +//! Features that need to remember a little data keep it here, in tables named after the feature (`agent_tunnel_*`, +//! ...), so we don't end up with one database file per feature. This crate only owns the file: it opens it, applies +//! the PRAGMAs and hands a connection to each feature. Each feature's tables, migrations and schema version live in +//! the feature's own adapter crate, such as `agent-tunnel-libsql`; the table prefix is the compatibility boundary. +//! +//! Queues are the exception: the job queue and the traffic audit tune their databases for queue work, so they keep +//! their own files. + +use anyhow::Context as _; +use camino::Utf8Path; +use libsql::{Connection, Database}; + +/// The name of the database file in Gateway's data directory. +pub const FILE_NAME: &str = "gateway.db"; + +const PRAGMAS: &str = " + PRAGMA journal_mode = WAL; + PRAGMA synchronous = FULL; + PRAGMA busy_timeout = 15000; + PRAGMA foreign_keys = ON; + PRAGMA temp_store = MEMORY; +"; + +/// The open `gateway.db`, ready to hand a connection to each feature that keeps tables in it. +#[derive(Debug)] +pub struct GatewayDb { + database: Database, +} + +impl GatewayDb { + /// Opens `gateway.db` in `data_dir`, creating it if needed. + pub async fn open(data_dir: &Utf8Path) -> anyhow::Result { + Self::open_path(data_dir.join(FILE_NAME).as_str()).await + } + + /// Opens the database at `path`, creating it if needed. + /// + /// With `:memory:`, every connection gets its own empty database, which is handy in tests. + pub async fn open_path(path: &str) -> anyhow::Result { + let database = libsql::Builder::new_local(path) + .build() + .await + .with_context(|| format!("build database {path}"))?; + let db = Self { database }; + + db.connect().await?; + + Ok(db) + } + + /// Gives a feature a connection of its own. + /// + /// SQLite tracks transactions per connection, so features sharing one connection would end up inside each + /// other's transactions. Each feature should call this once and keep its connection. + pub async fn connect(&self) -> anyhow::Result { + let conn = self.database.connect().context("connect to the gateway database")?; + + conn.execute_batch(PRAGMAS) + .await + .context("apply gateway database PRAGMAs")?; + + Ok(conn) + } +} + +/// Waits until the content of the file at `path` is on disk. +pub fn flush_file(path: &Utf8Path) -> std::io::Result<()> { + std::fs::OpenOptions::new().write(true).open(path)?.sync_all() +} + +/// Renames `from` to `to` and waits until the rename is on disk. +#[cfg(unix)] +pub fn rename_durably(from: &Utf8Path, to: &Utf8Path) -> std::io::Result<()> { + std::fs::rename(from, to)?; + + // On Unix, the new name is only durable once its directory is flushed. + let dir = match to.parent() { + Some(dir) if !dir.as_str().is_empty() => dir, + _ => Utf8Path::new("."), + }; + std::fs::File::open(dir)?.sync_all() +} + +/// Renames `from` to `to` and waits until the rename is on disk. +#[cfg(windows)] +pub fn rename_durably(from: &Utf8Path, to: &Utf8Path) -> std::io::Result<()> { + use std::os::windows::ffi::OsStrExt as _; + + use windows_sys::Win32::Storage::FileSystem::{MOVEFILE_WRITE_THROUGH, MoveFileExW}; + + let wide = |path: &Utf8Path| -> Vec { + path.as_std_path() + .as_os_str() + .encode_wide() + .chain(std::iter::once(0)) + .collect() + }; + let from = wide(from); + let to = wide(to); + + // SAFETY: Both paths are null-terminated UTF-16 strings that outlive the call. + let succeeded = unsafe { MoveFileExW(from.as_ptr(), to.as_ptr(), MOVEFILE_WRITE_THROUGH) }; + + if succeeded == 0 { + return Err(std::io::Error::last_os_error()); + } + + Ok(()) +} + +/// Removes the file at `path`, if there is one. +pub fn remove_file_if_exists(path: &Utf8Path) -> std::io::Result<()> { + match std::fs::remove_file(path) { + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()), + result => result, + } +} + +#[cfg(test)] +mod tests { + use camino::Utf8PathBuf; + use libsql::TransactionBehavior; + + use super::*; + + fn temp_data_dir() -> (tempfile::TempDir, Utf8PathBuf) { + let temp_dir = tempfile::tempdir().expect("create temporary directory"); + let data_dir = Utf8PathBuf::from_path_buf(temp_dir.path().to_path_buf()).expect("temporary path is UTF-8"); + (temp_dir, data_dir) + } + + async fn query_i64(conn: &Connection, sql: &str) -> i64 { + conn.query(sql, ()) + .await + .expect("run query") + .next() + .await + .expect("read row") + .expect("query returns a row") + .get::(0) + .expect("decode integer") + } + + #[tokio::test] + async fn open_creates_an_empty_database_without_a_schema_version() { + let (_temp_dir, data_dir) = temp_data_dir(); + + let conn = GatewayDb::open(&data_dir) + .await + .expect("open gateway database") + .connect() + .await + .expect("connect to gateway database"); + + assert!(data_dir.join(FILE_NAME).exists()); + assert_eq!(query_i64(&conn, "PRAGMA user_version").await, 0); + assert_eq!(query_i64(&conn, "SELECT count(*) FROM sqlite_master").await, 0); + } + + #[tokio::test] + async fn each_connection_has_its_own_transactions() { + let (_temp_dir, data_dir) = temp_data_dir(); + let db = GatewayDb::open(&data_dir).await.expect("open gateway database"); + let first = db.connect().await.expect("connect first feature"); + let second = db.connect().await.expect("connect second feature"); + + let tx = first + .transaction_with_behavior(TransactionBehavior::Deferred) + .await + .expect("begin a transaction on the first connection"); + let second_tx = second + .transaction_with_behavior(TransactionBehavior::Deferred) + .await + .expect("the second connection is not inside the first one's transaction"); + + drop(second_tx); + drop(tx); + } + + #[test] + fn copy_is_flushed_and_renamed_into_place() { + let (_temp_dir, data_dir) = temp_data_dir(); + let from = data_dir.join("gateway.db.tmp"); + let to = data_dir.join(FILE_NAME); + std::fs::write(&from, b"copy").expect("write the copy"); + + flush_file(&from).expect("flush the copy"); + rename_durably(&from, &to).expect("rename the copy"); + + assert!(!from.exists()); + assert_eq!(std::fs::read(&to).expect("read the target"), b"copy"); + remove_file_if_exists(&from).expect("a missing file is not an error"); + } +} diff --git a/devolutions-gateway/Cargo.toml b/devolutions-gateway/Cargo.toml index 6d68ec6e8..73db5a544 100644 --- a/devolutions-gateway/Cargo.toml +++ b/devolutions-gateway/Cargo.toml @@ -36,7 +36,8 @@ network-monitor.path = "../crates/network-monitor" sysevent.path = "../crates/sysevent" sysevent-codes.path = "../crates/sysevent-codes" agent-tunnel = { path = "../crates/agent-tunnel" } -agent-tunnel-libsql = { path = "../crates/agent-tunnel-libsql" } +agent-tunnel-libsql.path = "../crates/agent-tunnel-libsql" +gateway-db.path = "../crates/gateway-db" ironrdp-pdu = { version = "0.9", features = ["std"] } ironrdp-core = { version = "0.2", features = ["std"] } ironrdp-rdcleanpath = "0.2" diff --git a/devolutions-gateway/src/service.rs b/devolutions-gateway/src/service.rs index 97b733f17..258e15d23 100644 --- a/devolutions-gateway/src/service.rs +++ b/devolutions-gateway/src/service.rs @@ -275,13 +275,27 @@ async fn spawn_tasks(conf_handle: ConfHandle) -> anyhow::Result { ); let monitoring_state = Arc::new(network_monitor::State::new(Arc::new(filesystem_monitor_config_cache))?); + let data_dir = config::get_data_dir(); + agent_tunnel_libsql::import_2026_3_0_database(&data_dir) + .await + .context("failed to import the 2026.3.0 agent tunnel database")?; + let gateway_db = gateway_db::GatewayDb::open(&data_dir) + .await + .context("failed to open the gateway database")?; + // Initialize the agent tunnel when enabled. let agent_tunnel_handle = if conf.agent_tunnel.enabled { - let data_dir = config::get_data_dir(); let hostname = &conf.hostname; - let authorization_database = data_dir.join("agent_tunnel.db"); - let ca_manager = if authorization_database.exists() { + let agent_tunnel_conn = gateway_db + .connect() + .await + .context("failed to connect the agent tunnel to the gateway database")?; + let agents_have_ca = agent_tunnel_libsql::LibSqlAgentAuthorizationStore::bound_ca(&agent_tunnel_conn) + .await + .context("failed to read the agent tunnel CA identity")? + .is_some(); + let ca_manager = if agents_have_ca { agent_tunnel::cert::CaManager::load(&data_dir) } else { agent_tunnel::cert::CaManager::load_or_generate(&data_dir) @@ -291,7 +305,7 @@ async fn spawn_tasks(conf_handle: ConfHandle) -> anyhow::Result { .ca_spki_sha256() .context("failed to identify agent tunnel CA")?; let authorization_store = - agent_tunnel_libsql::LibSqlAgentAuthorizationStore::open(authorization_database.as_str(), ca_spki_sha256) + agent_tunnel_libsql::LibSqlAgentAuthorizationStore::open(agent_tunnel_conn, ca_spki_sha256) .await .context("failed to initialize Agent authorization database")?; let authorization_store: agent_tunnel::authorization::DynAgentAuthorizationStore = diff --git a/testsuite/Cargo.toml b/testsuite/Cargo.toml index a40691073..b6c687fc1 100644 --- a/testsuite/Cargo.toml +++ b/testsuite/Cargo.toml @@ -33,13 +33,14 @@ tokio-tungstenite = { version = "0.29", features = ["rustls-tls-native-roots"] } [dev-dependencies] agent-sysevent-codes.path = "../crates/agent-sysevent-codes" agent-tunnel = { path = "../crates/agent-tunnel", features = ["test-utils"] } -agent-tunnel-libsql = { path = "../crates/agent-tunnel-libsql" } agent-tunnel-proto = { path = "../crates/agent-tunnel-proto", features = ["serde"] } base64 = "0.23" camino = "1" devolutions-gateway-task = { path = "../crates/devolutions-gateway-task" } devolutions-gateway = { path = "../devolutions-gateway" } futures-util = "0.3" +agent-tunnel-libsql.path = "../crates/agent-tunnel-libsql" +gateway-db.path = "../crates/gateway-db" ipnetwork = "0.20" jmux-proto = { path = "../crates/jmux-proto" } libsql = { version = "0.9", default-features = false, features = ["core"] } diff --git a/testsuite/tests/agent_tunnel/common.rs b/testsuite/tests/agent_tunnel/common.rs index ca4af024e..e87b76d33 100644 --- a/testsuite/tests/agent_tunnel/common.rs +++ b/testsuite/tests/agent_tunnel/common.rs @@ -175,7 +175,13 @@ pub(super) async fn bind_test_listener() -> TestListener { let data_dir = Utf8PathBuf::from_path_buf(temp_dir.path().to_path_buf()).expect("use utf-8 temporary path"); let ca_manager = CaManager::load_or_generate(&data_dir).expect("generate test ca"); let ca_spki_sha256 = ca_manager.ca_spki_sha256().expect("hash test CA public key"); - let authorization_store = agent_tunnel_libsql::LibSqlAgentAuthorizationStore::open(":memory:", ca_spki_sha256) + let gateway_db = gateway_db::GatewayDb::open_path(":memory:") + .await + .expect("open test gateway database") + .connect() + .await + .expect("connect to test gateway database"); + let authorization_store = agent_tunnel_libsql::LibSqlAgentAuthorizationStore::open(gateway_db, ca_spki_sha256) .await .expect("open test Agent authorization store"); let listen_addr: SocketAddr = "127.0.0.1:0".parse().expect("parse listener address");