fix(catalog): validate identifier names used to build catalog paths

include/paimon/catalog/catalog.h

@@ -165,7 +165,8 @@ class PAIMON_EXPORT Catalog {
     /// @note This does not check whether the database actually exists.
     ///
     /// @param db_name The name of the database to get the location for.
-    /// @return A string representing the expected location of the database.
+    /// @return A string representing the expected location of the database, or an empty string
+    /// when the name does not form a valid location.
     virtual std::string GetDatabaseLocation(const std::string& db_name) const = 0;
 
     /// Returns the expected location of a specified table.

src/paimon/core/catalog/catalog_utils.cpp

@@ -18,10 +18,13 @@
 
 #include "paimon/core/catalog/catalog_utils.h"
 
+#include <algorithm>
+#include <cctype>
 #include <optional>
 
 #include "fmt/format.h"
 #include "paimon/catalog/catalog.h"
+#include "paimon/common/utils/string_utils.h"
 #include "paimon/result.h"
 
 namespace paimon {
@@ -33,6 +36,27 @@ Status SystemTableError(const Identifier& identifier, const std::string& action)
                                        action, identifier.ToString()));
 }
 
+/// Rejects names that cannot be used as a single path component: such a name would make the
+/// path built from it escape the directory it is joined to.
+Status CheckValidIdentifierName(const std::string& kind, const std::string& name) {
+    const char* reason = nullptr;
+    if (StringUtils::IsNullOrWhitespaceOnly(name)) {
+        reason = "cannot be empty or whitespace";
+    } else if (name == "." || name == "..") {
+        reason = "cannot be '.' or '..'";
+    } else if (name.find('/') != std::string::npos || name.find('\\') != std::string::npos) {
+        reason = "cannot contain path separators";
+    } else if (std::any_of(name.begin(), name.end(), [](char c) {
+                   return std::iscntrl(static_cast<unsigned char>(c)) != 0;
+               })) {
+        reason = "cannot contain control characters";
+    }
+    if (reason != nullptr) {
+        return Status::Invalid(fmt::format("{} name {}: '{}'", kind, reason, name));
+    }
+    return Status::OK();
+}
+
 }  // namespace
 
 bool CatalogUtils::IsSystemDatabase(const std::string& db_name) {
@@ -70,4 +94,31 @@ Status CatalogUtils::CheckNotBranch(const Identifier& identifier, const std::str
     return Status::OK();
 }
 
+Status CatalogUtils::CheckValidDatabaseName(const std::string& db_name) {
+    return CheckValidIdentifierName("database", db_name);
+}
+
+Status CatalogUtils::CheckValidTableName(const Identifier& identifier) {
+    PAIMON_ASSIGN_OR_RAISE(std::string data_table_name, identifier.GetDataTableName());
+    PAIMON_RETURN_NOT_OK(CheckValidIdentifierName("table", data_table_name));
+    PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> branch, identifier.GetBranchName());
+    if (branch) {
+        PAIMON_RETURN_NOT_OK(CheckValidIdentifierName("branch", branch.value()));
+    }
+    PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> system_table,
+                           identifier.GetSystemTableName());
+    if (system_table) {
+        PAIMON_RETURN_NOT_OK(CheckValidIdentifierName("system table", system_table.value()));
+    }
+    return Status::OK();
+}
+
+Status CatalogUtils::CheckValidBranchName(const std::string& branch) {
+    // An empty branch selects the main branch, see BranchManager::NormalizeBranch.
+    if (StringUtils::IsNullOrWhitespaceOnly(branch)) {
+        return Status::OK();
+    }
+    return CheckValidIdentifierName("branch", branch);
+}
+
 }  // namespace paimon

src/paimon/core/catalog/catalog_utils.h

@@ -43,6 +43,18 @@ class CatalogUtils {
 
     /// Fails when `identifier` carries a "$branch_" suffix.
     static Status CheckNotBranch(const Identifier& identifier, const std::string& action);
+
+    /// Fails when `db_name` cannot be used as a single path component, which is required to
+    /// keep the database path under the warehouse.
+    static Status CheckValidDatabaseName(const std::string& db_name);
+
+    /// Fails when any component parsed out of the identifier's table name (data table name,
+    /// branch name, system table name) cannot be used as a single path component.
+    static Status CheckValidTableName(const Identifier& identifier);
+
+    /// Fails when `branch` cannot be used as a single path component. An empty or
+    /// whitespace-only branch selects the main branch and is accepted.
+    static Status CheckValidBranchName(const std::string& branch);
 };
 
 }  // namespace paimon

src/paimon/core/catalog/file_system_catalog.cpp

@@ -87,7 +87,7 @@ Status FileSystemCatalog::CreateDatabaseImpl(const std::string& db_name,
             fmt::join(options, ", "));
         PAIMON_LOG_DEBUG(logger_, "%s", log_msg.c_str());
     }
-    std::string db_path = NewDatabasePath(warehouse_, db_name);
+    PAIMON_ASSIGN_OR_RAISE(std::string db_path, NewDatabasePath(warehouse_, db_name));
     PAIMON_RETURN_NOT_OK(fs_->Mkdirs(db_path));
     return Status::OK();
 }
@@ -96,14 +96,18 @@ Result<bool> FileSystemCatalog::DatabaseExists(const std::string& db_name) const
     if (CatalogUtils::IsSystemDatabase(db_name)) {
         return true;
     }
-    return fs_->Exists(NewDatabasePath(warehouse_, db_name));
+    PAIMON_ASSIGN_OR_RAISE(std::string db_path, NewDatabasePath(warehouse_, db_name));
+    return fs_->Exists(db_path);
 }
 
 Result<bool> FileSystemCatalog::TableExists(const Identifier& identifier) const {
     // Handle sys database global tables
     if (CatalogUtils::IsSystemDatabase(identifier.GetDatabaseName())) {
         return GlobalSystemTableLoader::IsSupported(identifier.GetTableName(), catalog_options_);
     }
+    // The branch component is dropped when the data table identifier is rebuilt below, so the
+    // identifier is validated as a whole here.
+    PAIMON_RETURN_NOT_OK(CatalogUtils::CheckValidTableName(identifier));
     PAIMON_ASSIGN_OR_RAISE(bool is_system_table, identifier.IsSystemTable());
     if (is_system_table) {
         PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> system_table_name,
@@ -123,7 +127,8 @@ Result<bool> FileSystemCatalog::TableExists(const Identifier& identifier) const
 }
 
 std::string FileSystemCatalog::GetDatabaseLocation(const std::string& db_name) const {
-    return NewDatabasePath(warehouse_, db_name);
+    // An invalid name has no valid location, keep the same convention as RestCatalog.
+    return NewDatabasePath(warehouse_, db_name).value_or("");
 }
 
 Result<std::string> FileSystemCatalog::GetTableLocation(const Identifier& identifier) const {
@@ -204,16 +209,19 @@ Result<bool> FileSystemCatalog::IsSystemTable(const Identifier& identifier) {
     return IsSpecifiedSystemTable(identifier);
 }
 
-std::string FileSystemCatalog::NewDatabasePath(const std::string& warehouse,
-                                               const std::string& db_name) {
+Result<std::string> FileSystemCatalog::NewDatabasePath(const std::string& warehouse,
+                                                       const std::string& db_name) {
+    PAIMON_RETURN_NOT_OK(CatalogUtils::CheckValidDatabaseName(db_name));
     return PathUtil::JoinPath(warehouse, db_name + DB_SUFFIX);
 }
 
 Result<std::string> FileSystemCatalog::NewDataTablePath(const std::string& warehouse,
                                                         const Identifier& identifier) {
+    PAIMON_RETURN_NOT_OK(CatalogUtils::CheckValidTableName(identifier));
     PAIMON_ASSIGN_OR_RAISE(std::string data_table_name, identifier.GetDataTableName());
-    return PathUtil::JoinPath(NewDatabasePath(warehouse, identifier.GetDatabaseName()),
-                              data_table_name);
+    PAIMON_ASSIGN_OR_RAISE(std::string database_path,
+                           NewDatabasePath(warehouse, identifier.GetDatabaseName()));
+    return PathUtil::JoinPath(database_path, data_table_name);
 }
 
 Result<std::vector<std::string>> FileSystemCatalog::ListDatabases() const {
@@ -235,7 +243,7 @@ Result<std::vector<std::string>> FileSystemCatalog::ListTables(const std::string
     if (CatalogUtils::IsSystemDatabase(db_name)) {
         return GlobalSystemTableLoader::GetSupportedTableNames(catalog_options_);
     }
-    std::string database_path = NewDatabasePath(warehouse_, db_name);
+    PAIMON_ASSIGN_OR_RAISE(std::string database_path, NewDatabasePath(warehouse_, db_name));
     std::vector<BasicFileStatus> file_status_list;
     PAIMON_RETURN_NOT_OK(fs_->ListDir(database_path, &file_status_list));
     std::vector<std::string> table_names;
@@ -284,6 +292,9 @@ Result<std::shared_ptr<Schema>> FileSystemCatalog::LoadTableSchema(
                                system_table->ArrowSchema());
         return std::make_shared<SystemTableSchema>(std::move(arrow_schema));
     }
+    // The branch component is dropped when the data table identifier is rebuilt below, so the
+    // identifier is validated as a whole here.
+    PAIMON_RETURN_NOT_OK(CatalogUtils::CheckValidTableName(identifier));
     PAIMON_ASSIGN_OR_RAISE(bool is_system_table, identifier.IsSystemTable());
     if (is_system_table) {
         PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> system_table_name,
@@ -342,7 +353,7 @@ Status FileSystemCatalog::DropDatabase(const std::string& name, bool ignore_if_n
         }
     }
 
-    std::string db_path = NewDatabasePath(warehouse_, name);
+    PAIMON_ASSIGN_OR_RAISE(std::string db_path, NewDatabasePath(warehouse_, name));
 
     if (cascade) {
         // List all tables in the database and drop them
@@ -511,6 +522,7 @@ Status FileSystemCatalog::RenameTable(const Identifier& from_table, const Identi
 
 Result<std::vector<SnapshotInfo>> FileSystemCatalog::ListSnapshots(
     const Identifier& identifier, const std::string& branch) const {
+    PAIMON_RETURN_NOT_OK(CatalogUtils::CheckValidBranchName(branch));
     PAIMON_ASSIGN_OR_RAISE(bool exists, TableExists(identifier));
     if (!exists) {
         return Status::NotExist(fmt::format("table {} does not exist", identifier.ToString()));

src/paimon/core/catalog/file_system_catalog.h

@@ -70,7 +70,12 @@ class FileSystemCatalog : public Catalog {
                                                     const std::string& branch) const override;
 
  private:
-    static std::string NewDatabasePath(const std::string& warehouse, const std::string& db_name);
+    /// Fails when `db_name` cannot be used as a single path component, so that the returned
+    /// path always stays under `warehouse`.
+    static Result<std::string> NewDatabasePath(const std::string& warehouse,
+                                               const std::string& db_name);
+    /// Fails when the database name or any component of the table name cannot be used as a
+    /// single path component, so that the returned path always stays under `warehouse`.
     static Result<std::string> NewDataTablePath(const std::string& warehouse,
                                                 const Identifier& identifier);
     static Result<bool> IsSpecifiedSystemTable(const Identifier& identifier);

src/paimon/core/catalog/file_system_catalog_test.cpp

@@ -1270,4 +1270,149 @@ TEST(FileSystemCatalogTest, TestDropTableWithBranchExternalPaths) {
     ASSERT_FALSE(external_exists);
 }
 
+TEST(FileSystemCatalogTest, TestRejectInvalidNames) {
+    std::map<std::string, std::string> options;
+    options[Options::FILE_SYSTEM] = "local";
+    options[Options::FILE_FORMAT] = "orc";
+    ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options));
+    auto dir = UniqueTestDirectory::Create();
+    ASSERT_TRUE(dir);
+    auto fs = core_options.GetFileSystem();
+    // The warehouse is nested inside the test directory, so that the test can assert the
+    // surrounding directory stays untouched.
+    std::string warehouse = PathUtil::JoinPath(dir->Str(), "warehouse");
+    ASSERT_OK(fs->Mkdirs(warehouse));
+    std::string outer_db_path = PathUtil::JoinPath(dir->Str(), "outside.db");
+    FileSystemCatalog catalog(fs, warehouse, options);
+
+    // A rejected database name must fail without creating anything on disk.
+    ASSERT_NOK_WITH_MSG(catalog.CreateDatabase("../outside", {}, /*ignore_if_exists=*/false),
+                        "cannot contain path separators");
+    ASSERT_OK_AND_ASSIGN(bool path_exists, fs->Exists(outer_db_path));
+    ASSERT_FALSE(path_exists);
+
+    // A directory that already exists next to the warehouse must not be deleted either.
+    ASSERT_OK(fs->Mkdirs(outer_db_path));
+    ASSERT_NOK_WITH_MSG(catalog.DropDatabase("../outside", /*ignore_if_not_exists=*/true,
+                                             /*cascade=*/true),
+                        "cannot contain path separators");
+    ASSERT_OK_AND_ASSIGN(path_exists, fs->Exists(outer_db_path));
+    ASSERT_TRUE(path_exists);
+
+    ASSERT_NOK_WITH_MSG(catalog.DatabaseExists("../outside"), "cannot contain path separators");
+    ASSERT_NOK_WITH_MSG(catalog.ListTables("../outside"), "cannot contain path separators");
+    ASSERT_EQ(catalog.GetDatabaseLocation("../outside"), "");
+
+    arrow::FieldVector fields = {arrow::field("f0", arrow::int32())};
+    arrow::Schema typed_schema(fields);
+    ASSERT_OK(catalog.CreateDatabase("db1", {}, /*ignore_if_exists=*/false));
+    {
+        ::ArrowSchema schema;
+        ASSERT_TRUE(arrow::ExportSchema(typed_schema, &schema).ok());
+        ASSERT_OK(catalog.CreateTable(Identifier("db1", "t"), &schema, {}, {}, options,
+                                      /*ignore_if_exists=*/false));
+    }
+
+    // All table entries reject invalid names before touching the file system. The schema is
+    // never imported on these paths, so a single exported schema can be reused.
+    ::ArrowSchema schema;
+    ASSERT_TRUE(arrow::ExportSchema(typed_schema, &schema).ok());
+    const Identifier rejected_db_table("../outside", "t");
+    const Identifier rejected_table("db1", "../evil");
+    ASSERT_NOK_WITH_MSG(catalog.CreateTable(rejected_db_table, &schema, {}, {}, options, false),
+                        "cannot contain path separators");
+    ASSERT_NOK_WITH_MSG(catalog.CreateTable(rejected_table, &schema, {}, {}, options, false),
+                        "cannot contain path separators");
+    ArrowSchemaRelease(&schema);
+
+    ASSERT_NOK_WITH_MSG(catalog.GetTableLocation(rejected_table), "cannot contain path separators");
+    ASSERT_NOK_WITH_MSG(catalog.GetTable(rejected_table), "cannot contain path separators");
+    ASSERT_NOK_WITH_MSG(catalog.TableExists(rejected_table), "cannot contain path separators");
+    ASSERT_NOK_WITH_MSG(catalog.DropTable(rejected_table, /*ignore_if_not_exists=*/true),
+                        "cannot contain path separators");
+    ASSERT_NOK_WITH_MSG(catalog.RenameTable(Identifier("db1", "t"), rejected_table,
+                                            /*ignore_if_not_exists=*/false),
+                        "cannot contain path separators");
+
+    // The branch component of a table name and the branch argument become path components too.
+    ASSERT_NOK_WITH_MSG(catalog.GetTableLocation(Identifier("db1", "t$branch_../../x")),
+                        "branch name cannot contain path separators");
+    ASSERT_NOK_WITH_MSG(catalog.ListSnapshots(Identifier("db1", "t"), "../../x"),
+                        "branch name cannot contain path separators");
+
+    // A system table identifier keeps its own branch component, which the entries resolving the
+    // data table must reject as well.
+    const Identifier rejected_branch_system_table("db1", "t$branch_../../x$snapshots");
+    ASSERT_NOK_WITH_MSG(catalog.TableExists(rejected_branch_system_table),
+                        "branch name cannot contain path separators");
+    ASSERT_NOK_WITH_MSG(catalog.LoadTableSchema(rejected_branch_system_table),
+                        "branch name cannot contain path separators");
+    ASSERT_NOK_WITH_MSG(catalog.GetTable(rejected_branch_system_table),
+                        "branch name cannot contain path separators");
+
+    // The surrounding directory is untouched and the valid table still works.
+    ASSERT_OK_AND_ASSIGN(path_exists, fs->Exists(PathUtil::JoinPath(dir->Str(), "db1.db")));
+    ASSERT_FALSE(path_exists);
+    ASSERT_OK_AND_ASSIGN(bool table_exists, catalog.TableExists(Identifier("db1", "t")));
+    ASSERT_TRUE(table_exists);
+}
+
+TEST(FileSystemCatalogTest, TestIdentifierNameValidationRules) {
+    std::map<std::string, std::string> options;
+    options[Options::FILE_SYSTEM] = "local";
+    options[Options::FILE_FORMAT] = "orc";
+    ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options));
+    auto dir = UniqueTestDirectory::Create();
+    ASSERT_TRUE(dir);
+    FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str(), options);
+    ASSERT_OK(catalog.CreateDatabase("db1", {}, /*ignore_if_exists=*/false));
+
+    arrow::FieldVector fields = {arrow::field("f0", arrow::int32())};
+    arrow::Schema typed_schema(fields);
+    struct InvalidName {
+        std::string name;
+        std::string db_error;
+        // An empty table name is already rejected by the identifier itself.
+        std::string table_error;
+    };
+    const std::vector<InvalidName> invalid_names = {
+        {"", "cannot be empty or whitespace", "Invalid table name"},
+        {"   ", "cannot be empty or whitespace", "cannot be empty or whitespace"},
+        {".", "cannot be '.' or '..'", "cannot be '.' or '..'"},
+        {"..", "cannot be '.' or '..'", "cannot be '.' or '..'"},
+        {"../escaped", "cannot contain path separators", "cannot contain path separators"},
+        {"nested/name", "cannot contain path separators", "cannot contain path separators"},
+        {"back\\slash", "cannot contain path separators", "cannot contain path separators"},
+        {"line\nfeed", "cannot contain control characters", "cannot contain control characters"},
+        {std::string("nul\0byte", 8), "cannot contain control characters",
+         "cannot contain control characters"},
+    };
+    ::ArrowSchema schema;
+    ASSERT_TRUE(arrow::ExportSchema(typed_schema, &schema).ok());
+    for (const auto& invalid_name : invalid_names) {
+        ASSERT_NOK_WITH_MSG(catalog.CreateDatabase(invalid_name.name, {},
+                                                   /*ignore_if_exists=*/true),
+                            invalid_name.db_error);
+        ASSERT_NOK_WITH_MSG(catalog.CreateTable(Identifier("db1", invalid_name.name), &schema, {},
+                                                {}, options, /*ignore_if_exists=*/true),
+                            invalid_name.table_error);
+    }
+    ArrowSchemaRelease(&schema);
+
+    // Names that merely contain a dot or non-ascii characters stay usable.
+    for (const char* db_name : {"my.db", "a..b", "数据"}) {
+        ASSERT_OK(catalog.CreateDatabase(db_name, {}, /*ignore_if_exists=*/false));
+        ASSERT_OK_AND_ASSIGN(bool db_exists, catalog.DatabaseExists(db_name));
+        ASSERT_TRUE(db_exists);
+    }
+    for (const char* table_name : {"orders", "订单"}) {
+        ::ArrowSchema valid_schema;
+        ASSERT_TRUE(arrow::ExportSchema(typed_schema, &valid_schema).ok());
+        ASSERT_OK(catalog.CreateTable(Identifier("db1", table_name), &valid_schema, {}, {}, options,
+                                      /*ignore_if_exists=*/false));
+        ASSERT_OK_AND_ASSIGN(bool table_exists, catalog.TableExists(Identifier("db1", table_name)));
+        ASSERT_TRUE(table_exists);
+    }
+}
+
 }  // namespace paimon::test