fix: validate filesystem catalog identifiers

bindings/c/src/identifier.rs

@@ -19,7 +19,7 @@ use std::ffi::{c_char, c_void};
 
 use paimon::catalog::Identifier;
 
-use crate::error::validate_cstr;
+use crate::error::{paimon_error, validate_cstr};
 use crate::result::paimon_result_identifier_new;
 use crate::types::paimon_identifier;
 
@@ -50,8 +50,17 @@ pub unsafe extern "C" fn paimon_identifier_new(
             }
         }
     };
+    let identifier = match Identifier::try_new(db, obj) {
+        Ok(identifier) => identifier,
+        Err(e) => {
+            return paimon_result_identifier_new {
+                identifier: std::ptr::null_mut(),
+                error: paimon_error::from_paimon(e),
+            }
+        }
+    };
     let wrapper = Box::new(paimon_identifier {
-        inner: Box::into_raw(Box::new(Identifier::new(db, obj))) as *mut c_void,
+        inner: Box::into_raw(Box::new(identifier)) as *mut c_void,
     });
     paimon_result_identifier_new {
         identifier: Box::into_raw(wrapper),
@@ -72,3 +81,39 @@ pub unsafe extern "C" fn paimon_identifier_free(id: *mut paimon_identifier) {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::ffi::CString;
+
+    use super::*;
+    use crate::error::paimon_error_free;
+
+    #[test]
+    fn test_paimon_identifier_new_should_reject_path_control_names() {
+        let database = CString::new("db").unwrap();
+        let object = CString::new("../escaped").unwrap();
+
+        let result = unsafe { paimon_identifier_new(database.as_ptr(), object.as_ptr()) };
+
+        assert!(result.identifier.is_null());
+        assert!(!result.error.is_null());
+        unsafe {
+            paimon_error_free(result.error);
+        }
+    }
+
+    #[test]
+    fn test_paimon_identifier_new_should_allow_safe_names() {
+        let database = CString::new("db").unwrap();
+        let object = CString::new("table$snapshots").unwrap();
+
+        let result = unsafe { paimon_identifier_new(database.as_ptr(), object.as_ptr()) };
+
+        assert!(result.error.is_null());
+        assert!(!result.identifier.is_null());
+        unsafe {
+            paimon_identifier_free(result.identifier);
+        }
+    }
+}

crates/paimon/src/catalog/filesystem.rs

@@ -215,6 +215,8 @@ impl Catalog for FileSystemCatalog {
         ignore_if_exists: bool,
         properties: HashMap<String, String>,
     ) -> Result<()> {
+        Identifier::validate_database_name(name)?;
+
         if properties.contains_key(DB_LOCATION_PROP) {
             return Err(Error::ConfigInvalid {
                 message: "Cannot specify location for a database when using fileSystem catalog."
@@ -239,6 +241,8 @@ impl Catalog for FileSystemCatalog {
     }
 
     async fn get_database(&self, name: &str) -> Result<Database> {
+        Identifier::validate_database_name(name)?;
+
         if !self.database_exists(name).await? {
             return Err(Error::DatabaseNotExist {
                 database: name.to_string(),
@@ -254,6 +258,8 @@ impl Catalog for FileSystemCatalog {
         ignore_if_not_exists: bool,
         cascade: bool,
     ) -> Result<()> {
+        Identifier::validate_database_name(name)?;
+
         let path = self.database_path(name);
 
         let database_exists = self.database_exists(name).await?;
@@ -279,6 +285,8 @@ impl Catalog for FileSystemCatalog {
     }
 
     async fn get_table(&self, identifier: &Identifier) -> Result<Table> {
+        identifier.validate()?;
+
         let table_path = self.table_path(identifier);
 
         if !self.table_exists(identifier).await? {
@@ -304,6 +312,8 @@ impl Catalog for FileSystemCatalog {
     }
 
     async fn list_tables(&self, database_name: &str) -> Result<Vec<String>> {
+        Identifier::validate_database_name(database_name)?;
+
         let path = self.database_path(database_name);
 
         if !self.database_exists(database_name).await? {
@@ -321,6 +331,8 @@ impl Catalog for FileSystemCatalog {
         creation: Schema,
         ignore_if_exists: bool,
     ) -> Result<()> {
+        identifier.validate()?;
+
         let table_path = self.table_path(identifier);
 
         let table_exists = self.table_exists(identifier).await?;
@@ -348,6 +360,8 @@ impl Catalog for FileSystemCatalog {
     }
 
     async fn drop_table(&self, identifier: &Identifier, ignore_if_not_exists: bool) -> Result<()> {
+        identifier.validate()?;
+
         let table_path = self.table_path(identifier);
 
         let table_exists = self.table_exists(identifier).await?;
@@ -372,6 +386,9 @@ impl Catalog for FileSystemCatalog {
         to: &Identifier,
         ignore_if_not_exists: bool,
     ) -> Result<()> {
+        from.validate()?;
+        to.validate()?;
+
         let from_path = self.table_path(from);
         let to_path = self.table_path(to);
 
@@ -403,6 +420,8 @@ impl Catalog for FileSystemCatalog {
         changes: Vec<crate::spec::SchemaChange>,
         ignore_if_not_exists: bool,
     ) -> Result<()> {
+        identifier.validate()?;
+
         let table_path = self.table_path(identifier);
         if !self.table_exists(identifier).await? {
             if ignore_if_not_exists {
@@ -511,6 +530,31 @@ mod tests {
         assert!(!catalog.database_exists("db1").await.unwrap());
     }
 
+    #[tokio::test]
+    async fn test_create_database_should_reject_path_traversal_name() {
+        let (temp_dir, catalog) = create_test_catalog();
+        let escaped_path = temp_dir.path().join("escaped.db");
+
+        let result = catalog
+            .create_database("../escaped", false, HashMap::new())
+            .await;
+
+        assert!(matches!(result, Err(Error::IdentifierInvalid { .. })));
+        assert!(!escaped_path.exists());
+    }
+
+    #[tokio::test]
+    async fn test_drop_database_should_reject_path_traversal_name() {
+        let (temp_dir, catalog) = create_test_catalog();
+        let escaped_path = temp_dir.path().join("escaped.db");
+        std::fs::create_dir(&escaped_path).unwrap();
+
+        let result = catalog.drop_database("../escaped", false, true).await;
+
+        assert!(matches!(result, Err(Error::IdentifierInvalid { .. })));
+        assert!(escaped_path.exists());
+    }
+
     #[tokio::test]
     async fn test_table_operations() {
         let (_temp_dir, catalog) = create_test_catalog();
@@ -568,6 +612,54 @@ mod tests {
         assert!(!tables.contains(&"table1".to_string()));
     }
 
+    #[tokio::test]
+    async fn test_create_table_should_reject_path_traversal_object_name() {
+        let (temp_dir, catalog) = create_test_catalog();
+        catalog
+            .create_database("db1", false, HashMap::new())
+            .await
+            .unwrap();
+        let escaped_path = temp_dir.path().join("table_escape");
+
+        let result = catalog
+            .create_table(
+                &Identifier::new("db1", "../../table_escape"),
+                testing_schema(),
+                false,
+            )
+            .await;
+
+        assert!(matches!(result, Err(Error::IdentifierInvalid { .. })));
+        assert!(!escaped_path.exists());
+    }
+
+    #[tokio::test]
+    async fn test_rename_table_should_reject_path_traversal_target_name() {
+        let (temp_dir, catalog) = create_test_catalog();
+        catalog
+            .create_database("db1", false, HashMap::new())
+            .await
+            .unwrap();
+        let source = Identifier::new("db1", "source");
+        catalog
+            .create_table(&source, testing_schema(), false)
+            .await
+            .unwrap();
+        let escaped_path = temp_dir.path().join("renamed_escape");
+
+        let result = catalog
+            .rename_table(
+                &source,
+                &Identifier::new("db1", "../../renamed_escape"),
+                false,
+            )
+            .await;
+
+        assert!(matches!(result, Err(Error::IdentifierInvalid { .. })));
+        assert!(!escaped_path.exists());
+        assert!(catalog.get_table(&source).await.is_ok());
+    }
+
     #[tokio::test]
     async fn test_list_partitions_default_table_not_found_errors() {
         let (_temp_dir, catalog) = create_test_catalog();

crates/paimon/src/catalog/mod.rs

@@ -29,6 +29,7 @@ mod rest;
 use std::collections::HashMap;
 use std::fmt;
 
+use crate::{Error, Result};
 pub use database::*;
 pub use factory::*;
 pub use filesystem::*;
@@ -76,6 +77,29 @@ impl Identifier {
         }
     }
 
+    /// Create an identifier from database and object name, validating both names.
+    pub fn try_new(database: impl Into<String>, object: impl Into<String>) -> Result<Self> {
+        let identifier = Self::new(database, object);
+        identifier.validate()?;
+        Ok(identifier)
+    }
+
+    /// Validate this identifier's database and object names.
+    pub fn validate(&self) -> Result<()> {
+        Self::validate_database_name(&self.database)?;
+        Self::validate_object_name(&self.object)
+    }
+
+    /// Validate a database name for path-safe catalog use.
+    pub fn validate_database_name(name: &str) -> Result<()> {
+        validate_identifier_name("database", name)
+    }
+
+    /// Validate an object name for path-safe catalog use.
+    pub fn validate_object_name(name: &str) -> Result<()> {
+        validate_identifier_name("object", name)
+    }
+
     /// Database name.
     pub fn database(&self) -> &str {
         &self.database
@@ -97,6 +121,28 @@ impl Identifier {
     }
 }
 
+fn validate_identifier_name(kind: &str, name: &str) -> Result<()> {
+    let invalid = if name.trim().is_empty() {
+        Some("cannot be empty or whitespace")
+    } else if matches!(name, "." | "..") {
+        Some("cannot be '.' or '..'")
+    } else if name.contains('/') || name.contains('\\') {
+        Some("cannot contain path separators")
+    } else if name.chars().any(char::is_control) {
+        Some("cannot contain control characters")
+    } else {
+        None
+    };
+
+    if let Some(reason) = invalid {
+        return Err(Error::IdentifierInvalid {
+            message: format!("{kind} name {reason}: {name:?}"),
+        });
+    }
+
+    Ok(())
+}
+
 impl fmt::Display for Identifier {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         write!(f, "{}", self.full_name())
@@ -119,7 +165,6 @@ use async_trait::async_trait;
 use crate::api::PagedList;
 use crate::spec::{Partition, Schema, SchemaChange};
 use crate::table::Table;
-use crate::Result;
 
 /// Catalog API for reading and writing metadata (databases, tables) in Paimon.
 ///
@@ -257,3 +302,46 @@ pub trait Catalog: Send + Sync {
         ))
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_identifier_try_new_should_reject_path_control_names() {
+        for (database, object) in [
+            ("", "table"),
+            ("   ", "table"),
+            (".", "table"),
+            ("..", "table"),
+            ("../escaped", "table"),
+            ("db\\escaped", "table"),
+            ("db\nescaped", "table"),
+            ("db", ""),
+            ("db", "   "),
+            ("db", "."),
+            ("db", ".."),
+            ("db", "../escaped"),
+            ("db", "nested/table"),
+            ("db", "nested\\table"),
+            ("db", "table\0name"),
+        ] {
+            let result = Identifier::try_new(database, object);
+            assert!(
+                matches!(result, Err(Error::IdentifierInvalid { .. })),
+                "expected invalid identifier for database={database:?}, object={object:?}, got {result:?}"
+            );
+        }
+    }
+
+    #[test]
+    fn test_identifier_try_new_should_allow_system_suffix_and_unicode_names() {
+        let identifier = Identifier::try_new("analytics", "orders$snapshots").unwrap();
+        assert_eq!(identifier.database(), "analytics");
+        assert_eq!(identifier.object(), "orders$snapshots");
+
+        let identifier = Identifier::try_new("数据", "订单").unwrap();
+        assert_eq!(identifier.database(), "数据");
+        assert_eq!(identifier.object(), "订单");
+    }
+}