TableEngineGrant_version2

docs/en/sql-reference/statements/grant.md

@@ -209,6 +209,7 @@ Hierarchy of privileges:
     - `SHOW NAMED COLLECTIONS`
     - `SHOW NAMED COLLECTIONS SECRETS`
     - `NAMED COLLECTION`
+- [TABLE ENGINE](#grant-table-engine)
 
 Examples of how this hierarchy is treated:
 
@@ -505,6 +506,7 @@ and
 [`format_display_secrets_in_show_and_select` format setting](../../operations/settings/formats#format_display_secrets_in_show_and_select)
 are turned on.
 
+
 ### NAMED COLLECTION ADMIN
 
 Allows a certain operation on a specified named collection. Before version 23.7 it was called NAMED COLLECTION CONTROL, and after 23.7 NAMED COLLECTION ADMIN was added and NAMED COLLECTION CONTROL is preserved as an alias.
@@ -524,6 +526,17 @@ Unlike all other grants (CREATE, DROP, ALTER, SHOW) grant NAMED COLLECTION was a
 Assuming a named collection is called abc, we grant privilege CREATE NAMED COLLECTION to user john.
 - `GRANT CREATE NAMED COLLECTION ON abc TO john`
 
+
+### TABLE ENGINE
+
+Allows using a specified table engine when creating a table. Applies to [table engines](../../engines/table-engines/index.md).
+
+**Examples**
+
+- `GRANT TABLE ENGINE ON * TO john`
+- `GRANT TABLE ENGINE ON TinyLog TO john`
+
+
 ### ALL
 
 Grants all the privileges on regulated entity to a user account or a role.

programs/server/config.xml

@@ -742,6 +742,10 @@
              It also enables 'changeable_in_readonly' constraint type -->
         <settings_constraints_replace_previous>true</settings_constraints_replace_previous>
 
+        <!-- By default, for backward compatibility create table with a specific table engine ignores grant,
+             however you can change this behaviour by setting this to true -->
+        <table_engines_require_grant>false</table_engines_require_grant>
+
         <!-- Number of seconds since last access a role is stored in the Role Cache -->
         <role_cache_expiration_time_seconds>600</role_cache_expiration_time_seconds>
     </access_control_improvements>

src/Access/AccessControl.cpp

@@ -285,6 +285,7 @@ void AccessControl::setUpFromMainConfig(const Poco::Util::AbstractConfiguration
     setSelectFromSystemDatabaseRequiresGrant(config_.getBool("access_control_improvements.select_from_system_db_requires_grant", false));
     setSelectFromInformationSchemaRequiresGrant(config_.getBool("access_control_improvements.select_from_information_schema_requires_grant", false));
     setSettingsConstraintsReplacePrevious(config_.getBool("access_control_improvements.settings_constraints_replace_previous", false));
+    setTableEnginesRequireGrant(config_.getBool("access_control_improvements.table_engines_require_grant", false));
 
     addStoragesFromMainConfig(config_, config_path_, get_zookeeper_function_);
 

src/Access/AccessControl.h

@@ -182,6 +182,9 @@ class AccessControl : public MultipleAccessStorage
     void setSettingsConstraintsReplacePrevious(bool enable) { settings_constraints_replace_previous = enable; }
     bool doesSettingsConstraintsReplacePrevious() const { return settings_constraints_replace_previous; }
 
+    void setTableEnginesRequireGrant(bool enable) { table_engines_require_grant = enable; }
+    bool doesTableEnginesRequireGrant() const { return table_engines_require_grant; }
+
     std::shared_ptr<const ContextAccess> getContextAccess(const ContextAccessParams & params) const;
 
     std::shared_ptr<const EnabledRoles> getEnabledRoles(
@@ -258,6 +261,7 @@ class AccessControl : public MultipleAccessStorage
     std::atomic_bool select_from_system_db_requires_grant = false;
     std::atomic_bool select_from_information_schema_requires_grant = false;
     std::atomic_bool settings_constraints_replace_previous = false;
+    std::atomic_bool table_engines_require_grant = false;
     std::atomic_int bcrypt_workfactor = 12;
     std::atomic<AuthenticationType> default_password_type = AuthenticationType::SHA256_PASSWORD;
 };

src/Access/Common/AccessFlags.cpp

@@ -101,8 +101,9 @@ namespace
         const Flags & getTableFlags() const { return all_flags_for_target[TABLE]; }
         const Flags & getColumnFlags() const { return all_flags_for_target[COLUMN]; }
         const Flags & getDictionaryFlags() const { return all_flags_for_target[DICTIONARY]; }
-        const Flags & getNamedCollectionFlags() const { return all_flags_for_target[NAMED_COLLECTION]; }
+        const Flags & getTableEngineFlags() const { return all_flags_for_target[TABLE_ENGINE]; }
         const Flags & getUserNameFlags() const { return all_flags_for_target[USER_NAME]; }
+        const Flags & getNamedCollectionFlags() const { return all_flags_for_target[NAMED_COLLECTION]; }
         const Flags & getAllFlagsGrantableOnGlobalLevel() const { return getAllFlags(); }
         const Flags & getAllFlagsGrantableOnGlobalWithParameterLevel() const { return getGlobalWithParameterFlags(); }
         const Flags & getAllFlagsGrantableOnDatabaseLevel() const { return all_flags_grantable_on_database_level; }
@@ -122,6 +123,7 @@ namespace
             DICTIONARY,
             NAMED_COLLECTION,
             USER_NAME,
+            TABLE_ENGINE,
         };
 
         struct Node;
@@ -301,7 +303,7 @@ namespace
                 collectAllFlags(child.get());
 
             all_flags_grantable_on_table_level = all_flags_for_target[TABLE] | all_flags_for_target[DICTIONARY] | all_flags_for_target[COLUMN];
-            all_flags_grantable_on_global_with_parameter_level = all_flags_for_target[NAMED_COLLECTION] | all_flags_for_target[USER_NAME];
+            all_flags_grantable_on_global_with_parameter_level = all_flags_for_target[NAMED_COLLECTION] | all_flags_for_target[USER_NAME] | all_flags_for_target[TABLE_ENGINE];
             all_flags_grantable_on_database_level = all_flags_for_target[DATABASE] | all_flags_grantable_on_table_level;
         }
 
@@ -352,7 +354,7 @@ namespace
         std::unordered_map<std::string_view, Flags> keyword_to_flags_map;
         std::vector<Flags> access_type_to_flags_mapping;
         Flags all_flags;
-        Flags all_flags_for_target[static_cast<size_t>(USER_NAME) + 1];
+        Flags all_flags_for_target[static_cast<size_t>(TABLE_ENGINE) + 1];
         Flags all_flags_grantable_on_database_level;
         Flags all_flags_grantable_on_table_level;
         Flags all_flags_grantable_on_global_with_parameter_level;
@@ -376,7 +378,11 @@ std::unordered_map<AccessFlags::ParameterType, AccessFlags> AccessFlags::splitIn
     if (user_flags)
         result.emplace(ParameterType::USER_NAME, user_flags);
 
-    auto other_flags = (~named_collection_flags & ~user_flags) & *this;
+    auto table_engine_flags = AccessFlags::allTableEngineFlags() & *this;
+    if (table_engine_flags)
+        result.emplace(ParameterType::TABLE_ENGINE, table_engine_flags);
+
+    auto other_flags = (~named_collection_flags & ~user_flags & ~table_engine_flags) & *this;
     if (other_flags)
         result.emplace(ParameterType::NONE, other_flags);
 
@@ -395,6 +401,10 @@ AccessFlags::ParameterType AccessFlags::getParameterType() const
     if (AccessFlags::allUserNameFlags().contains(*this))
         return AccessFlags::USER_NAME;
 
+    /// All flags refer to TABLE ENGINE access type.
+    if (AccessFlags::allTableEngineFlags().contains(*this))
+        return AccessFlags::TABLE_ENGINE;
+
     throw Exception(ErrorCodes::MIXED_ACCESS_PARAMETER_TYPES, "Having mixed parameter types: {}", toString());
 }
 
@@ -414,6 +424,7 @@ AccessFlags AccessFlags::allColumnFlags() { return Helper::instance().getColumnF
 AccessFlags AccessFlags::allDictionaryFlags() { return Helper::instance().getDictionaryFlags(); }
 AccessFlags AccessFlags::allNamedCollectionFlags() { return Helper::instance().getNamedCollectionFlags(); }
 AccessFlags AccessFlags::allUserNameFlags() { return Helper::instance().getUserNameFlags(); }
+AccessFlags AccessFlags::allTableEngineFlags() { return Helper::instance().getTableEngineFlags(); }
 AccessFlags AccessFlags::allFlagsGrantableOnGlobalLevel() { return Helper::instance().getAllFlagsGrantableOnGlobalLevel(); }
 AccessFlags AccessFlags::allFlagsGrantableOnGlobalWithParameterLevel() { return Helper::instance().getAllFlagsGrantableOnGlobalWithParameterLevel(); }
 AccessFlags AccessFlags::allFlagsGrantableOnDatabaseLevel() { return Helper::instance().getAllFlagsGrantableOnDatabaseLevel(); }

src/Access/Common/AccessFlags.h

@@ -56,6 +56,7 @@ class AccessFlags
     enum ParameterType
     {
         NONE,
+        TABLE_ENGINE,
         NAMED_COLLECTION,
         USER_NAME,
     };
@@ -107,6 +108,9 @@ class AccessFlags
     /// Returns all the flags related to a user.
     static AccessFlags allUserNameFlags();
 
+    /// Returns all the flags related to a table engine.
+    static AccessFlags allTableEngineFlags();
+
     /// Returns all the flags which could be granted on the global level.
     /// The same as allFlags().
     static AccessFlags allFlagsGrantableOnGlobalLevel();

src/Access/Common/AccessType.h

@@ -12,7 +12,7 @@ enum class AccessType
 /// Macro M should be defined as M(name, aliases, node_type, parent_group_name)
 /// where name is identifier with underscores (instead of spaces);
 /// aliases is a string containing comma-separated list;
-/// node_type either specifies access type's level (GLOBAL/NAMED_COLLECTION/USER_NAME/DATABASE/TABLE/DICTIONARY/VIEW/COLUMNS),
+/// node_type either specifies access type's level (GLOBAL/NAMED_COLLECTION/USER_NAME/TABLE_ENGINE/DATABASE/TABLE/DICTIONARY/VIEW/COLUMNS),
 /// or specifies that the access type is a GROUP of other access types;
 /// parent_group_name is the name of the group containing this access type (or NONE if there is no such group).
 /// NOTE A parent group must be declared AFTER all its children.
@@ -153,6 +153,8 @@ enum class AccessType
     M(NAMED_COLLECTION_ADMIN, "NAMED COLLECTION CONTROL", NAMED_COLLECTION, ALL) \
     M(SET_DEFINER, "", USER_NAME, ALL) \
     \
+    M(TABLE_ENGINE, "TABLE ENGINE", TABLE_ENGINE, ALL) \
+    \
     M(SYSTEM_SHUTDOWN, "SYSTEM KILL, SHUTDOWN", GLOBAL, SYSTEM) \
     M(SYSTEM_DROP_DNS_CACHE, "SYSTEM DROP DNS, DROP DNS CACHE, DROP DNS", GLOBAL, SYSTEM_DROP_CACHE) \
     M(SYSTEM_DROP_MARK_CACHE, "SYSTEM DROP MARK, DROP MARK CACHE, DROP MARKS", GLOBAL, SYSTEM_DROP_CACHE) \

src/Access/ContextAccess.cpp

@@ -547,6 +547,9 @@ bool ContextAccess::checkAccessImplHelper(AccessFlags flags, const Args &... arg
     if (flags & AccessType::CLUSTER && !access_control->doesOnClusterQueriesRequireClusterGrant())
         flags &= ~AccessType::CLUSTER;
 
+    if (flags & AccessType::TABLE_ENGINE && !access_control->doesTableEnginesRequireGrant())
+        flags &= ~AccessType::TABLE_ENGINE;
+
     if (!flags)
         return true;
 

src/Access/tests/gtest_access_rights_ops.cpp

@@ -54,7 +54,7 @@ TEST(AccessRights, Union)
               "SYSTEM MOVES, SYSTEM PULLING REPLICATION LOG, SYSTEM CLEANUP, SYSTEM VIEWS, SYSTEM SENDS, SYSTEM REPLICATION QUEUES, SYSTEM VIRTUAL PARTS UPDATE, "
               "SYSTEM DROP REPLICA, SYSTEM SYNC REPLICA, SYSTEM RESTART REPLICA, "
               "SYSTEM RESTORE REPLICA, SYSTEM WAIT LOADING PARTS, SYSTEM SYNC DATABASE REPLICA, SYSTEM FLUSH DISTRIBUTED, dictGet ON db1.*, "
-              "GRANT SET DEFINER ON db1, GRANT NAMED COLLECTION ADMIN ON db1");
+              "GRANT TABLE ENGINE ON db1, GRANT SET DEFINER ON db1, GRANT NAMED COLLECTION ADMIN ON db1");
 }
 
 

src/Interpreters/InterpreterCreateQuery.cpp

@@ -723,8 +723,10 @@ InterpreterCreateQuery::TableProperties InterpreterCreateQuery::getTableProperti
     if (create.storage && create.storage->engine)
     {
         auto source_access_type = StorageFactory::instance().getSourceAccessType(create.storage->engine->name);
-        if (source_access_type != AccessType::NONE)
+        const auto & access_control = getContext()->getAccessControl();
+        if (source_access_type != AccessType::NONE && !access_control.doesTableEnginesRequireGrant())
             getContext()->checkAccess(source_access_type);
+        getContext()->checkAccess(AccessType::TABLE_ENGINE, create.storage->engine->name);
     }
 
     TableProperties properties;
@@ -1829,8 +1831,11 @@ AccessRightsElements InterpreterCreateQuery::getRequiredAccess() const
     if (create.storage && create.storage->engine)
     {
         auto source_access_type = StorageFactory::instance().getSourceAccessType(create.storage->engine->name);
-        if (source_access_type != AccessType::NONE)
+        const auto & access_control = getContext()->getAccessControl();
+        /// We just need to check GRANT TABLE ENGINE for sources if grant of table engine is enabled.
+        if (source_access_type != AccessType::NONE && !access_control.doesTableEnginesRequireGrant())
             required_access.emplace_back(source_access_type);
+        required_access.emplace_back(AccessType::TABLE_ENGINE, create.storage->engine->name);
     }
 
     return required_access;

src/Storages/System/StorageSystemPrivileges.cpp

@@ -30,6 +30,7 @@ namespace
         COLUMN,
         NAMED_COLLECTION,
         USER_NAME,
+        TABLE_ENGINE,
     };
 
     DataTypeEnum8::Values getLevelEnumValues()
@@ -43,6 +44,7 @@ namespace
         enum_values.emplace_back("COLUMN", static_cast<Int8>(COLUMN));
         enum_values.emplace_back("NAMED_COLLECTION", static_cast<Int8>(NAMED_COLLECTION));
         enum_values.emplace_back("USER_NAME", static_cast<Int8>(USER_NAME));
+        enum_values.emplace_back("TABLE_ENGINE", static_cast<Int8>(TABLE_ENGINE));
         return enum_values;
     }
 }

tests/integration/test_grant_and_revoke/configs/config.xml

@@ -0,0 +1,5 @@
+<clickhouse>
+     <access_control_improvements>
+        <table_engines_require_grant>true</table_engines_require_grant>
+    </access_control_improvements>
+</clickhouse>

tests/integration/test_grant_and_revoke/test.py

@@ -5,9 +5,8 @@
 cluster = ClickHouseCluster(__file__)
 instance = cluster.add_instance(
     "instance",
-    user_configs=[
-        "configs/users.d/users.xml",
-    ],
+    main_configs=["configs/config.xml"],
+    user_configs=["configs/users.d/users.xml"],
 )
 
 
@@ -370,6 +369,7 @@ def test_implicit_create_temporary_table_grant():
     )
 
     instance.query("GRANT CREATE TABLE ON test.* TO A")
+    instance.query("GRANT TABLE ENGINE ON Memory TO A")
     instance.query("CREATE TEMPORARY TABLE tmp(name String)", user="A")
 
     instance.query("REVOKE CREATE TABLE ON *.* FROM A")
@@ -718,3 +718,54 @@ def test_current_grants_override():
             "REVOKE SELECT ON test.* FROM B",
         ]
     )
+
+
+def test_table_engine_grant_and_revoke():
+    instance.query("DROP USER IF EXISTS A")
+    instance.query("CREATE USER A")
+    instance.query("GRANT CREATE TABLE ON test.table1 TO A")
+    assert "Not enough privileges" in instance.query_and_get_error(
+        "CREATE TABLE test.table1(a Integer) engine=TinyLog", user="A"
+    )
+
+    instance.query("GRANT TABLE ENGINE ON TinyLog TO A")
+
+    instance.query("CREATE TABLE test.table1(a Integer) engine=TinyLog", user="A")
+
+    assert instance.query("SHOW GRANTS FOR A") == TSV(
+        [
+            "GRANT TABLE ENGINE ON TinyLog TO A",
+            "GRANT CREATE TABLE ON test.table1 TO A",
+        ]
+    )
+
+    instance.query("REVOKE TABLE ENGINE ON TinyLog FROM A")
+
+    assert "Not enough privileges" in instance.query_and_get_error(
+        "CREATE TABLE test.table1(a Integer) engine=TinyLog", user="A"
+    )
+
+    instance.query("REVOKE CREATE TABLE ON test.table1 FROM A")
+    instance.query("DROP TABLE test.table1")
+
+    assert instance.query("SHOW GRANTS FOR A") == TSV([])
+
+
+def test_table_engine_and_source_grant():
+    instance.query("DROP USER IF EXISTS A")
+    instance.query("CREATE USER A")
+
+    instance.query("GRANT CREATE TABLE ON test.table1 TO A")
+    instance.query("GRANT TABLE ENGINE ON PostgreSQL TO A")
+    # We don't need the following statement as GRANT TABLE ENGINE covers it already.
+    # instance.query("GRANT POSTGRES ON *.* TO A")
+
+    instance.query(
+        """
+        CREATE TABLE test.table1(a Integer) 
+        engine=PostgreSQL('localhost:5432', 'dummy', 'dummy', 'dummy', 'dummy');
+        """,
+        user="A",
+    )
+
+    instance.query("DROP TABLE test.table1")

tests/queries/0_stateless/01271_show_privileges.reference

@@ -103,6 +103,7 @@ SHOW NAMED COLLECTIONS SECRETS	['SHOW NAMED COLLECTIONS SECRETS']	NAMED_COLLECTI
 NAMED COLLECTION	['NAMED COLLECTION USAGE','USE NAMED COLLECTION']	NAMED_COLLECTION	NAMED COLLECTION ADMIN
 NAMED COLLECTION ADMIN	['NAMED COLLECTION CONTROL']	NAMED_COLLECTION	ALL
 SET DEFINER	[]	USER_NAME	ALL
+TABLE ENGINE	['TABLE ENGINE']	TABLE_ENGINE	ALL
 SYSTEM SHUTDOWN	['SYSTEM KILL','SHUTDOWN']	GLOBAL	SYSTEM
 SYSTEM DROP DNS CACHE	['SYSTEM DROP DNS','DROP DNS CACHE','DROP DNS']	GLOBAL	SYSTEM DROP CACHE
 SYSTEM DROP MARK CACHE	['SYSTEM DROP MARK','DROP MARK CACHE','DROP MARKS']	GLOBAL	SYSTEM DROP CACHE