Merge pull request ClickHouse#88742 from ClickHouse/backport/25.8/88513

Backport ClickHouse#88513 to 25.8: Keeper improvement: add config for checking node ACL on removal

Author: clickhouse-gh[bot]

src/Coordination/CoordinationSettings.cpp

@@ -69,7 +69,8 @@ namespace ErrorCodes
     DECLARE(UInt64, log_slow_total_threshold_ms, 5000, "Requests for which the total latency is larger than this settings will be logged", 0) \
     DECLARE(UInt64, log_slow_cpu_threshold_ms, 100, "Requests for which the CPU (preprocessing and processing) latency is larger than this settings will be logged", 0) \
     DECLARE(UInt64, log_slow_connection_operation_threshold_ms, 1000, "Log message if a certain operation took too long inside a single connection", 0) \
-    DECLARE(Bool, use_xid_64, false, "Enable 64-bit XID. It is disabled by default because of backward compatibility", 0)
+    DECLARE(Bool, use_xid_64, false, "Enable 64-bit XID. It is disabled by default because of backward compatibility", 0) \
+    DECLARE(Bool, check_node_acl_on_remove, false, "When trying to remove a node, check ACLs from both the node itself and the parent node. If disabled, default behaviour will be used where only ACL from the parent node is checked", 0) \
 
 DECLARE_SETTINGS_TRAITS(CoordinationSettingsTraits, LIST_OF_COORDINATION_SETTINGS)
 IMPLEMENT_SETTINGS_TRAITS(CoordinationSettingsTraits, LIST_OF_COORDINATION_SETTINGS)
@@ -277,6 +278,11 @@ void KeeperConfigurationAndSettings::dump(WriteBufferFromOwnString & buf) const
     write_int(coordination_settings[CoordinationSetting::log_slow_cpu_threshold_ms]);
     writeText("log_slow_connection_operation_threshold_ms=", buf);
     write_int(coordination_settings[CoordinationSetting::log_slow_connection_operation_threshold_ms]);
+
+    writeText("use_xid_64=", buf);
+    write_bool(coordination_settings[CoordinationSetting::use_xid_64]);
+    writeText("check_node_acl_on_remove=", buf);
+    write_bool(coordination_settings[CoordinationSetting::check_node_acl_on_remove]);
 }
 
 KeeperConfigurationAndSettingsPtr

src/Coordination/KeeperStorage.cpp

@@ -52,6 +52,7 @@ namespace DB
 namespace CoordinationSetting
 {
     extern const CoordinationSettingsUInt64 log_slow_cpu_threshold_ms;
+    extern const CoordinationSettingsBool check_node_acl_on_remove;
 }
 
 namespace ErrorCodes
@@ -1828,6 +1829,12 @@ processLocal(const Coordination::ZooKeeperGetRequest & zk_request, Storage & sto
 template <typename Storage>
 bool checkAuth(const Coordination::ZooKeeperRemoveRequest & zk_request, Storage & storage, int64_t session_id, bool is_local)
 {
+    if (auto check_node_acl = storage.keeper_context->getCoordinationSettings()[CoordinationSetting::check_node_acl_on_remove];
+        check_node_acl && !storage.checkACL(zk_request.getPath(), Coordination::ACL::Delete, session_id, is_local))
+    {
+        return false;
+    }
+
     return storage.checkACL(Coordination::parentNodePath(zk_request.getPath()), Coordination::ACL::Delete, session_id, is_local);
 }
 

tests/integration/test_keeper_remove_acl/__init__.py

@@ -0,0 +1,7 @@
+<clickhouse>
+    <keeper_server>
+        <coordination_settings>
+            <check_node_acl_on_remove>0</check_node_acl_on_remove>
+        </coordination_settings>
+    </keeper_server>
+</clickhouse>

tests/integration/test_keeper_remove_acl/configs/check_node_acl_on_remove.xml

@@ -0,0 +1,25 @@
+<clickhouse>
+    <keeper_server>
+        <tcp_port>9181</tcp_port>
+        <server_id>1</server_id>
+        <log_storage_path>/var/lib/clickhouse/coordination/log</log_storage_path>
+        <snapshot_storage_path>/var/lib/clickhouse/coordination/snapshots</snapshot_storage_path>
+
+        <coordination_settings>
+            <operation_timeout_ms>5000</operation_timeout_ms>
+            <session_timeout_ms>10000</session_timeout_ms>
+            <snapshot_distance>75</snapshot_distance>
+            <raft_logs_level>trace</raft_logs_level>
+        </coordination_settings>
+
+        <raft_configuration>
+            <server>
+                <id>1</id>
+                <hostname>node</hostname>
+                <port>9234</port>
+                <can_become_leader>true</can_become_leader>
+                <priority>3</priority>
+            </server>
+        </raft_configuration>
+    </keeper_server>
+</clickhouse>

tests/integration/test_keeper_remove_acl/configs/enable_keeper1.xml

@@ -0,0 +1,89 @@
+#!/usr/bin/env python3
+
+import pytest
+import logging
+
+import helpers.keeper_utils as keeper_utils
+from helpers.cluster import ClickHouseCluster
+
+from kazoo.security import make_digest_acl
+from kazoo.exceptions import NoAuthError
+
+
+cluster = ClickHouseCluster(__file__)
+node = cluster.add_instance(
+    "node",
+    main_configs=["configs/enable_keeper1.xml", "configs/check_node_acl_on_remove.xml"],
+    stay_alive=True,
+)
+
+
+@pytest.fixture(scope="module")
+def started_cluster():
+    try:
+        cluster.start()
+
+        yield cluster
+
+    finally:
+        cluster.shutdown()
+
+
+def wait_nodes():
+    keeper_utils.wait_nodes(cluster, [node])
+
+
+def get_fake_zk(nodename, timeout=30.0):
+    return keeper_utils.get_fake_zk(cluster, nodename, timeout=timeout)
+
+
+def stop_zk_connection(zk_conn):
+    zk_conn.stop()
+    zk_conn.close()
+
+
+def test_server_restart(started_cluster):
+    try:
+        wait_nodes()
+
+        node.stop_clickhouse()
+        node.replace_in_config(
+            "/etc/clickhouse-server/config.d/check_node_acl_on_remove.xml", "1", "0"
+        )
+        node.start_clickhouse()
+
+        def create_node_with_acl():
+            node_zk = get_fake_zk("node")
+            node_zk.add_auth("digest", "clickhouse:password")
+
+            if node_zk.exists("/test_acl_node"):
+                node_zk.delete("/test_acl_node")
+
+            acl = make_digest_acl("clickhouse", "password", all=True)
+            node_zk.create("/test_acl_node", b"test_data", acl=[acl])
+            stop_zk_connection(node_zk)
+
+        def delete_node():
+            node_zk = get_fake_zk("node")
+            node_zk.delete("/test_acl_node")
+            stop_zk_connection(node_zk)
+
+        create_node_with_acl()
+        delete_node()
+        node.stop_clickhouse()
+        node.replace_in_config(
+            "/etc/clickhouse-server/config.d/check_node_acl_on_remove.xml", "0", "1"
+        )
+        node.start_clickhouse()
+
+        create_node_with_acl()
+
+        with pytest.raises(NoAuthError):
+            delete_node()
+    finally:
+        try:
+            stop_zk_connection(
+                node_zk,
+            )
+        except:
+            pass