diff --git a/sync-server/Cargo.lock b/sync-server/Cargo.lock index 810b4b93..f143afcc 100644 --- a/sync-server/Cargo.lock +++ b/sync-server/Cargo.lock @@ -2135,8 +2135,10 @@ dependencies = [ "serde_yaml", "sha2", "sqlx", + "tempfile", "thiserror 2.0.17", "tokio", + "tower", "tower-http", "tracing", "tracing-subscriber", diff --git a/sync-server/Cargo.toml b/sync-server/Cargo.toml index a333545b..bc39c34b 100644 --- a/sync-server/Cargo.toml +++ b/sync-server/Cargo.toml @@ -36,6 +36,10 @@ ts-rs = { version = "10.1", features = ["uuid-impl", "chrono-impl", "no-serde-w base64 = "0.22.1" reconcile-text = { version = "0.8.0", features = ["serde"] } +[dev-dependencies] +tempfile = "3" +tower = { version = "0.5", features = ["util"] } + [profile.release] codegen-units = 1 lto = true diff --git a/sync-server/README.md b/sync-server/README.md index 90fc3e48..fc54337c 100644 --- a/sync-server/README.md +++ b/sync-server/README.md @@ -93,6 +93,23 @@ endpoint uses the same history, so losing the final notification is recoverable. A fresh client reads `/vault_snapshot`, applies it durably, then replays after its watermark. Cursor notifications are ephemeral and do not consume event IDs. +Cursor expiry and disconnect broadcast the remaining clients, including an empty +list when the last client leaves. + +## Vault names and database files + +Vault names and configured allowlists are trimmed and lowercased consistently. +Empty names, `.` and `..`, path separators, and control characters are rejected. +Other Unicode spellings remain distinct identities, including composed and +decomposed characters. + +Databases use `vaults/.sqlite` underneath the +configured database directory. This prevents filesystem case and Unicode aliases +from sharing a database. This is the only supported storage layout; top-level +`.sqlite` files are not loaded or migrated. Back up the entire database +directory, including the `vaults` subdirectory. + +Vault snapshots query metadata and byte lengths without loading document contents. SQLite uses WAL, `synchronous=FULL`, and `fullfsync` where supported. On Unix, startup also flushes database directory entries; the filesystem must support those durability diff --git a/sync-server/src/app_state/cursors.rs b/sync-server/src/app_state/cursors.rs index 2c57a49d..65b941ed 100644 --- a/sync-server/src/app_state/cursors.rs +++ b/sync-server/src/app_state/cursors.rs @@ -39,7 +39,9 @@ impl Cursors { ) { let mut vault_to_cursors = self.vault_to_cursors.lock().await; - let all_device_cursors = vault_to_cursors.entry(vault_id).or_insert_with(Vec::new); + let all_device_cursors = vault_to_cursors + .entry(vault_id.clone()) + .or_insert_with(Vec::new); all_device_cursors.retain(|c| &c.client_cursors.device_id != device_id); all_device_cursors.push(ClientCursorsWithTimeToLive::new(ClientCursors { @@ -48,8 +50,7 @@ impl Cursors { documents_with_cursors: document_to_cursors, })); - drop(vault_to_cursors); // Explicitly drop the lock before broadcasting to avoid deadlock - self.broadcast_cursors().await; + self.broadcast_cursors(&vault_id, all_device_cursors).await; } pub async fn get_cursors(&self, vault_id: &VaultId) -> Vec { @@ -78,31 +79,42 @@ impl Cursors { async fn remove_expired_cursors(&self) { let mut vault_to_cursors = self.vault_to_cursors.lock().await; - for (_vault_id, cursors) in vault_to_cursors.iter_mut() { + for (vault_id, cursors) in vault_to_cursors.iter_mut() { + let previous_len = cursors.len(); cursors.retain(|cursor| !cursor.is_expired(self.config.cursor_timeout)); + if cursors.len() != previous_len { + self.broadcast_cursors(vault_id, cursors).await; + } } + vault_to_cursors.retain(|_, cursors| !cursors.is_empty()); } - async fn broadcast_cursors(&self) { - let vault_to_cursors = self.vault_to_cursors.lock().await; - - for (vault_id, cursors) in vault_to_cursors.iter() { - self.broadcasts - .send( - vault_id.clone(), - Notification::Cursors(CursorPositionFromServer { - clients: cursors.iter().map(|c| c.client_cursors.clone()).collect(), - }), - ) - .await; - } + async fn broadcast_cursors(&self, vault_id: &str, cursors: &[ClientCursorsWithTimeToLive]) { + self.broadcasts + .send( + vault_id.to_owned(), + Notification::Cursors(CursorPositionFromServer { + clients: cursors.iter().map(|c| c.client_cursors.clone()).collect(), + }), + ) + .await; } pub async fn remove_cursors_of_device(&self, vault_id: &str, device_id: &str) { let mut vault_to_cursors = self.vault_to_cursors.lock().await; if let Some(cursors) = vault_to_cursors.get_mut(vault_id) { + let previous_len = cursors.len(); + cursors.retain(|c| c.client_cursors.device_id != device_id); + + if cursors.len() != previous_len { + self.broadcast_cursors(vault_id, cursors).await; + } + + if cursors.is_empty() { + vault_to_cursors.remove(vault_id); + } } } } @@ -125,3 +137,96 @@ impl ClientCursorsWithTimeToLive { self.last_updated.elapsed() > ttl } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::config::server_config::ServerConfig; + + #[tokio::test] + async fn disconnect_broadcasts_remaining_clients_and_the_final_empty_state() { + let broadcasts = Broadcasts::new(&ServerConfig::default()); + let cursors = Cursors::new(&DatabaseConfig::default(), &broadcasts); + for device in ["one", "two"] { + cursors + .update_cursors( + "vault".to_owned(), + "user".to_owned(), + &device.to_owned(), + vec![], + ) + .await; + } + let mut receiver = broadcasts.get_receiver("vault".to_owned()).await; + let mut other = broadcasts.get_receiver("other".to_owned()).await; + cursors.remove_cursors_of_device("vault", "one").await; + let Notification::Cursors(update) = receiver.try_recv().unwrap() else { + panic!("expected cursors") + }; + assert_eq!(update.clients.len(), 1); + assert_eq!(update.clients[0].device_id, "two"); + + cursors.remove_cursors_of_device("vault", "two").await; + let Notification::Cursors(update) = receiver.try_recv().unwrap() else { + panic!("expected cursors") + }; + assert!(update.clients.is_empty()); + assert!(other.try_recv().is_err()); + cursors.remove_cursors_of_device("vault", "two").await; + assert!(receiver.try_recv().is_err()); + } + + #[tokio::test] + async fn expiration_broadcasts_only_changed_vaults_including_empty_ones() { + let broadcasts = Broadcasts::new(&ServerConfig::default()); + let config = DatabaseConfig::default(); + let cursors = Cursors::new(&config, &broadcasts); + for (vault, device) in [("vault", "expired"), ("vault", "live"), ("other", "live")] { + cursors + .update_cursors( + vault.to_owned(), + "user".to_owned(), + &device.to_owned(), + vec![], + ) + .await; + } + cursors + .vault_to_cursors + .lock() + .await + .get_mut("vault") + .unwrap()[0] + .last_updated = std::time::Instant::now() + .checked_sub(config.cursor_timeout + Duration::from_secs(1)) + .unwrap(); + let mut receiver = broadcasts.get_receiver("vault".to_owned()).await; + let mut other = broadcasts.get_receiver("other".to_owned()).await; + cursors.remove_expired_cursors().await; + let Notification::Cursors(update) = receiver.try_recv().unwrap() else { + panic!("expected cursors") + }; + assert_eq!(update.clients.len(), 1); + assert_eq!(update.clients[0].device_id, "live"); + assert!(other.try_recv().is_err()); + + cursors + .vault_to_cursors + .lock() + .await + .get_mut("vault") + .unwrap()[0] + .last_updated = std::time::Instant::now() + .checked_sub(config.cursor_timeout + Duration::from_secs(1)) + .unwrap(); + cursors.remove_expired_cursors().await; + let Notification::Cursors(update) = receiver.try_recv().unwrap() else { + panic!("expected cursors") + }; + assert!(update.clients.is_empty()); + assert!(cursors.get_cursors(&"vault".to_owned()).await.is_empty()); + cursors.remove_expired_cursors().await; + assert!(receiver.try_recv().is_err()); + assert!(other.try_recv().is_err()); + } +} diff --git a/sync-server/src/app_state/database.rs b/sync-server/src/app_state/database.rs index 3bd5e9c0..c0199821 100644 --- a/sync-server/src/app_state/database.rs +++ b/sync-server/src/app_state/database.rs @@ -1,9 +1,14 @@ use core::time::Duration; -use std::{collections::HashMap, sync::Arc}; +use std::{ + collections::HashMap, + path::{Path, PathBuf}, + sync::Arc, +}; use anyhow::{Context as _, Result}; use log::info; use models::VaultId; +use sha2::{Digest, Sha256}; use sqlx::{ConnectOptions, sqlite::SqliteConnectOptions}; use sqlx::{Pool, Sqlite, sqlite::SqlitePoolOptions}; use tokio::sync::Mutex; @@ -14,7 +19,15 @@ pub mod models; mod mutations; mod queries; -use crate::config::database_config::DatabaseConfig; +#[cfg(test)] +mod tests; + +use crate::{ + config::database_config::DatabaseConfig, + utils::normalize_vault_id::{normalize_string, validate_vault_id}, +}; + +const VAULT_DATABASE_DIRECTORY: &str = "vaults"; #[derive(Clone)] struct PoolWithTimestamp { @@ -41,7 +54,10 @@ pub type Transaction<'a> = sqlx::Transaction<'a, Sqlite>; impl Database { pub async fn try_new(config: &DatabaseConfig) -> Result { - tokio::fs::create_dir_all(&config.databases_directory_path) + let vault_directory = config + .databases_directory_path + .join(VAULT_DATABASE_DIRECTORY); + tokio::fs::create_dir_all(&vault_directory) .await .with_context(|| { format!( @@ -50,34 +66,25 @@ impl Database { ) })?; - let mut connection_pools = std::collections::HashMap::new(); - info!("Applying pending database migrations"); - let mut entries = tokio::fs::read_dir(&config.databases_directory_path).await?; + + // Filenames encode only a digest, so their vault names cannot be + // recovered here. Check migrations now and open pools lazily by name. + let mut entries = tokio::fs::read_dir(&vault_directory).await?; while let Some(entry) = entries.next_entry().await? { - if !entry.file_name().to_string_lossy().ends_with(".sqlite") { - continue; + if entry.file_type().await?.is_file() + && entry.path().extension().is_some_and(|ext| ext == "sqlite") + { + Self::open_database(config, &entry.path()) + .await? + .close() + .await; } - - let vault: VaultId = entry - .file_name() - .to_string_lossy() - .trim_end_matches(".sqlite") - .to_owned(); - - let pool = Self::create_vault_database(config, &vault).await?; - connection_pools.insert( - vault.clone(), - PoolWithTimestamp { - pool, - last_accessed: Instant::now(), - }, - ); } let database = Self { config: config.clone(), - connection_pools: Arc::new(Mutex::new(connection_pools)), + connection_pools: Arc::default(), }; // Start background task to cleanup idle connection pools @@ -86,16 +93,21 @@ impl Database { Ok(database) } - async fn create_vault_database( - config: &DatabaseConfig, - vault: &VaultId, - ) -> Result> { - let file_name = config + fn database_path(config: &DatabaseConfig, vault: &str) -> PathBuf { + // Hash the normalized vault name into a fixed-length ASCII filename so + // filesystem case/Unicode aliasing (e.g. composed vs. decomposed é on + // macOS) cannot make separately authorized vaults share a database. + // This is only a storage filename; the user-facing vault name is unchanged. + let digest = Sha256::digest(vault.as_bytes()); + config .databases_directory_path - .join(format!("{vault}.sqlite")); + .join(VAULT_DATABASE_DIRECTORY) + .join(format!("{digest:x}.sqlite")) + } + async fn open_database(config: &DatabaseConfig, file_name: &Path) -> Result> { let connection_options = SqliteConnectOptions::new() - .filename(file_name.clone()) + .filename(file_name) .create_if_missing(true) .auto_vacuum(sqlx::sqlite::SqliteAutoVacuum::Full) .busy_timeout(Duration::from_secs(3600)) @@ -126,10 +138,13 @@ impl Database { } async fn get_connection_pool(&self, vault: &VaultId) -> Result> { + let vault = normalize_string(vault); + validate_vault_id(&vault)?; let mut pools = self.connection_pools.lock().await; - if !pools.contains_key(vault) { - let pool = Self::create_vault_database(&self.config, vault).await?; + if !pools.contains_key(&vault) { + let file_name = Self::database_path(&self.config, &vault); + let pool = Self::open_database(&self.config, &file_name).await?; pools.insert( vault.clone(), PoolWithTimestamp { @@ -140,7 +155,7 @@ impl Database { } let pool_with_timestamp = pools - .get_mut(vault) + .get_mut(&vault) .expect("Pool was just inserted or already exists"); // Update last accessed time diff --git a/sync-server/src/app_state/database/models.rs b/sync-server/src/app_state/database/models.rs index 81cb6343..57c4cd94 100644 --- a/sync-server/src/app_state/database/models.rs +++ b/sync-server/src/app_state/database/models.rs @@ -22,16 +22,18 @@ pub struct StoredDocumentVersion { pub device_id: DeviceId, } -#[derive(TS, Debug, Clone, Serialize, Deserialize)] +#[derive(TS, Debug, Clone, Serialize, Deserialize, sqlx::FromRow)] #[serde(rename_all = "camelCase")] #[ts(export)] pub struct DocumentVersionWithoutContent { #[ts(as = "f64")] pub vault_update_id: VaultUpdateId, + #[sqlx(try_from = "uuid::fmt::Hyphenated")] pub document_id: DocumentId, pub updated_date: DateTime, pub user_id: UserId, pub device_id: DeviceId, + #[sqlx(try_from = "i64")] pub content_size: usize, } diff --git a/sync-server/src/app_state/database/queries.rs b/sync-server/src/app_state/database/queries.rs index f9225ef4..2134a58a 100644 --- a/sync-server/src/app_state/database/queries.rs +++ b/sync-server/src/app_state/database/queries.rs @@ -1,8 +1,8 @@ use super::{ Database, Transaction, models::{ - DocumentId, EventBatch, FileManifest, StoredDocumentVersion, VaultId, VaultSnapshot, - VaultUpdateId, + DocumentId, DocumentVersionWithoutContent, EventBatch, FileManifest, StoredDocumentVersion, + VaultId, VaultSnapshot, VaultUpdateId, }, }; use anyhow::{Result, ensure}; @@ -99,8 +99,10 @@ impl Database { let head_event_id = Self::latest_event_id(&mut tx).await?; let file_manifest = Self::current_file_manifest(&mut tx).await?; - let versions = sqlx::query_as::<_, StoredDocumentVersion>( - "SELECT d.* FROM latest_document_versions d + let documents = sqlx::query_as::<_, DocumentVersionWithoutContent>( + "SELECT d.vault_update_id, d.document_id, d.updated_date, d.user_id, d.device_id, + length(d.content) AS content_size + FROM latest_document_versions d JOIN file_manifest_entries m ON m.document_id = d.document_id WHERE m.file_manifest_id = ? ORDER BY d.vault_update_id", ) @@ -108,8 +110,6 @@ impl Database { .fetch_all(&mut *tx) .await?; - let documents = versions.into_iter().map(Into::into).collect(); - Ok(VaultSnapshot { head_event_id, file_manifest, diff --git a/sync-server/src/app_state/database/tests.rs b/sync-server/src/app_state/database/tests.rs new file mode 100644 index 00000000..3181384a --- /dev/null +++ b/sync-server/src/app_state/database/tests.rs @@ -0,0 +1,160 @@ +use super::*; +use models::{DocumentId, EventRecord, FileManifest, StoredDocumentVersion, VaultEvent}; + +fn config(directory: &tempfile::TempDir) -> DatabaseConfig { + DatabaseConfig { + databases_directory_path: directory.path().to_owned(), + ..Default::default() + } +} + +async fn insert_content( + database: &Database, + vault: &str, + document_id: DocumentId, + content: &[u8], +) -> StoredDocumentVersion { + let mut tx = database + .create_write_transaction(&vault.to_owned()) + .await + .unwrap(); + let request_id = uuid::Uuid::new_v4(); + let version = StoredDocumentVersion { + vault_update_id: Database::allocate_event(&mut tx, request_id, b"test") + .await + .unwrap(), + document_id, + updated_date: chrono::Utc::now(), + content: content.to_vec(), + user_id: "user".to_owned(), + device_id: "device".to_owned(), + }; + Database::insert_document_version(&mut tx, &version) + .await + .unwrap(); + Database::write_event( + &mut tx, + &EventRecord { + event_id: version.vault_update_id, + request_id, + event: VaultEvent::Content { + document: (&version).into(), + }, + }, + ) + .await + .unwrap(); + tx.commit().await.unwrap(); + version +} + +#[tokio::test] +async fn unicode_vaults_remain_isolated_after_reopening() { + let directory = tempfile::tempdir().unwrap(); + let config = config(&directory); + let database = Database::try_new(&config).await.unwrap(); + let document = uuid::Uuid::new_v4(); + let names = ["café", "cafe\u{301}", "σ", "ς"]; + + for name in names { + assert!( + database + .get_latest_document_version(&name.to_owned(), &document, None) + .await + .unwrap() + .is_none() + ); + insert_content(&database, name, document, name.as_bytes()).await; + } + + let reopened = Database::try_new(&config).await.unwrap(); + for name in names { + let version = reopened + .get_latest_document_version(&name.to_owned(), &document, None) + .await + .unwrap() + .unwrap(); + assert_eq!(version.content, name.as_bytes()); + } +} + +#[tokio::test] +async fn rejects_invalid_vault_names_at_the_database_boundary() { + let directory = tempfile::tempdir().unwrap(); + let config = DatabaseConfig { + databases_directory_path: directory.path().join("databases"), + ..Default::default() + }; + let database = Database::try_new(&config).await.unwrap(); + for name in [ + "", + " ", + ".", + "..", + "../escaped", + "/absolute", + "a/b", + "a\\b", + "a\0b", + ] { + assert!( + database.vault_snapshot(&name.to_owned()).await.is_err(), + "accepted {name:?}" + ); + } + assert!(!directory.path().join("escaped.sqlite").exists()); + assert!(database.connection_pools.lock().await.is_empty()); +} + +#[tokio::test] +async fn snapshot_returns_latest_referenced_metadata_and_byte_lengths() { + let directory = tempfile::tempdir().unwrap(); + let database = Database::try_new(&config(&directory)).await.unwrap(); + let vault = "snapshot".to_owned(); + let text_id = uuid::Uuid::new_v4(); + let binary_id = uuid::Uuid::new_v4(); + insert_content(&database, &vault, text_id, b"old").await; + let binary = insert_content(&database, &vault, binary_id, &[0, 255, 0, 128]).await; + let text = insert_content(&database, &vault, text_id, "café 🌍".as_bytes()).await; + insert_content(&database, &vault, uuid::Uuid::new_v4(), b"unreferenced").await; + + let mut tx = database.create_write_transaction(&vault).await.unwrap(); + let request_id = uuid::Uuid::new_v4(); + let manifest = FileManifest { + file_manifest_id: Database::allocate_event(&mut tx, request_id, b"manifest") + .await + .unwrap(), + entries: [ + (text_id, "text.md".to_owned()), + (binary_id, "binary.dat".to_owned()), + ] + .into(), + }; + Database::insert_file_manifest(&mut tx, &manifest) + .await + .unwrap(); + Database::write_event( + &mut tx, + &EventRecord { + event_id: manifest.file_manifest_id, + request_id, + event: VaultEvent::FileManifest { + file_manifest: manifest.clone(), + }, + }, + ) + .await + .unwrap(); + tx.commit().await.unwrap(); + + let snapshot = database.vault_snapshot(&vault).await.unwrap(); + assert_eq!(snapshot.head_event_id, manifest.file_manifest_id); + assert_eq!(snapshot.file_manifest.entries, manifest.entries); + assert_eq!(snapshot.documents.len(), 2); + for (actual, expected) in snapshot.documents.iter().zip([binary, text]) { + assert_eq!( + serde_json::to_value(actual).unwrap(), + serde_json::to_value(models::DocumentVersionWithoutContent::from(expected)).unwrap() + ); + } +} diff --git a/sync-server/src/config/server_config.rs b/sync-server/src/config/server_config.rs index 4a9da0f4..59d1480f 100644 --- a/sync-server/src/config/server_config.rs +++ b/sync-server/src/config/server_config.rs @@ -7,7 +7,7 @@ use crate::consts::{ DEFAULT_MERGEABLE_FILE_EXTENSIONS, DEFAULT_PORT, DEFAULT_RESPONSE_TIMEOUT_SECONDS, }; -#[derive(Debug, Deserialize, Serialize, Clone, Default)] +#[derive(Debug, Deserialize, Serialize, Clone)] pub struct ServerConfig { #[serde(default = "default_host")] pub host: String, @@ -28,6 +28,19 @@ pub struct ServerConfig { pub mergeable_file_extensions: Vec, } +impl Default for ServerConfig { + fn default() -> Self { + Self { + host: default_host(), + port: default_port(), + max_body_size_mb: default_max_body_size_mb(), + max_clients_per_vault: default_max_clients_per_vault(), + response_timeout: default_response_timeout(), + mergeable_file_extensions: default_mergeable_file_extensions(), + } + } +} + fn default_host() -> String { debug!("Using default server host: {DEFAULT_HOST}"); DEFAULT_HOST.to_owned() @@ -60,3 +73,36 @@ fn default_mergeable_file_extensions() -> Vec { .map(|s| (*s).to_owned()) .collect() } + +#[cfg(test)] +mod tests { + use super::*; + use crate::config::Config; + + #[tokio::test] + async fn fresh_and_omitted_server_configs_use_the_documented_defaults() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("config.yml"); + let fresh = Config::read_or_create(&path).await.unwrap(); + let reread = Config::load_from_file(&path).await.unwrap(); + let omitted: Config = serde_yaml::from_str("{}").unwrap(); + let explicit: Config = serde_yaml::from_str("server: {}").unwrap(); + for config in [fresh, reread, omitted, explicit] { + assert_eq!(config.server.host, DEFAULT_HOST); + assert_eq!(config.server.port, DEFAULT_PORT); + assert_eq!(config.server.max_body_size_mb, DEFAULT_MAX_BODY_SIZE_MB); + assert_eq!( + config.server.max_clients_per_vault, + DEFAULT_MAX_CLIENTS_PER_VAULT + ); + assert_eq!( + config.server.response_timeout, + DEFAULT_RESPONSE_TIMEOUT_SECONDS + ); + assert_eq!( + config.server.mergeable_file_extensions, + DEFAULT_MERGEABLE_FILE_EXTENSIONS + ); + } + } +} diff --git a/sync-server/src/config/user_config.rs b/sync-server/src/config/user_config.rs index 8b2537f0..0bb3379b 100644 --- a/sync-server/src/config/user_config.rs +++ b/sync-server/src/config/user_config.rs @@ -2,7 +2,10 @@ use bimap::BiHashMap; use rand::{Rng, distr::Alphanumeric, rng}; use serde::{Deserialize, Deserializer, Serialize, de::Error}; -use crate::app_state::database::models::VaultId; +use crate::{ + app_state::database::models::VaultId, + utils::normalize_vault_id::{normalize_string, validate_vault_id}, +}; #[derive(Debug, Deserialize, Serialize, Clone)] pub struct UserConfig { @@ -71,9 +74,23 @@ pub enum VaultAccess { #[derive(Debug, Deserialize, Serialize, Clone, Default)] pub struct AllowListedVaults { + #[serde(deserialize_with = "deserialize_allowed_vaults")] pub allowed: Vec, } +fn deserialize_allowed_vaults<'de, D: Deserializer<'de>>( + deserializer: D, +) -> Result, D::Error> { + Vec::::deserialize(deserializer)? + .into_iter() + .map(|name| { + let vault = normalize_string(&name); + validate_vault_id(&vault).map_err(D::Error::custom)?; + Ok(vault) + }) + .collect() +} + fn default_users() -> Vec { vec![User { name: "admin".to_owned(), @@ -95,6 +112,24 @@ mod tests { use super::*; + #[test] + fn allowlists_normalize_case_and_whitespace_but_preserve_unicode_identity() { + let list: AllowListedVaults = serde_json::from_value(json!({ + "allowed": [" MyVault ", "CAFÉ", "CAFE\u{301}"] + })) + .unwrap(); + assert_eq!(list.allowed, ["myvault", "café", "cafe\u{301}"]); + } + + #[test] + fn allowlists_reject_invalid_vault_names() { + for name in ["", " ", ".", "..", "../escaped", "a/b", "a\\b", "a\0b"] { + assert!( + serde_json::from_value::(json!({"allowed": [name]})).is_err() + ); + } + } + #[test] fn test_validate_users_unique_names_and_tokens() { let config_json = json!({ diff --git a/sync-server/src/server.rs b/sync-server/src/server.rs index 27ff1155..28137746 100644 --- a/sync-server/src/server.rs +++ b/sync-server/src/server.rs @@ -4,6 +4,9 @@ mod endpoints; pub(crate) mod requests; pub(crate) mod responses; +#[cfg(test)] +mod tests; + use anyhow::{Context as _, Result}; use auth::auth_middleware; use axum::{ diff --git a/sync-server/src/server/auth.rs b/sync-server/src/server/auth.rs index e56f4acc..c601ef54 100644 --- a/sync-server/src/server/auth.rs +++ b/sync-server/src/server/auth.rs @@ -14,8 +14,8 @@ use log::info; use crate::{ app_state::{AppState, database::models::VaultId}, config::user_config::{AllowListedVaults, User, VaultAccess}, - errors::{SyncServerError, permission_denied_error, unauthenticated_error}, - utils::normalize::normalize_string, + errors::{SyncServerError, client_error, permission_denied_error, unauthenticated_error}, + utils::normalize_vault_id::{normalize_string, validate_vault_id}, }; pub async fn auth_middleware( @@ -40,6 +40,7 @@ pub async fn auth_middleware( } pub fn auth(state: &AppState, token: &str, vault_id: &VaultId) -> Result { + validate_vault_id(vault_id).map_err(client_error)?; let user = state .config .users diff --git a/sync-server/src/server/endpoints/events.rs b/sync-server/src/server/endpoints/events.rs index ef113864..d73367df 100644 --- a/sync-server/src/server/endpoints/events.rs +++ b/sync-server/src/server/endpoints/events.rs @@ -11,12 +11,12 @@ use crate::{ database::models::{EventBatch, VaultId}, }, errors::{SyncServerError, client_error, server_error}, - utils::normalize::normalize, + utils::normalize_vault_id::normalize_vault_id, }; #[derive(Deserialize)] pub struct EventsPath { - #[serde(deserialize_with = "normalize")] + #[serde(deserialize_with = "normalize_vault_id")] vault_id: VaultId, } diff --git a/sync-server/src/server/endpoints/fetch_document_version_content.rs b/sync-server/src/server/endpoints/fetch_document_version_content.rs index 9fdd0ad8..ed9d3ae1 100644 --- a/sync-server/src/server/endpoints/fetch_document_version_content.rs +++ b/sync-server/src/server/endpoints/fetch_document_version_content.rs @@ -12,12 +12,12 @@ use crate::{ database::models::{DocumentId, VaultId, VaultUpdateId}, }, errors::{SyncServerError, not_found_error, server_error}, - utils::normalize::normalize, + utils::normalize_vault_id::normalize_vault_id, }; #[derive(Deserialize)] pub struct FetchDocumentVersionContentPathParams { - #[serde(deserialize_with = "normalize")] + #[serde(deserialize_with = "normalize_vault_id")] vault_id: VaultId, document_id: DocumentId, diff --git a/sync-server/src/server/endpoints/fetch_latest_document_version.rs b/sync-server/src/server/endpoints/fetch_latest_document_version.rs index a6b662e9..a05f632f 100644 --- a/sync-server/src/server/endpoints/fetch_latest_document_version.rs +++ b/sync-server/src/server/endpoints/fetch_latest_document_version.rs @@ -12,12 +12,12 @@ use crate::{ database::models::{DocumentId, DocumentVersion, VaultId}, }, errors::{SyncServerError, not_found_error, server_error}, - utils::normalize::normalize, + utils::normalize_vault_id::normalize_vault_id, }; #[derive(Deserialize)] pub struct FetchLatestDocumentVersionPathParams { - #[serde(deserialize_with = "normalize")] + #[serde(deserialize_with = "normalize_vault_id")] vault_id: VaultId, document_id: DocumentId, diff --git a/sync-server/src/server/endpoints/get_file_manifest.rs b/sync-server/src/server/endpoints/get_file_manifest.rs index cfafda02..d844ff41 100644 --- a/sync-server/src/server/endpoints/get_file_manifest.rs +++ b/sync-server/src/server/endpoints/get_file_manifest.rs @@ -14,12 +14,12 @@ use crate::{ }, }, errors::{SyncServerError, server_error}, - utils::normalize::normalize, + utils::normalize_vault_id::normalize_vault_id, }; #[derive(Deserialize)] pub struct GetFileManifestPath { - #[serde(deserialize_with = "normalize")] + #[serde(deserialize_with = "normalize_vault_id")] vault_id: VaultId, } diff --git a/sync-server/src/server/endpoints/ping.rs b/sync-server/src/server/endpoints/ping.rs index c42219be..8e910feb 100644 --- a/sync-server/src/server/endpoints/ping.rs +++ b/sync-server/src/server/endpoints/ping.rs @@ -14,12 +14,12 @@ use crate::{ app_state::{AppState, database::models::VaultId}, consts::SUPPORTED_API_VERSION, errors::SyncServerError, - utils::normalize::normalize, + utils::normalize_vault_id::normalize_vault_id, }; #[derive(Deserialize)] pub struct PingPathParams { - #[serde(deserialize_with = "normalize")] + #[serde(deserialize_with = "normalize_vault_id")] vault_id: VaultId, } diff --git a/sync-server/src/server/endpoints/put_file_content.rs b/sync-server/src/server/endpoints/put_file_content.rs index b09ff536..26c9cfa3 100644 --- a/sync-server/src/server/endpoints/put_file_content.rs +++ b/sync-server/src/server/endpoints/put_file_content.rs @@ -14,7 +14,7 @@ use crate::{ requests::{PushContent, PutFileContent}, responses::DocumentUpdateResponse, }, - utils::normalize::normalize, + utils::normalize_vault_id::normalize_vault_id, }; use anyhow::anyhow; use axum::{ @@ -30,7 +30,7 @@ use sha2::{Digest, Sha256}; #[derive(Deserialize)] pub struct PutFileContentPath { - #[serde(deserialize_with = "normalize")] + #[serde(deserialize_with = "normalize_vault_id")] vault_id: VaultId, document_id: DocumentId, } diff --git a/sync-server/src/server/endpoints/put_file_manifest.rs b/sync-server/src/server/endpoints/put_file_manifest.rs index db99c374..667c7827 100644 --- a/sync-server/src/server/endpoints/put_file_manifest.rs +++ b/sync-server/src/server/endpoints/put_file_manifest.rs @@ -18,12 +18,12 @@ use crate::{ }, errors::{SyncServerError, client_error, server_error}, server::{requests::PushFileManifest, responses::FileManifestUpdateResponse}, - utils::normalize::normalize, + utils::normalize_vault_id::normalize_vault_id, }; #[derive(Deserialize)] pub struct PutFileManifestPath { - #[serde(deserialize_with = "normalize")] + #[serde(deserialize_with = "normalize_vault_id")] vault_id: VaultId, } diff --git a/sync-server/src/server/endpoints/vault_snapshot.rs b/sync-server/src/server/endpoints/vault_snapshot.rs index 552517c5..1d90f9e0 100644 --- a/sync-server/src/server/endpoints/vault_snapshot.rs +++ b/sync-server/src/server/endpoints/vault_snapshot.rs @@ -11,12 +11,12 @@ use crate::{ database::models::{VaultId, VaultSnapshot}, }, errors::{SyncServerError, server_error}, - utils::normalize::normalize, + utils::normalize_vault_id::normalize_vault_id, }; #[derive(Deserialize)] pub struct VaultSnapshotPath { - #[serde(deserialize_with = "normalize")] + #[serde(deserialize_with = "normalize_vault_id")] vault_id: VaultId, } diff --git a/sync-server/src/server/endpoints/websocket.rs b/sync-server/src/server/endpoints/websocket.rs index 2be4b8f2..6d7f4a10 100644 --- a/sync-server/src/server/endpoints/websocket.rs +++ b/sync-server/src/server/endpoints/websocket.rs @@ -9,7 +9,7 @@ use crate::{ }, }, errors::{SyncServerError, client_error, server_error}, - utils::normalize::normalize, + utils::normalize_vault_id::normalize_vault_id, }; use axum::{ extract::{ @@ -26,7 +26,7 @@ use tokio::sync::broadcast::error::RecvError; #[derive(Deserialize)] pub struct WebSocketPathParams { - #[serde(deserialize_with = "normalize")] + #[serde(deserialize_with = "normalize_vault_id")] vault_id: VaultId, } diff --git a/sync-server/src/server/tests.rs b/sync-server/src/server/tests.rs new file mode 100644 index 00000000..493e5283 --- /dev/null +++ b/sync-server/src/server/tests.rs @@ -0,0 +1,121 @@ +use super::*; +use axum::{ + body::{Body, to_bytes}, + http::StatusCode, +}; +use serde_json::{Value, json}; +use tower::ServiceExt; + +async fn app(directory: &tempfile::TempDir) -> Router { + let mut config: Config = serde_json::from_value(json!({ + "users": { "user_configs": [ + { "name": "admin", "token": "admin", "vault_access": { "type": "allow_access_to_all" } }, + { "name": "mixed", "token": "mixed", "vault_access": { "type": "allow_list", "allowed": [" MyVault "] } }, + { "name": "composed", "token": "composed", "vault_access": { "type": "allow_list", "allowed": ["café"] } }, + { "name": "decomposed", "token": "decomposed", "vault_access": { "type": "allow_list", "allowed": ["cafe\u{301}"] } } + ] } + })).unwrap(); + config.database.databases_directory_path = directory.path().join("databases"); + let state = AppState::try_new(config).await.unwrap(); + get_authed_routes(state.clone()).with_state(state) +} + +async fn request(app: Router, path: &str, token: &str, body: Option) -> (StatusCode, Value) { + let response = app + .oneshot( + Request::builder() + .uri(path) + .method(if body.is_some() { + Method::PUT + } else { + Method::GET + }) + .header(http::header::AUTHORIZATION, format!("Bearer {token}")) + .header(http::header::CONTENT_TYPE, "application/json") + .header("Device-Id", "test") + .body(body.map_or_else(Body::empty, |value| Body::from(value.to_string()))) + .unwrap(), + ) + .await + .unwrap(); + let status = response.status(); + let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + ( + status, + serde_json::from_slice(&bytes).unwrap_or(Value::Null), + ) +} + +#[tokio::test] +async fn http_rejects_encoded_traversal_and_accepts_normalized_allowlists() { + let directory = tempfile::tempdir().unwrap(); + let app = app(&directory).await; + for vault in [ + "..%2Fescaped", + "%2Fabsolute", + "a%5Cb", + "%2E%2E", + "%20", + "a%00b", + ] { + let (status, _) = request( + app.clone(), + &format!("/vaults/{vault}/vault_snapshot"), + "admin", + None, + ) + .await; + assert_eq!(status, StatusCode::BAD_REQUEST, "vault: {vault}"); + } + assert!(!directory.path().join("escaped.sqlite").exists()); + for vault in ["MyVault", "myvault", "%20MYVAULT%20"] { + let (status, _) = request( + app.clone(), + &format!("/vaults/{vault}/file_manifest"), + "mixed", + None, + ) + .await; + assert_eq!(status, StatusCode::OK, "vault: {vault}"); + } + assert_eq!( + request(app, "/vaults/other/file_manifest", "mixed", None) + .await + .0, + StatusCode::FORBIDDEN + ); +} + +#[tokio::test] +async fn http_keeps_unicode_vaults_separate_across_restart() { + let directory = tempfile::tempdir().unwrap(); + let router = app(&directory).await; + let document = uuid::Uuid::new_v4(); + let composed = format!("/vaults/caf%C3%A9/documents/{document}"); + let decomposed = format!("/vaults/cafe%CC%81/documents/{document}"); + let push = json!({ + "requestId": uuid::Uuid::new_v4(), "parentVersionId": null, + "content": { "type": "Snapshot", "value": "aGVsbG8=" } + }); + assert_eq!( + request(router, &composed, "composed", Some(push)).await.0, + StatusCode::OK + ); + + let reopened = app(&directory).await; + assert_eq!( + request(reopened.clone(), &composed, "decomposed", None) + .await + .0, + StatusCode::FORBIDDEN + ); + assert_eq!( + request(reopened.clone(), &decomposed, "decomposed", None) + .await + .0, + StatusCode::NOT_FOUND + ); + let (status, version) = request(reopened, &composed, "composed", None).await; + assert_eq!(status, StatusCode::OK); + assert_eq!(version["contentBase64"], "aGVsbG8="); +} diff --git a/sync-server/src/utils.rs b/sync-server/src/utils.rs index 4798e39c..a652bec6 100644 --- a/sync-server/src/utils.rs +++ b/sync-server/src/utils.rs @@ -1,4 +1,4 @@ pub mod log_level; -pub mod normalize; +pub mod normalize_vault_id; pub mod portable_path; pub mod rotating_file_writer; diff --git a/sync-server/src/utils/normalize.rs b/sync-server/src/utils/normalize.rs deleted file mode 100644 index 6553dd25..00000000 --- a/sync-server/src/utils/normalize.rs +++ /dev/null @@ -1,13 +0,0 @@ -use serde::{Deserialize, Deserializer}; - -pub fn normalize<'de, D>(deserializer: D) -> Result -where - D: Deserializer<'de>, -{ - let s = String::deserialize(deserializer)?; - Ok(normalize_string(&s)) -} - -pub fn normalize_string(s: &str) -> String { - s.trim().to_lowercase() -} diff --git a/sync-server/src/utils/normalize_vault_id.rs b/sync-server/src/utils/normalize_vault_id.rs new file mode 100644 index 00000000..7cabf3e3 --- /dev/null +++ b/sync-server/src/utils/normalize_vault_id.rs @@ -0,0 +1,29 @@ +use anyhow::{Result, ensure}; +use serde::{Deserialize, Deserializer, de::Error}; + +pub fn normalize_vault_id<'de, D>(deserializer: D) -> Result +where + D: Deserializer<'de>, +{ + let s = String::deserialize(deserializer)?; + let vault = normalize_string(&s); + validate_vault_id(&vault).map_err(D::Error::custom)?; + Ok(vault) +} + +pub fn normalize_string(s: &str) -> String { + s.trim().to_lowercase() +} + +pub fn validate_vault_id(vault: &str) -> Result<()> { + ensure!( + !vault.is_empty() + && vault != "." + && vault != ".." + && !vault + .chars() + .any(|c| c.is_control() || matches!(c, '/' | '\\')), + "Vault names must be nonempty names without path separators, traversal components, or control characters" + ); + Ok(()) +}