fix!: centralize entity name validation for remote APIs (#4490)

Centralize the entity name validation for tables, views, functions,
namespaces, secrets, and databases. This helps in several ways. One is
that percent-encoding is no longer needed in most cases (the exception
is database names, which can contain slashes). Another good thing is
that LLMs looking at this repository can now understand what the naming
rules are, since they will be the same here as on the server side.
Finally, this will help with the upcoming authZ code, since it also
needs to serialize these names, and needs to know that they don't
contain (among other things) colons.

rust/lancedb/src/connection.rs

@@ -38,10 +38,7 @@ use crate::remote::{
     },
 };
 use crate::secrets::SecretInfo;
-use crate::utils::{
-    validate_namespace, validate_secret_component, validate_secret_reference,
-    validate_view_reference,
-};
+use crate::utils::{validate_namespace, validate_secret_reference, validate_view_reference};
 use crate::view::ViewDescription;
 use lance::io::ObjectStoreParams;
 pub use lance_file::version::LanceFileVersion;
@@ -811,9 +808,7 @@ impl Connection {
             let database = database.clone();
             let namespace_path = namespace_path.clone();
             async move {
-                for segment in &namespace_path {
-                    validate_secret_component("Secret namespace path segment", segment)?;
-                }
+                validate_namespace(&namespace_path)?;
                 database.list_secrets(&namespace_path, options).await
             }
         })

rust/lancedb/src/remote/catalog.rs

@@ -19,6 +19,7 @@ use crate::catalog::{
     ListDatabasesResponse,
 };
 use crate::database::Database;
+use crate::utils::validate_database_name;
 use crate::{Error, Result};
 
 /// Authentication and client settings shared by a catalog and its databases.
@@ -154,29 +155,6 @@ impl RemoteCatalog {
         })
     }
 
-    fn validate_name(&self, name: &str) -> Result<()> {
-        let delimiter = self
-            .options
-            .client_config
-            .id_delimiter
-            .as_deref()
-            .unwrap_or("$");
-        if name.is_empty()
-            || name.trim() != name
-            || !name.is_ascii()
-            || name.chars().any(char::is_control)
-            || name.contains(delimiter)
-            || matches!(name, "." | "..")
-        {
-            return Err(Error::InvalidInput {
-                message: format!(
-                    "Invalid database name '{name}': expected a nonempty ASCII name without surrounding whitespace, control characters, or namespace delimiter '{delimiter}'"
-                ),
-            });
-        }
-        Ok(())
-    }
-
     fn database(&self, name: &str) -> Result<Arc<dyn Database>> {
         Ok(Arc::new(RemoteDatabase::for_catalog(
             &self.endpoint,
@@ -203,7 +181,7 @@ impl Catalog for RemoteCatalog {
     }
 
     async fn create_database(&self, request: CreateDatabaseRequest) -> Result<Arc<dyn Database>> {
-        self.validate_name(&request.name)?;
+        validate_database_name(&request.name)?;
         self.root
             .create_namespace(CreateNamespaceRequest {
                 id: Some(vec![request.name.clone()]),
@@ -231,7 +209,7 @@ impl Catalog for RemoteCatalog {
     }
 
     async fn drop_database(&self, request: DropDatabaseRequest) -> Result<()> {
-        self.validate_name(&request.name)?;
+        validate_database_name(&request.name)?;
         let result = self
             .root
             .drop_namespace(DropNamespaceRequest {
@@ -286,7 +264,7 @@ impl Catalog for RemoteCatalog {
     }
 
     async fn connect_database(&self, name: &str) -> Result<Arc<dyn Database>> {
-        self.validate_name(name)?;
+        validate_database_name(name)?;
         self.root
             .describe_namespace(DescribeNamespaceRequest {
                 id: Some(vec![name.into()]),
@@ -676,7 +654,24 @@ mod tests {
         }
         let catalog =
             RemoteCatalog::try_new("http://127.0.0.1:1", RemoteCatalogOptions::default()).unwrap();
-        for name in ["", "a$b", "\r\ninjected", "..", "café", " padded "] {
+        for name in [
+            "",
+            "/",
+            "//",
+            "a$b",
+            "\r\ninjected",
+            "..",
+            "café",
+            " padded ",
+            "a b",
+            "a?b",
+            "/db",
+            "db/",
+            "a//b",
+            "a/../b",
+            "a/./b",
+            &"a".repeat(65),
+        ] {
             assert!(matches!(
                 catalog.create_database(name.into()).await,
                 Err(Error::InvalidInput { .. })

rust/lancedb/src/remote/client.rs

@@ -12,6 +12,7 @@ use std::{collections::HashMap, future::Future, str::FromStr, sync::Arc, time::D
 use crate::error::{Error, Result};
 use crate::remote::db::RemoteOptions;
 use crate::remote::retry::{ResolvedRetryConfig, RetryCounter};
+use crate::utils::validate_database_name;
 
 const REQUEST_ID_HEADER: HeaderName = HeaderName::from_static("x-request-id");
 
@@ -383,6 +384,7 @@ pub fn parse_db_url(db_url: &str) -> Result<ParsedDbUrl> {
             message: format!("Invalid encoded database name: {err}"),
         })?
         .into_owned();
+    validate_database_name(&db_name)?;
     let db_prefix = {
         let prefix = parsed_url.path().trim_start_matches('/');
         if prefix.is_empty() {
@@ -1204,6 +1206,29 @@ mod tests {
         assert_eq!(parsed.db_prefix.as_deref(), Some("prefix"));
     }
 
+    #[test]
+    fn test_parse_database_uri_validates_decoded_name() {
+        for name in [
+            "a b",
+            "a$b",
+            "a?b",
+            "café",
+            "/db",
+            "db/",
+            "a//b",
+            "a/../b",
+            "a/./b",
+            &"a".repeat(65),
+        ] {
+            let uri = format!("db://{}", urlencoding::encode(name));
+            assert!(
+                matches!(parse_db_url(&uri), Err(Error::InvalidInput { .. })),
+                "accepted {name:?}"
+            );
+        }
+        assert!(parse_db_url(&format!("db://{}", "a".repeat(64))).is_ok());
+    }
+
     #[test]
     fn test_timeout_config_default() {
         let config = TimeoutConfig::default();

rust/lancedb/src/remote/db.rs

@@ -36,7 +36,10 @@ use crate::remote::util::stream_as_body;
 use crate::secrets::SecretBinding;
 use crate::secrets::SecretInfo;
 use crate::table::BaseTable;
-use crate::utils::{reject_relative_segment, validate_table_name};
+use crate::utils::{
+    validate_database_name, validate_function_name, validate_namespace, validate_secret_name,
+    validate_table_name, validate_view_name,
+};
 use crate::view::ViewDescription;
 
 use super::client::{
@@ -230,6 +233,7 @@ pub struct RemoteDatabase<S: HttpSend = Sender> {
     /// TLS configuration for mTLS support
     tls_config: Option<super::client::TlsConfig>,
     sql_client: Option<SqlClient>,
+    is_catalog_root: bool,
 }
 
 #[derive(Clone)]
@@ -320,6 +324,9 @@ impl RemoteDatabase {
         name: Option<&str>,
         options: &super::catalog::RemoteCatalogOptions,
     ) -> Result<Self> {
+        if let Some(name) = name {
+            validate_database_name(name)?;
+        }
         let scope = super::catalog::ScopedHeaderProvider {
             provider: options.client_config.header_provider.clone(),
             database: name.map(str::to_string),
@@ -348,6 +355,7 @@ impl RemoteDatabase {
         )?;
         if name.is_none() {
             db.sql_client = None;
+            db.is_catalog_root = true;
         }
         Ok(db)
     }
@@ -422,11 +430,38 @@ impl RemoteDatabase {
             namespace_context_provider,
             tls_config: client_config.tls_config,
             sql_client: Some(sql_client),
+            is_catalog_root: false,
         })
     }
 }
 
 impl<S: HttpSend> RemoteDatabase<S> {
+    // A catalog's root namespaces are database names, which may contain slashes.
+    fn namespace_identifier(&self, namespace: &[String]) -> Result<String> {
+        match namespace.first() {
+            None => Ok(ID_DELIMITER.to_string()),
+            Some(first) => {
+                if self.is_catalog_root {
+                    // When dealing with the catalog root, the first component is a
+                    // database name. Those need to be validated differently because
+                    // they can contain slashes. Also, we need to use percent encoding
+                    // for the slashes.
+                    validate_database_name(first)?;
+                    validate_namespace(&namespace[1..])?;
+                    Ok(namespace
+                        .iter()
+                        .map(|component| urlencoding::encode(component).into_owned())
+                        .collect::<Vec<_>>()
+                        .join(ID_DELIMITER))
+                } else {
+                    // If we're not dealing with the catalog root, build the namespace identifier in the
+                    // usual way.
+                    build_namespace_identifier(namespace)
+                }
+            }
+        }
+    }
+
     /// Post a request whose body carries a credential.
     ///
     /// Shared by the create and alter verbs, which declare their own request
@@ -537,6 +572,7 @@ mod test_utils {
                 namespace_context_provider: None,
                 tls_config: None,
                 sql_client: None,
+                is_catalog_root: false,
             }
         }
 
@@ -560,6 +596,7 @@ mod test_utils {
                 namespace_context_provider,
                 tls_config: config.tls_config.clone(),
                 sql_client: None,
+                is_catalog_root: false,
             }
         }
     }
@@ -581,93 +618,60 @@ impl From<&CreateTableMode> for &'static str {
     }
 }
 
-/// The path segment addressing one object: its namespace path and its name.
-///
-/// One builder for tables, Secrets, Functions and materialized views: the
-/// identifier grammar belongs to the namespace spec, not to an object type. An
-/// empty path addresses an object with no namespace.
-///
-/// Components are checked for addressability, not a character set. The name's
-/// grammar is the caller's, so a table reports [`Error::InvalidTableName`], a
-/// Function admits names a table may not, and a catalog database carries the
-/// `/` that [`RemoteCatalog`] allows.
-///
-/// [`RemoteCatalog`]: super::catalog::RemoteCatalog
-fn build_object_identifier(what: &str, name: &str, namespace: &[String]) -> Result<String> {
-    for segment in namespace {
-        reject_unaddressable_component("namespace segment", segment)?;
+fn build_compound_identifier(name: &str, namespace: &[String]) -> Result<String> {
+    validate_namespace(namespace)?;
+    let mut identifier = namespace.join("$");
+    if !namespace.is_empty() {
+        identifier.push('$');
     }
-    reject_unaddressable_component(what, name)?;
-    Ok(join_identifier(
-        namespace.iter().map(String::as_str).chain([name]),
-    ))
+    identifier.push_str(name);
+    Ok(identifier)
 }
 
-/// What a component may not be if the join is to survive being split back
-/// apart: empty, a segment URL parsing resolves away, or the delimiter itself.
-///
-/// Each erases a boundary no encoding of the joined form recovers. `["prod",
-/// ""]` joins to `prod$`, which reads back as `["prod"]`, so a drop reaches the
-/// parent of the namespace the caller named.
-///
-/// Not a character set: per-component percent-encoding makes the wider set
-/// safe, since a `/` in a name reaches the service as `%2F`, still one
-/// segment.
-fn reject_unaddressable_component(what: &str, value: &str) -> Result<()> {
-    if value.is_empty() {
-        return Err(Error::InvalidInput {
-            message: format!(
-                "{what} must not be empty: the identifier would carry two delimiters in a row, \
-                 and splitting it back apart would name a different object"
-            ),
-        });
-    }
-    reject_relative_segment(what, value)?;
-    if value.contains(ID_DELIMITER) {
-        return Err(Error::InvalidInput {
-            message: format!(
-                "{what} '{value}' contains the identifier delimiter '{ID_DELIMITER}', so the \
-                 namespace path and the name it joins could not be told apart"
-            ),
-        });
-    }
-    Ok(())
-}
-
-/// The path segment addressing one table. A wrapper for the error type:
-/// callers match on [`Error::InvalidTableName`].
+/// Build a full path to a table. Note that we do not need to percent-encode
+/// this when using it in a URL because the validation functions ensure that
+/// only URL-safe characters are present.
 fn build_table_identifier(name: &str, namespace: &[String]) -> Result<String> {
     validate_table_name(name)?;
-    build_object_identifier("table name", name, namespace)
+    build_compound_identifier(name, namespace)
 }
 
-/// Join components into the `{id}` a route addresses: each percent-encoded,
-/// then joined by the delimiter.
-///
-/// Per component rather than over the joined string, so the delimiter stays a
-/// delimiter and nothing inside a component can end the path segment.
-///
-/// A second line, not the first: a component from the object charset is all
-/// unreserved and encodes to itself, so the route reads as the caller wrote it.
-/// It does not cover `.` and `..`, which are unreserved too and resolve away
-/// after decoding -- [`build_object_identifier`] refuses those.
-fn join_identifier<'a>(components: impl Iterator<Item = &'a str>) -> String {
-    components
-        .map(|component| urlencoding::encode(component).into_owned())
-        .collect::<Vec<_>>()
-        .join(ID_DELIMITER)
-}
-
-/// The path segment addressing one namespace.
+/// Translate an array of namespace components into a namespace string.
+/// Note that we do not need to percent-encode this when using it in a URL
+/// because validate_namespace ensures that only URL-safe characters are
+/// present.
 fn build_namespace_identifier(namespace: &[String]) -> Result<String> {
-    for segment in namespace {
-        reject_unaddressable_component("namespace segment", segment)?;
-    }
+    validate_namespace(namespace)?;
     if namespace.is_empty() {
         // According to the namespace spec, use delimiter to represent root namespace
-        return Ok(ID_DELIMITER.to_string());
+        Ok(ID_DELIMITER.to_string())
+    } else {
+        Ok(namespace.join(ID_DELIMITER))
     }
-    Ok(join_identifier(namespace.iter().map(String::as_str)))
+}
+
+/// Build a full path to a function. Note that we do not need to percent-encode
+/// this when using it in a URL because the validation functions ensure that
+/// only URL-safe characters are present.
+fn build_function_identifier(name: &str, namespace: &[String]) -> Result<String> {
+    validate_function_name(name)?;
+    build_compound_identifier(name, namespace)
+}
+
+/// Build a full path to a secret. Note that we do not need to percent-encode
+/// this when using it in a URL because the validation functions ensure that
+/// only URL-safe characters are present.
+fn build_secret_identifier(name: &str, namespace: &[String]) -> Result<String> {
+    validate_secret_name(name)?;
+    build_compound_identifier(name, namespace)
+}
+
+/// Build a full path to a view. Note that we do not need to percent-encode
+/// this when using it in a URL because the validation functions ensure that
+/// only URL-safe characters are present.
+fn build_view_identifier(name: &str, namespace: &[String]) -> Result<String> {
+    validate_view_name(name)?;
+    build_compound_identifier(name, namespace)
 }
 
 /// Build a secure cache key using length prefixes.
@@ -945,7 +949,7 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
         request: FunctionRegistrationRequest,
         namespace_path: &[String],
     ) -> Result<Job<FunctionVersion>> {
-        let function_id = build_object_identifier("Function name", &request.name, namespace_path)?;
+        let function_id = build_function_identifier(&request.name, namespace_path)?;
         let req = self
             .client
             .post(&format!("/v1/function/{function_id}/create"))
@@ -976,7 +980,7 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
         version: &str,
         namespace_path: &[String],
     ) -> Result<FunctionVersion> {
-        let function_id = build_object_identifier("Function name", name, namespace_path)?;
+        let function_id = build_function_identifier(name, namespace_path)?;
         let req = self
             .client
             .post(&format!("/v1/function/{function_id}/describe"))
@@ -1036,7 +1040,7 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
         version: &str,
         namespace_path: &[String],
     ) -> Result<(bool, Job)> {
-        let function_id = build_object_identifier("Function name", name, namespace_path)?;
+        let function_id = build_function_identifier(name, namespace_path)?;
         let req = self
             .client
             .post(&format!("/v1/function/{function_id}/drop"))
@@ -1081,7 +1085,7 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
         value: &str,
         namespace_path: &[String],
     ) -> Result<()> {
-        let secret_id = build_object_identifier("Secret name", name, namespace_path)?;
+        let secret_id = build_secret_identifier(name, namespace_path)?;
         self.post_secret_write(
             &format!("/v1/secret/{secret_id}/create"),
             &RemoteCreateSecretRequest { value },
@@ -1090,7 +1094,7 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
     }
 
     async fn alter_secret(&self, name: &str, value: &str, namespace_path: &[String]) -> Result<()> {
-        let secret_id = build_object_identifier("Secret name", name, namespace_path)?;
+        let secret_id = build_secret_identifier(name, namespace_path)?;
         self.post_secret_write(
             &format!("/v1/secret/{secret_id}/alter"),
             &RemoteAlterSecretRequest { value },
@@ -1127,7 +1131,7 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
     }
 
     async fn drop_secret(&self, name: &str, namespace_path: &[String]) -> Result<()> {
-        let secret_id = build_object_identifier("Secret name", name, namespace_path)?;
+        let secret_id = build_secret_identifier(name, namespace_path)?;
         let req = self.client.post(&format!("/v1/secret/{secret_id}/drop"));
         let (request_id, response) = self.client.send(req).await?;
         self.client.check_response(&request_id, response).await?;
@@ -1135,7 +1139,7 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
     }
 
     async fn describe_secret(&self, name: &str, namespace_path: &[String]) -> Result<SecretInfo> {
-        let secret_id = build_object_identifier("Secret name", name, namespace_path)?;
+        let secret_id = build_secret_identifier(name, namespace_path)?;
         let req = self
             .client
             .post(&format!("/v1/secret/{secret_id}/describe"));
@@ -1150,7 +1154,7 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
         query: &str,
         namespace_path: &[String],
     ) -> Result<ViewDescription> {
-        let view_id = build_object_identifier("View name", name, namespace_path)?;
+        let view_id = build_view_identifier(name, namespace_path)?;
         let req = self
             .client
             .post(&format!("/v1/view/{view_id}/create"))
@@ -1167,7 +1171,7 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
         name: &str,
         namespace_path: &[String],
     ) -> Result<ViewDescription> {
-        let view_id = build_object_identifier("View name", name, namespace_path)?;
+        let view_id = build_view_identifier(name, namespace_path)?;
         let req = self.client.post(&format!("/v1/view/{view_id}/describe"));
         let (request_id, response) = self.client.send(req).await?;
         let response = self.client.check_response(&request_id, response).await?;
@@ -1184,7 +1188,7 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
     }
 
     async fn drop_view_async(&self, name: &str, namespace_path: &[String]) -> Result<Job> {
-        let view_id = build_object_identifier("View name", name, namespace_path)?;
+        let view_id = build_view_identifier(name, namespace_path)?;
         let req = self.client.post(&format!("/v1/view/{view_id}/drop"));
         let (request_id, response) = self.client.send(req).await?;
         let response = self.client.check_response(&request_id, response).await?;
@@ -1407,7 +1411,7 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
 
     async fn list_tables(&self, request: ListTablesRequest) -> Result<ListTablesResponse> {
         let namespace_parts = request.id.as_deref().unwrap_or(&[]);
-        let namespace_id = build_namespace_identifier(namespace_parts)?;
+        let namespace_id = self.namespace_identifier(namespace_parts)?;
         let mut req = self
             .client
             .get(&format!("/v1/namespace/{}/table/list", namespace_id));
@@ -1674,7 +1678,7 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
         request: ListNamespacesRequest,
     ) -> Result<ListNamespacesResponse> {
         let namespace_parts = request.id.as_deref().unwrap_or(&[]);
-        let namespace_id = build_namespace_identifier(namespace_parts)?;
+        let namespace_id = self.namespace_identifier(namespace_parts)?;
         let mut req = self
             .client
             .get(&format!("/v1/namespace/{}/list", namespace_id));
@@ -1696,7 +1700,7 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
         request: CreateNamespaceRequest,
     ) -> Result<CreateNamespaceResponse> {
         let namespace_parts = request.id.as_deref().unwrap_or(&[]);
-        let namespace_id = build_namespace_identifier(namespace_parts)?;
+        let namespace_id = self.namespace_identifier(namespace_parts)?;
         let mut req = self
             .client
             .post(&format!("/v1/namespace/{}/create", namespace_id));
@@ -1727,7 +1731,7 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
 
     async fn drop_namespace(&self, request: DropNamespaceRequest) -> Result<DropNamespaceResponse> {
         let namespace_parts = request.id.as_deref().unwrap_or(&[]);
-        let namespace_id = build_namespace_identifier(namespace_parts)?;
+        let namespace_id = self.namespace_identifier(namespace_parts)?;
         let mut req = self
             .client
             .post(&format!("/v1/namespace/{}/drop", namespace_id));
@@ -1761,7 +1765,7 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
         request: DescribeNamespaceRequest,
     ) -> Result<DescribeNamespaceResponse> {
         let namespace_parts = request.id.as_deref().unwrap_or(&[]);
-        let namespace_id = build_namespace_identifier(namespace_parts)?;
+        let namespace_id = self.namespace_identifier(namespace_parts)?;
         let req = self
             .client
             .post(&format!("/v1/namespace/{}/describe", namespace_id))
@@ -1887,6 +1891,40 @@ mod tests {
         remote::{ARROW_STREAM_CONTENT_TYPE, ClientConfig, HeaderProvider, JSON_CONTENT_TYPE},
     };
 
+    #[tokio::test]
+    async fn test_catalog_namespace_identifier_validates_all_namespace_components() {
+        let mut db = super::RemoteDatabase::new_mock(|_| -> http::Response<String> {
+            panic!("identifier validation must not send a request")
+        });
+        db.is_catalog_root = true;
+
+        for (path, expected) in [
+            (vec![], "$"),
+            (vec!["team/search"], "team%2Fsearch"),
+            (vec!["team/search", "ns"], "team%2Fsearch$ns"),
+            (vec!["team/search", "ns", "child"], "team%2Fsearch$ns$child"),
+        ] {
+            let path = path.into_iter().map(String::from).collect::<Vec<_>>();
+            assert_eq!(db.namespace_identifier(&path).unwrap(), expected);
+        }
+
+        for path in [
+            vec!["../search"],
+            vec!["team/search", "bad/name"],
+            vec!["team/search", "ns", "bad/name"],
+            vec!["team/search", "public", "child"],
+        ] {
+            let path = path.into_iter().map(String::from).collect::<Vec<_>>();
+            assert!(
+                matches!(
+                    db.namespace_identifier(&path),
+                    Err(Error::InvalidInput { .. })
+                ),
+                "accepted {path:?}"
+            );
+        }
+    }
+
     #[test]
     fn test_cache_key_security() {
         // Test that cache keys are unique regardless of delimiter manipulation
@@ -4045,7 +4083,7 @@ mod tests {
                 .expect_err("an illegal component must be refused");
             assert!(!*reached.lock().unwrap(), "{name:?} reached the transport");
             assert!(
-                error.to_string().contains("Secret name"),
+                error.to_string().contains("secret name"),
                 "{name:?}: {error}"
             );
         }
@@ -4076,7 +4114,7 @@ mod tests {
                 .await
                 .expect_err("a dot-only name must be refused");
             assert!(
-                by_name.to_string().contains("relative path segments"),
+                by_name.to_string().contains("Illegal secret name"),
                 "{by_name}"
             );
 
@@ -4086,7 +4124,9 @@ mod tests {
                 .await
                 .expect_err("a dot-only namespace segment must be refused");
             assert!(
-                by_segment.to_string().contains("relative path segments"),
+                by_segment
+                    .to_string()
+                    .contains("Illegal namespace component"),
                 "{by_segment}"
             );
             assert!(
@@ -4165,28 +4205,20 @@ mod tests {
         }
     }
 
-    /// A segment outside the table charset still addresses one segment: the
-    /// service decides whether it may exist, and percent-encoding is what keeps
-    /// the question reaching the right route. A catalog database is named this
-    /// way.
+    /// Slashes belong to catalog database names, not table namespace segments.
     #[tokio::test]
-    async fn test_a_namespace_segment_outside_the_charset_is_encoded_not_refused() {
-        use std::sync::{Arc, Mutex};
-        let seen = Arc::new(Mutex::new(String::new()));
-        let path = seen.clone();
-        let conn = Connection::new_with_handler(move |request| {
-            *path.lock().unwrap() = request.url().path().to_string();
-            http::Response::builder().status(200).body("{}").unwrap()
+    async fn test_a_namespace_segment_outside_the_charset_is_refused() {
+        let conn = Connection::new_with_handler(|_| -> http::Response<String> {
+            panic!("invalid namespace reached the transport")
         });
-        conn.drop_table("t", &["team/search".to_string()])
-            .await
-            .unwrap();
-        assert_eq!(*seen.lock().unwrap(), "/v1/table/team%2Fsearch$t/drop/");
+        assert!(matches!(
+            conn.drop_table("t", &["team/search".to_string()]).await,
+            Err(Error::InvalidInput { .. })
+        ));
     }
 
-    /// A Function name is percent-encoded, which covers everything but the
-    /// relative segment: `..` is unreserved, so it survives encoding and is
-    /// then resolved away, posting a registration body to `/v1/create`.
+    /// Reject relative segments and their encoded forms before sending a request,
+    /// so a function name cannot change the route through URL normalization.
     #[tokio::test]
     async fn test_a_relative_segment_function_name_is_refused() {
         use std::sync::{Arc, Mutex};
@@ -4201,10 +4233,7 @@ mod tests {
                 .drop_function(name, "fv_1", &[])
                 .await
                 .expect_err("a dot-only Function name must be refused");
-            assert!(
-                error.to_string().contains("relative path segments"),
-                "{error}"
-            );
+            assert!(matches!(error, Error::InvalidInput { .. }), "{error}");
             assert!(!*reached.lock().unwrap(), "{name:?} reached the transport");
         }
     }
@@ -4585,7 +4614,7 @@ mod tests {
             };
             // A view is not a table, so the refusal says so.
             assert!(
-                error.contains("view name") || error.contains("view namespace path segment"),
+                error.contains("view name") || error.contains("namespace component"),
                 "{error}"
             );
         }

rust/lancedb/src/utils/mod.rs

@@ -19,15 +19,27 @@ use std::pin::Pin;
 use crate::error::{Error, Result};
 use datafusion_physical_plan::SendableRecordBatchStream;
 
-/// The characters any object name may contain: a table, a namespace segment, a
-/// Secret, a materialized view.
-///
-/// No positional rule on top of it -- a name may begin with `_`, `-` or `.`,
-/// as LanceDB namespaces already do. `.` and `..` are excluded separately, by
-/// [`reject_relative_segment`]: that is a property of where a name sits in a
-/// URL, not of the name. Length is the service's to bound.
-static OBJECT_NAME_REGEX: std::sync::LazyLock<regex::Regex> =
-    std::sync::LazyLock::new(|| regex::Regex::new(r"^[A-Za-z0-9_.\-]+$").unwrap());
+/// The maximum length of a table name. Table names are used in URLs and in the keys used with the
+/// object store.
+const MAX_TABLE_NAME_CHARS: usize = 128;
+
+/// The maximum length of a component of a namespace.
+const MAX_NAMESPACE_NAME_CHARS: usize = 64;
+
+/// The maximum number of namespace components that are allowed.
+const MAX_NAMESPACE_COMPONENTS: usize = 10;
+
+/// The maximum length of a view name.
+const MAX_VIEW_NAME_CHARS: usize = 128;
+
+/// The maximum length of a secret name.
+const MAX_SECRET_NAME_CHARS: usize = 128;
+
+/// The maximum length of a function name.
+const MAX_FUNCTION_NAME_CHARS: usize = 128;
+
+/// The maximum length of a database name.
+const MAX_DB_NAME_CHARS: usize = 64;
 
 pub trait PatchStoreParam {
     fn patch_with_store_wrapper(
@@ -86,129 +98,190 @@ impl PatchReadParam for ReadParams {
     }
 }
 
-/// The reason `.` and `..` are refused wherever a name becomes a path segment.
-const RELATIVE_SEGMENT_REASON: &str =
-    "'.' and '..' are read as relative path segments and cannot address an object";
-
-/// Whether URL parsing would resolve this component away rather than keep it.
-///
-/// Exactly `.` and `..`, and their percent-encoded spellings -- resolution
-/// happens after decoding, so `%2E%2E` collapses as surely as `..` does, and
-/// `drop_table("..")` would reach `/v1/drop/`. No wider than that: `...` is an
-/// ordinary segment that addresses fine.
-fn is_relative_segment(value: &str) -> bool {
-    let decoded = value.replace("%2e", ".").replace("%2E", ".");
-    decoded == "." || decoded == ".."
+/// Translate a string into something that can be safely logged. This involves removing control
+/// characters that could be used in an exploit like CVE-2003-0063.
+fn sanitize(input: &str) -> String {
+    input.chars().filter(|c| !c.is_control()).collect()
 }
 
-/// Refuse a path component that URL parsing resolves as a relative segment.
-///
-/// Reachable on its own for an identifier with no other validator: a Function
-/// name has no client-side grammar, so this is the only rule that applies.
-pub(crate) fn reject_relative_segment(what: &str, value: &str) -> Result<()> {
-    if is_relative_segment(value) {
-        return Err(Error::InvalidInput {
-            message: format!("invalid {what} '{value}': {RELATIVE_SEGMENT_REASON}"),
+pub fn validate_table_name(name: &str) -> Result<()> {
+    if name.is_empty() || name == "." || name == ".." {
+        return Err(Error::InvalidTableName {
+            name: name.to_string(),
+            reason: "Illegal table name.".to_string(),
+        });
+    }
+    if name.len() > MAX_TABLE_NAME_CHARS {
+        return Err(Error::InvalidTableName {
+            name: sanitize(name),
+            reason: format!(
+                "Table name cannot be more than {MAX_TABLE_NAME_CHARS} characters long."
+            ),
+        });
+    }
+    if !name
+        .chars()
+        .all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_' || c == '.')
+    {
+        return Err(Error::InvalidTableName {
+            name: sanitize(name),
+            reason: "Only alphanumeric characters, dashes, underscores, and\
+                periods are allowed in table names."
+                .to_string(),
         });
     }
     Ok(())
 }
 
-/// Every rule an object name obeys: non-empty, inside [`OBJECT_NAME_REGEX`],
-/// and addressable as a path segment.
-///
-/// Returns the reason rather than an [`Error`], because the error type is each
-/// API's own -- a table reports [`Error::InvalidTableName`], the rest
-/// [`Error::InvalidInput`]. Sharing the rules but not the error keeps a table,
-/// a namespace segment and a Secret from drifting apart.
-fn check_object_name(name: &str) -> std::result::Result<(), &'static str> {
-    if name.is_empty() {
-        return Err("it must not be empty");
+/// Validate that an input string contains alphanumerics, or dashes, underscores, and periods.
+fn validate_view_or_namespace_or_secret(input: &str, kind: &str, max_len: usize) -> Result<()> {
+    if input.is_empty() || input == "." || input == ".." {
+        return Err(Error::InvalidInput {
+            message: format!("Illegal {} '{}'.", kind, input),
+        });
     }
-    if !OBJECT_NAME_REGEX.is_match(name) {
-        return Err(
-            "it may contain only alphanumeric characters, underscores, hyphens and periods",
-        );
+    if input.len() > max_len {
+        return Err(Error::InvalidInput {
+            message: format!(
+                "Invalid {} '{}' is more than {} characters long.",
+                kind,
+                sanitize(input),
+                max_len
+            ),
+        });
     }
-    if is_relative_segment(name) {
-        return Err(RELATIVE_SEGMENT_REASON);
+    if !input
+        .chars()
+        .all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_' || c == '.')
+    {
+        return Err(Error::InvalidInput {
+            message: format!(
+                "Invalid {} '{}': only alphanumeric characters, \
+                         dashes, underscores, and periods are allowed.",
+                kind,
+                sanitize(input)
+            ),
+        });
     }
     Ok(())
 }
 
-/// Validate a table name.
-pub fn validate_table_name(name: &str) -> Result<()> {
-    check_object_name(name).map_err(|reason| Error::InvalidTableName {
-        name: name.to_string(),
-        reason: reason.to_string(),
-    })
-}
-
 /// Validate one component of a namespace path -- a single segment, not the
 /// whole path. [`validate_namespace`] covers a path.
-pub fn validate_namespace_name(name: &str) -> Result<()> {
-    check_object_name(name).map_err(|reason| Error::InvalidInput {
-        message: format!("invalid namespace name '{name}': {reason}"),
-    })
+pub fn validate_namespace_component(name: &str) -> Result<()> {
+    validate_view_or_namespace_or_secret(name, "namespace component", MAX_NAMESPACE_NAME_CHARS)
 }
 
-/// Validate one component of a Secret identifier: a Secret name, or one segment
-/// of the namespace path holding it.
-///
-/// The join decides identity, and the service only sees what the split
-/// produced. `"a$b"` is not a name the service accepts, but joined and split it
-/// reads as the namespace `a` and the name `b` -- a different Secret that may
-/// already exist. This is not a second opinion on the name; it is what lets the
-/// service have one.
-pub fn validate_secret_component(what: &str, value: &str) -> Result<()> {
-    check_object_name(value).map_err(|reason| Error::InvalidInput {
-        message: format!("invalid {what} '{value}': {reason}"),
-    })
+/// Validate all components of a namespace
+pub fn validate_namespace(namespace: &[String]) -> Result<()> {
+    if namespace.len() > 1 && namespace.first().map(String::as_str) == Some("public") {
+        return Err(Error::InvalidInput {
+            message: "Namespaces that begin with public cannot have multiple namespace components."
+                .to_string(),
+        });
+    }
+    if namespace.len() > MAX_NAMESPACE_COMPONENTS {
+        return Err(Error::InvalidInput {
+            message: format!(
+                "Cannot have a namespace with more than {MAX_NAMESPACE_COMPONENTS} components."
+            ),
+        });
+    }
+    for component in namespace {
+        validate_namespace_component(component)?;
+    }
+    Ok(())
+}
+
+/// Validate a secret name.
+pub fn validate_secret_name(name: &str) -> Result<()> {
+    validate_view_or_namespace_or_secret(name, "secret name", MAX_SECRET_NAME_CHARS)
 }
 
 /// Validate a Secret name and every segment of the namespace path holding it.
 pub fn validate_secret_reference(name: &str, namespace_path: &[String]) -> Result<()> {
-    for segment in namespace_path {
-        validate_secret_component("Secret namespace path segment", segment)?;
-    }
-    validate_secret_component("Secret name", name)
+    validate_namespace(namespace_path)?;
+    validate_secret_name(name)
+}
+
+pub fn validate_view_name(name: &str) -> Result<()> {
+    validate_view_or_namespace_or_secret(name, "view name", MAX_VIEW_NAME_CHARS)
 }
 
 /// Validate a view name and every segment of the namespace path holding it.
-///
-/// Worded for a view rather than deferring to [`validate_table_name`]: a view
-/// is not a table, and a caller who mistypes one should not be told their
-/// table name is invalid. The rule is the same one every object name follows,
-/// which is what lets the `$`-joined identifier split back apart.
 pub fn validate_view_reference(name: &str, namespace_path: &[String]) -> Result<()> {
-    for segment in namespace_path {
-        validate_view_component("view namespace path segment", segment)?;
+    validate_namespace(namespace_path)?;
+    validate_view_name(name)
+}
+
+/// Validate a function name. These do not allow dashes or periods.
+pub fn validate_function_name(name: &str) -> Result<()> {
+    if name.is_empty() {
+        return Err(Error::InvalidInput {
+            message: format!("Invalid function name '{}'.", name),
+        });
     }
-    validate_view_component("view name", name)
+    if name.len() > MAX_FUNCTION_NAME_CHARS {
+        return Err(Error::InvalidInput {
+            message: format!(
+                "Invalid function name '{}' is more than {} characters long.",
+                sanitize(name),
+                MAX_FUNCTION_NAME_CHARS
+            ),
+        });
+    }
+    if !name.chars().all(|c| c.is_ascii_alphanumeric() || c == '_') {
+        return Err(Error::InvalidInput {
+            message: format!(
+                "Invalid function name '{}': only alphanumeric \
+                              characters and underscores are allowed.",
+                sanitize(name)
+            ),
+        });
+    }
+    Ok(())
 }
 
-/// Validate one component of a view identifier: a view name, or one segment of
-/// the namespace path holding it.
-pub fn validate_view_component(what: &str, value: &str) -> Result<()> {
-    check_object_name(value).map_err(|reason| Error::InvalidInput {
-        message: format!("invalid {what} '{value}': {reason}"),
-    })
-}
-
-/// Validate all components of a namespace
-///
-/// Iterates through all namespace components and validates each one.
-/// Returns an error if any component is invalid.
-///
-/// # Arguments
-/// * `namespace` - The namespace components to validate
-///
-/// # Returns
-/// * `Ok(())` if all namespace components are valid
-/// * `Err(Error)` if any component is invalid
-pub fn validate_namespace(namespace: &[String]) -> Result<()> {
-    for component in namespace {
-        validate_namespace_name(component)?;
+/// Validate a database name, including slash-separated components.
+pub fn validate_database_name(name: &str) -> Result<()> {
+    if name.len() > MAX_DB_NAME_CHARS {
+        return Err(Error::InvalidInput {
+            message: format!(
+                "Invalid database name '{}' is more than {} characters long.",
+                sanitize(name),
+                MAX_DB_NAME_CHARS
+            ),
+        });
+    }
+    if !name
+        .chars()
+        .all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_' || c == '.' || c == '/')
+    {
+        return Err(Error::InvalidInput {
+            message: format!(
+                "Invalid database name '{}': only alphanumeric characters, dashes, \
+                              underscores, periods, and slashes are allowed.",
+                sanitize(name)
+            ),
+        });
+    }
+    for component in name.split('/') {
+        if component.is_empty() {
+            return Err(Error::InvalidInput {
+                message: format!(
+                    "Invalid database name '{}': invalid empty path component.",
+                    sanitize(name)
+                ),
+            });
+        }
+        if component == "." || component == ".." {
+            return Err(Error::InvalidInput {
+                message: format!(
+                    "Invalid database name '{}': illegal path component.",
+                    sanitize(name)
+                ),
+            });
+        }
     }
     Ok(())
 }
@@ -927,6 +1000,37 @@ mod tests {
         assert!(err.contains("text.embedding"));
     }
 
+    #[test]
+    fn test_validate_database_name() {
+        for name in [
+            "db",
+            "team/search",
+            "Team_1/db-2.3",
+            &"a".repeat(MAX_DB_NAME_CHARS),
+        ] {
+            assert!(validate_database_name(name).is_ok(), "rejected {name:?}");
+        }
+        for name in [
+            "",
+            "/",
+            "//",
+            ".",
+            "..",
+            "/db",
+            "db/",
+            "a//b",
+            "a/./b",
+            "a/../b",
+            "a$b",
+            "a b",
+            "a?b",
+            "café",
+            &"a".repeat(MAX_DB_NAME_CHARS + 1),
+        ] {
+            assert!(validate_database_name(name).is_err(), "accepted {name:?}");
+        }
+    }
+
     #[test]
     fn test_validate_table_name() {
         assert!(validate_table_name("my_table").is_ok());
@@ -947,27 +1051,27 @@ mod tests {
     }
 
     #[test]
-    fn test_validate_namespace_name() {
+    fn test_validate_namespace_component() {
         // Valid namespace names
-        assert!(validate_namespace_name("ns1").is_ok());
-        assert!(validate_namespace_name("namespace_123").is_ok());
-        assert!(validate_namespace_name("my-namespace").is_ok());
-        assert!(validate_namespace_name("my.namespace").is_ok());
-        assert!(validate_namespace_name("NS_1.2.3").is_ok());
-        assert!(validate_namespace_name("a").is_ok());
-        assert!(validate_namespace_name("123").is_ok());
-        assert!(validate_namespace_name("_underscore").is_ok());
-        assert!(validate_namespace_name("-hyphen").is_ok());
-        assert!(validate_namespace_name(".period").is_ok());
+        assert!(validate_namespace_component("ns1").is_ok());
+        assert!(validate_namespace_component("namespace_123").is_ok());
+        assert!(validate_namespace_component("my-namespace").is_ok());
+        assert!(validate_namespace_component("my.namespace").is_ok());
+        assert!(validate_namespace_component("NS_1.2.3").is_ok());
+        assert!(validate_namespace_component("a").is_ok());
+        assert!(validate_namespace_component("123").is_ok());
+        assert!(validate_namespace_component("_underscore").is_ok());
+        assert!(validate_namespace_component("-hyphen").is_ok());
+        assert!(validate_namespace_component(".period").is_ok());
 
         // Invalid namespace names
-        assert!(validate_namespace_name("").is_err());
-        assert!(validate_namespace_name("namespace with spaces").is_err());
-        assert!(validate_namespace_name("namespace/with/slashes").is_err());
-        assert!(validate_namespace_name("namespace\\with\\backslashes").is_err());
-        assert!(validate_namespace_name("namespace$with$delimiter").is_err());
-        assert!(validate_namespace_name("namespace@special").is_err());
-        assert!(validate_namespace_name("namespace#hash").is_err());
+        assert!(validate_namespace_component("").is_err());
+        assert!(validate_namespace_component("namespace with spaces").is_err());
+        assert!(validate_namespace_component("namespace/with/slashes").is_err());
+        assert!(validate_namespace_component("namespace\\with\\backslashes").is_err());
+        assert!(validate_namespace_component("namespace$with$delimiter").is_err());
+        assert!(validate_namespace_component("namespace@special").is_err());
+        assert!(validate_namespace_component("namespace#hash").is_err());
     }
 
     #[test]
@@ -983,6 +1087,9 @@ mod tests {
         // Empty namespace (root) is valid
         assert!(validate_namespace(&[]).is_ok());
 
+        // Valid public namespace (which is equivalent to the root namespace.)
+        assert!(validate_namespace(&["public".to_string()]).is_ok());
+
         // Invalid: contains empty component
         assert!(validate_namespace(&["ns1".to_string(), "".to_string()]).is_err());
 
@@ -994,6 +1101,9 @@ mod tests {
         assert!(validate_namespace(&["ns1".to_string(), "ns/2".to_string()]).is_err());
         assert!(validate_namespace(&["ns1".to_string(), "ns$2".to_string()]).is_err());
 
+        // Invalid: starts with public but contains other components.
+        assert!(validate_namespace(&["public".to_string(), "foo".to_string()]).is_err());
+
         // Valid: underscores, hyphens, and periods are allowed
         assert!(
             validate_namespace(&["ns_1".to_string(), "ns-2".to_string(), "ns.3".to_string()])