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

include/paimon/catalog/catalog.h

@@ -162,19 +162,20 @@ class PAIMON_EXPORT Catalog {
 
     /// Returns the expected location of a specified database.
     ///
-    /// @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.
-    virtual std::string GetDatabaseLocation(const std::string& db_name) const = 0;
+    /// @return A result containing the expected location of the database, or an error status on
+    /// failure. An implementation that builds the location from the warehouse path, such as the
+    /// file system catalog, answers without checking whether the database exists. One that resolves
+    /// the location on a server, such as the REST catalog, propagates the server's error and so
+    /// fails for a database that does not exist.
+    virtual Result<std::string> GetDatabaseLocation(const std::string& db_name) const = 0;
 
     /// Returns the expected location of a specified table.
     ///
-    /// @note This does not check whether the table actually exists.
-    ///
     /// @param identifier The table identifier containing database and table name.
     /// @return A result containing the expected location of the table, or an error status on
-    /// failure.
+    /// failure. Whether a missing table is an error depends on the implementation, in the same way
+    /// as for `GetDatabaseLocation`.
     virtual Result<std::string> GetTableLocation(const Identifier& identifier) const = 0;
 
     /// Returns the root path of the catalog.

src/paimon/common/utils/path_util.cpp

@@ -21,6 +21,8 @@
 
 #include <unistd.h>
 
+#include <algorithm>
+#include <cctype>
 #include <cerrno>
 #include <cstddef>
 #include <cstdint>
@@ -34,6 +36,41 @@
 #include "paimon/status.h"
 
 namespace paimon {
+namespace {
+
+/// Escapes the control characters of `name`, so that a rejected name cannot inject a line into
+/// the log the error is written to nor truncate the C string it is copied into. Bytes of a
+/// multi-byte sequence are left alone, since none of them is a control character.
+std::string EscapeControlCharacters(const std::string& name) {
+    std::string escaped;
+    escaped.reserve(name.size());
+    for (char c : name) {
+        switch (c) {
+            case '\\':
+                escaped += "\\\\";
+                break;
+            case '\n':
+                escaped += "\\n";
+                break;
+            case '\r':
+                escaped += "\\r";
+                break;
+            case '\t':
+                escaped += "\\t";
+                break;
+            default:
+                if (std::iscntrl(static_cast<unsigned char>(c)) != 0) {
+                    escaped += fmt::format("\\x{{{:02x}}}", static_cast<unsigned char>(c));
+                } else {
+                    escaped += c;
+                }
+        }
+    }
+    return escaped;
+}
+
+}  // namespace
+
 std::string Path::ToString() const {
     std::string ret;
     if (!scheme.empty()) {
@@ -169,4 +206,24 @@ Result<std::string> PathUtil::CreateTempPath(const std::string& path) noexcept {
     return JoinPath(GetParentDirPath(path), fmt::format(".{}.{}.tmp", GetName(path), uuid));
 }
 
+Status PathUtil::CheckSinglePathComponent(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, EscapeControlCharacters(name)));
+    }
+    return Status::OK();
+}
+
 }  // namespace paimon

src/paimon/common/utils/path_util.h

@@ -52,6 +52,15 @@ class PAIMON_EXPORT PathUtil {
     static Result<Path> ToPath(const std::string& path) noexcept;
     static Result<std::string> NormalizePath(const std::string& path) noexcept;
 
+    /// Fails when `name` cannot be used as a single path component, which is required to keep a
+    /// path built with `JoinPath` under the directory it is joined to: `name` must not be empty
+    /// or whitespace-only, must not be "." or "..", and must contain neither a path separator
+    /// nor a control character. `kind` names the rejected value in the error message, which
+    /// reads "<kind> name <reason>: '<name>'" and escapes the control characters of `name`.
+    ///
+    /// The check is purely lexical and needs no IO.
+    static Status CheckSinglePathComponent(const std::string& kind, const std::string& name);
+
  private:
     static std::string NormalizeInnerPath(const std::string& path) noexcept;
 };

src/paimon/common/utils/path_util_test.cpp

@@ -171,4 +171,43 @@ TEST(PathUtilsTest, TestCreateTempPath) {
     ASSERT_TRUE(StringUtils::EndsWith(tmp_name, ".tmp"));
 }
 
+TEST(PathUtilsTest, TestCheckSinglePathComponent) {
+    // Names that stay a single path component, including names that merely contain a dot and
+    // names outside ascii.
+    for (const char* name : {"db1", "my.db", "a..b", "a b", "数据", "\u00e9t\u00e9"}) {
+        ASSERT_OK(PathUtil::CheckSinglePathComponent("database", name));
+    }
+
+    ASSERT_NOK_WITH_MSG(PathUtil::CheckSinglePathComponent("database", ""),
+                        "database name cannot be empty or whitespace");
+    ASSERT_NOK_WITH_MSG(PathUtil::CheckSinglePathComponent("database", " \t\n "),
+                        "database name cannot be empty or whitespace");
+    ASSERT_NOK_WITH_MSG(PathUtil::CheckSinglePathComponent("table", "."),
+                        "table name cannot be '.' or '..'");
+    ASSERT_NOK_WITH_MSG(PathUtil::CheckSinglePathComponent("table", ".."),
+                        "table name cannot be '.' or '..'");
+    ASSERT_NOK_WITH_MSG(PathUtil::CheckSinglePathComponent("table", "../escaped"),
+                        "table name cannot contain path separators");
+    ASSERT_NOK_WITH_MSG(PathUtil::CheckSinglePathComponent("table", "back\\slash"),
+                        "table name cannot contain path separators");
+    ASSERT_NOK_WITH_MSG(PathUtil::CheckSinglePathComponent("branch", "line\nfeed"),
+                        "branch name cannot contain control characters");
+    ASSERT_NOK_WITH_MSG(PathUtil::CheckSinglePathComponent("branch", std::string("nul\0byte", 8)),
+                        "branch name cannot contain control characters");
+}
+
+TEST(PathUtilsTest, TestCheckSinglePathComponentEscapesRejectedName) {
+    // The rejected name is escaped, so that it can neither add a line to the log the error is
+    // written to nor truncate the C string it is copied into.
+    Status newline = PathUtil::CheckSinglePathComponent("database", "line\nfeed\r\t");
+    ASSERT_FALSE(newline.ok());
+    ASSERT_EQ(newline.ToString().find('\n'), std::string::npos);
+    ASSERT_NE(newline.ToString().find("line\\nfeed\\r\\t"), std::string::npos);
+
+    Status nul = PathUtil::CheckSinglePathComponent("database", std::string("nul\0byte", 8));
+    ASSERT_FALSE(nul.ok());
+    ASSERT_EQ(nul.ToString().find('\0'), std::string::npos);
+    ASSERT_NE(nul.ToString().find("nul\\x{00}byte"), std::string::npos);
+}
+
 }  // namespace paimon::test

src/paimon/core/catalog/catalog_utils.cpp

@@ -22,6 +22,8 @@
 
 #include "fmt/format.h"
 #include "paimon/catalog/catalog.h"
+#include "paimon/common/utils/path_util.h"
+#include "paimon/core/utils/branch_manager.h"
 #include "paimon/result.h"
 
 namespace paimon {
@@ -70,4 +72,26 @@ Status CatalogUtils::CheckNotBranch(const Identifier& identifier, const std::str
     return Status::OK();
 }
 
+Status CatalogUtils::CheckValidDatabaseName(const std::string& db_name) {
+    return PathUtil::CheckSinglePathComponent("database", db_name);
+}
+
+Status CatalogUtils::CheckValidTableName(const Identifier& identifier) {
+    PAIMON_ASSIGN_OR_RAISE(std::string data_table_name, identifier.GetDataTableName());
+    PAIMON_RETURN_NOT_OK(PathUtil::CheckSinglePathComponent("table", data_table_name));
+    PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> branch, identifier.GetBranchName());
+    if (branch) {
+        // The branch of an identifier selects the same directory as the `branch` option, so both
+        // go through the same check.
+        PAIMON_RETURN_NOT_OK(BranchManager::CheckValidBranch(branch.value()));
+    }
+    PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> system_table,
+                           identifier.GetSystemTableName());
+    if (system_table) {
+        PAIMON_RETURN_NOT_OK(
+            PathUtil::CheckSinglePathComponent("system table", system_table.value()));
+    }
+    return Status::OK();
+}
+
 }  // namespace paimon

src/paimon/core/catalog/catalog_utils.h

@@ -43,6 +43,14 @@ 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);
 };
 
 }  // 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,
@@ -122,7 +126,7 @@ Result<bool> FileSystemCatalog::TableExists(const Identifier& identifier) const
     return latest_schema != std::nullopt;
 }
 
-std::string FileSystemCatalog::GetDatabaseLocation(const std::string& db_name) const {
+Result<std::string> FileSystemCatalog::GetDatabaseLocation(const std::string& db_name) const {
     return NewDatabasePath(warehouse_, db_name);
 }
 
@@ -204,16 +208,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 +242,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 +291,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 +352,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 +521,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(BranchManager::CheckValidBranch(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

@@ -59,7 +59,7 @@ class FileSystemCatalog : public Catalog {
     Result<std::vector<std::string>> ListTables(const std::string& db_name) const override;
     Result<bool> DatabaseExists(const std::string& db_name) const override;
     Result<bool> TableExists(const Identifier& identifier) const override;
-    std::string GetDatabaseLocation(const std::string& db_name) const override;
+    Result<std::string> GetDatabaseLocation(const std::string& db_name) const override;
     Result<std::string> GetTableLocation(const Identifier& identifier) const override;
     Result<std::shared_ptr<Schema>> LoadTableSchema(const Identifier& identifier) const override;
     std::string GetRootPath() const override;
@@ -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);