Reject unsafe namespace names before filesystem access

libsql-server/src/connection/config.rs

@@ -65,9 +65,11 @@ impl Default for DatabaseConfig {
     }
 }
 
-impl From<&metadata::DatabaseConfig> for DatabaseConfig {
-    fn from(value: &metadata::DatabaseConfig) -> Self {
-        DatabaseConfig {
+impl TryFrom<&metadata::DatabaseConfig> for DatabaseConfig {
+    type Error = crate::Error;
+
+    fn try_from(value: &metadata::DatabaseConfig) -> Result<Self, Self::Error> {
+        Ok(DatabaseConfig {
             block_reads: value.block_reads,
             block_writes: value.block_writes,
             block_reason: value.block_reason.clone(),
@@ -79,16 +81,16 @@ impl From<&metadata::DatabaseConfig> for DatabaseConfig {
             allow_attach: value.allow_attach,
             max_row_size: value.max_row_size.unwrap_or_else(default_max_row_size),
             is_shared_schema: value.shared_schema.unwrap_or(false),
-            // namespace name is coming from primary, we assume it's valid
             shared_schema_name: value
                 .shared_schema_name
-                .clone()
-                .map(NamespaceName::new_unchecked),
+                .as_ref()
+                .map(|name| NamespaceName::from_string(name.clone()))
+                .transpose()?,
             durability_mode: match value.durability_mode {
                 None => DurabilityMode::default(),
                 Some(m) => DurabilityMode::from(metadata::DurabilityMode::try_from(m)),
             },
-        }
+        })
     }
 }
 
@@ -112,6 +114,22 @@ impl From<&DatabaseConfig> for metadata::DatabaseConfig {
     }
 }
 
+#[cfg(test)]
+mod namespace_tests {
+    use super::*;
+
+    #[test]
+    fn shared_schema_names_are_checked_in_replication_and_json() {
+        let wire = metadata::DatabaseConfig {
+            shared_schema_name: Some("../outside".into()),
+            ..metadata::DatabaseConfig::from(&DatabaseConfig::default())
+        };
+        assert!(DatabaseConfig::try_from(&wire).is_err());
+        let json = r#"{"block_reads":false,"block_writes":false,"block_reason":null,"max_db_pages":100,"heartbeat_url":null,"bottomless_db_id":null,"shared_schema_name":"../outside"}"#;
+        assert!(serde_json::from_str::<DatabaseConfig>(json).is_err());
+    }
+}
+
 /// Durability mode specifies the `PRAGMA SYNCHRONOUS` setting for the connection
 #[derive(PartialEq, Clone, Copy, Debug, Deserialize, Serialize, Default)]
 #[serde(rename_all = "lowercase")]

libsql-server/src/namespace/meta_store.rs

@@ -274,7 +274,7 @@ impl MetaStoreInner {
                     };
 
                     let config = match metadata::DatabaseConfig::decode(&v[..]) {
-                        Ok(c) => Arc::new(DatabaseConfig::from(&c)),
+                        Ok(c) => Arc::new(DatabaseConfig::try_from(&c)?),
                         Err(e) => {
                             tracing::warn!("unable to convert config: {}", e);
                             continue;
@@ -633,21 +633,21 @@ impl MetaStoreHandle {
         let config = match fs::read(config_path) {
             Ok(data) => {
                 let c = metadata::DatabaseConfig::decode(&data[..])?;
-                DatabaseConfig::from(&c)
+                DatabaseConfig::try_from(&c)?
             }
             Err(err) if err.kind() == io::ErrorKind::NotFound => DatabaseConfig::default(),
             Err(err) => return Err(Error::IOError(err)),
         };
 
         Ok(Self {
-            namespace: NamespaceName::new_unchecked("testmetastore"),
+            namespace: NamespaceName::from("testmetastore"),
             inner: HandleState::Internal(Arc::new(Mutex::new(Arc::new(config)))),
         })
     }
 
     pub fn internal() -> Self {
         MetaStoreHandle {
-            namespace: NamespaceName::new_unchecked("testmetastore"),
+            namespace: NamespaceName::from("testmetastore"),
             inner: HandleState::Internal(Arc::new(Mutex::new(Arc::new(DatabaseConfig::default())))),
         }
     }

libsql-server/src/namespace/name.rs

@@ -1,4 +1,7 @@
-use std::fmt;
+use std::{
+    fmt,
+    path::{Component, Path},
+};
 
 use bytes::Bytes;
 use serde::{de::Visitor, Deserialize};
@@ -45,8 +48,16 @@ impl NamespaceName {
     }
 
     fn validate(s: &str) -> crate::Result<()> {
-        if s.is_empty() {
-            tracing::warn!("invalid namespace: empty namespace");
+        // Names must be a single path component on both Unix and Windows.
+        // Keep harmless punctuation and Unicode rather than imposing an identifier alphabet.
+        let mut components = Path::new(s).components();
+        if s.is_empty()
+            || s.chars().any(|c| matches!(c, '/' | '\\' | '\0'))
+            || (cfg!(windows) && s.contains(':'))
+            || !matches!(components.next(), Some(Component::Normal(_)))
+            || components.next().is_some()
+        {
+            tracing::warn!("invalid namespace name");
             return Err(crate::error::Error::InvalidNamespace);
         }
 
@@ -67,9 +78,42 @@ impl NamespaceName {
     pub fn as_slice(&self) -> &[u8] {
         &self.0
     }
+}
 
-    pub(crate) fn new_unchecked(s: impl AsRef<str>) -> Self {
-        Self(Bytes::copy_from_slice(s.as_ref().as_bytes()))
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn only_single_safe_path_components_are_names() {
+        for name in [
+            "",
+            ".",
+            "..",
+            "../victim",
+            "a/b",
+            "a\\b",
+            "/tmp/victim",
+            "a\0b",
+        ] {
+            assert!(
+                NamespaceName::from_string(name.to_owned()).is_err(),
+                "{name:?}"
+            );
+            assert!(NamespaceName::from_bytes(Bytes::copy_from_slice(name.as_bytes())).is_err());
+            assert!(serde_json::from_str::<NamespaceName>(&format!("{name:?}")).is_err());
+        }
+        for name in ["tenant", "a..b", "hello world", "café!", "a:b"] {
+            if cfg!(windows) && name.contains(':') {
+                continue;
+            }
+            assert_eq!(
+                NamespaceName::from_string(name.to_owned())
+                    .unwrap()
+                    .as_str(),
+                name
+            );
+        }
     }
 }
 

libsql-server/src/replication/replicator_client.rs

@@ -185,7 +185,10 @@ impl ReplicatorClient for Client {
             }
 
             self.meta_store_handle
-                .store(DatabaseConfig::from(config))
+                .store(
+                    DatabaseConfig::try_from(config)
+                        .map_err(|e| Status::new(Code::InvalidArgument, e.to_string()))?,
+                )
                 .await
                 .map_err(|e| Error::Internal(e.into()))?;
 

libsql-server/src/schema/db.rs

@@ -127,7 +127,8 @@ pub(super) fn register_schema_migration_job(
     };
     let config_bytes = row.get_ref(1)?.as_blob().unwrap();
     // TODO: handle corrupted meta
-    let config = DatabaseConfig::from(&metadata::DatabaseConfig::decode(config_bytes).unwrap());
+    let config = DatabaseConfig::try_from(&metadata::DatabaseConfig::decode(config_bytes).unwrap())
+        .map_err(|e| Error::Registration(Box::new(e)))?;
     if !config.is_shared_schema {
         return Err(Error::NotASchema(schema.clone()));
     }