diff --git a/include/paimon/catalog/catalog.h b/include/paimon/catalog/catalog.h index 9213bde94..0dc575f6d 100644 --- a/include/paimon/catalog/catalog.h +++ b/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. diff --git a/src/paimon/core/catalog/catalog_utils.cpp b/src/paimon/core/catalog/catalog_utils.cpp index eae23ad8e..e06c7c4fd 100644 --- a/src/paimon/core/catalog/catalog_utils.cpp +++ b/src/paimon/core/catalog/catalog_utils.cpp @@ -18,10 +18,13 @@ #include "paimon/core/catalog/catalog_utils.h" +#include +#include #include #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(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 branch, identifier.GetBranchName()); + if (branch) { + PAIMON_RETURN_NOT_OK(CheckValidIdentifierName("branch", branch.value())); + } + PAIMON_ASSIGN_OR_RAISE(std::optional 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 diff --git a/src/paimon/core/catalog/catalog_utils.h b/src/paimon/core/catalog/catalog_utils.h index e92cd8443..0f4f6780e 100644 --- a/src/paimon/core/catalog/catalog_utils.h +++ b/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 diff --git a/src/paimon/core/catalog/file_system_catalog.cpp b/src/paimon/core/catalog/file_system_catalog.cpp index 85907c122..c38fc46ce 100644 --- a/src/paimon/core/catalog/file_system_catalog.cpp +++ b/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,7 +96,8 @@ Result 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 FileSystemCatalog::TableExists(const Identifier& identifier) const { @@ -104,6 +105,9 @@ Result FileSystemCatalog::TableExists(const Identifier& identifier) const 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 system_table_name, @@ -123,7 +127,8 @@ Result 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 FileSystemCatalog::GetTableLocation(const Identifier& identifier) const { @@ -204,16 +209,19 @@ Result FileSystemCatalog::IsSystemTable(const Identifier& identifier) { return IsSpecifiedSystemTable(identifier); } -std::string FileSystemCatalog::NewDatabasePath(const std::string& warehouse, - const std::string& db_name) { +Result 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 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> FileSystemCatalog::ListDatabases() const { @@ -235,7 +243,7 @@ Result> 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 file_status_list; PAIMON_RETURN_NOT_OK(fs_->ListDir(database_path, &file_status_list)); std::vector table_names; @@ -284,6 +292,9 @@ Result> FileSystemCatalog::LoadTableSchema( system_table->ArrowSchema()); return std::make_shared(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 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> 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())); diff --git a/src/paimon/core/catalog/file_system_catalog.h b/src/paimon/core/catalog/file_system_catalog.h index 3925aff84..6b27455ba 100644 --- a/src/paimon/core/catalog/file_system_catalog.h +++ b/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 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 NewDataTablePath(const std::string& warehouse, const Identifier& identifier); static Result IsSpecifiedSystemTable(const Identifier& identifier); diff --git a/src/paimon/core/catalog/file_system_catalog_test.cpp b/src/paimon/core/catalog/file_system_catalog_test.cpp index 9f6a568f6..4c61fab91 100644 --- a/src/paimon/core/catalog/file_system_catalog_test.cpp +++ b/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 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 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 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