Fix database restore failure when there are two database admins (#16958)

Author: pixcc

ydb/apps/ydb/CHANGELOG.md

@@ -6,6 +6,7 @@
 * Support coordination nodes in `ydb scheme rmdir --recursive`.
 * Fixed return code of command `ydb workload * run --check-canonical` for the case when benchmark query results differ from canonical ones.
 * Fixed scheme error in `ydb admin cluster dump` when specifying a domain database.
+* Fixed unauthorized error in `ydb admin database restore` when multiple database admins are in dump.
 
 ## 2.20.0 ##
 

ydb/public/lib/ydb_cli/dump/restore_impl.cpp

@@ -635,7 +635,10 @@ TRestoreResult TRestoreClient::RestoreUsers(TTableClient& client, const TFsPath&
             auto alterStatementResult = client.RetryOperationSync([&](TSession session) {
                 return session.ExecuteSchemeQuery(alterStatement).ExtractValueSync();
             });
-            if (!alterStatementResult.IsSuccess()) {
+            if (alterStatementResult.GetStatus() == EStatus::UNAUTHORIZED) {
+                LOG_W("Not enough rights to restore user from statement " << alterStatement.Quote() << ", skipping");
+                continue;
+            } else if (!alterStatementResult.IsSuccess()) {
                 LOG_E("Failed to execute statement for restoring user: "
                     << alterStatement.Quote() << ", error: "
                     << alterStatementResult.GetIssues().ToOneLineString());

ydb/tests/functional/ydb_cli/test_ydb_backup.py

@@ -1306,6 +1306,8 @@ def setup_class(cls):
                 "enable_database_admin"
             ],
             domain_login_only=False,
+            enforce_user_token_requirement=True,
+            default_clusteradmin="root@builtin",
         ))
         cls.cluster.start()
 
@@ -1317,15 +1319,17 @@ def setup_class(cls):
             storage_pool_units_count={
                 'hdd': 1
             },
-            timeout_seconds=100
+            timeout_seconds=100,
+            token=cls.cluster.config.default_clusteradmin
         )
 
         cls.database_nodes = cls.cluster.register_and_start_slots(cls.database, count=3)
-        cls.cluster.wait_tenant_up(cls.database)
+        cls.cluster.wait_tenant_up(cls.database, cls.cluster.config.default_clusteradmin)
 
         driver_config = ydb.DriverConfig(
             database=cls.database,
-            endpoint="%s:%s" % (cls.cluster.nodes[1].host, cls.cluster.nodes[1].port))
+            endpoint="%s:%s" % (cls.cluster.nodes[1].host, cls.cluster.nodes[1].port),
+            credentials=ydb.AuthTokenCredentials(cls.cluster.config.default_clusteradmin))
         cls.driver = ydb.Driver(driver_config)
         cls.driver.wait(timeout=4)
 
@@ -1340,8 +1344,8 @@ def set_test_name(cls, request):
         cls.test_name = request.node.name
 
     @classmethod
-    def create_backup(cls, command, expected_files, additional_args=[]):
-        backup_files_dir = output_path(cls.test_name, "backup_files_dir")
+    def create_backup(cls, command, expected_files, output, additional_args=[], token=""):
+        backup_files_dir = output_path(cls.test_name, output)
         execution = yatest.common.execute(
             [
                 backup_bin(),
@@ -1350,7 +1354,8 @@ def create_backup(cls, command, expected_files, additional_args=[]):
             ]
             + command
             + ["--output", backup_files_dir]
-            + additional_args
+            + additional_args,
+            env={"YDB_TOKEN": token}
         )
 
         list_result = fs_recursive_list(backup_files_dir, ListMode.FILES)
@@ -1370,32 +1375,59 @@ def create_backup(cls, command, expected_files, additional_args=[]):
         )
 
     @classmethod
-    def create_cluster_backup(cls, expected_files, additional_args=[]):
+    def create_cluster_backup(cls, expected_files, output="backup_files_dir", additional_args=[]):
         cls.create_backup(
             [
                 "--database", cls.root_dir,
                 "admin", "cluster", "dump",
             ],
             expected_files,
-            additional_args
+            output,
+            additional_args,
+            token=cls.cluster.config.default_clusteradmin,
         )
 
     @classmethod
-    def create_database_backup(cls, expected_files, additional_args=[]):
+    def create_database_backup(cls, expected_files, output="backup_files_dir", additional_args=[]):
         cls.create_backup(
             [
                 "--database", cls.database,
+                "--user", "dbadmin1", "--no-password",
                 "admin", "database", "dump",
             ],
             expected_files,
+            output,
             additional_args
         )
 
+    @classmethod
+    def setup_sample_data(cls):
+        pool = ydb.SessionPool(cls.driver)
+        with pool.checkout() as session:
+            session.execute_scheme("CREATE GROUP people")
+            session.execute_scheme("CREATE USER alice PASSWORD '1234'")
+            session.execute_scheme("CREATE USER bob PASSWORD '1234'")
+            session.execute_scheme("ALTER GROUP people ADD USER alice")
+            session.execute_scheme("ALTER GROUP people ADD USER bob")
+
+            session.execute_scheme("CREATE GROUP dbadmins")
+            session.execute_scheme("CREATE USER dbadmin1")
+            session.execute_scheme("CREATE USER dbadmin2 PASSWORD '1234'")
+            session.execute_scheme("ALTER GROUP dbadmins ADD USER dbadmin1")
+            session.execute_scheme("ALTER GROUP dbadmins ADD USER dbadmin2")
+            session.execute_scheme(f'GRANT "ydb.generic.use" on `{cls.database}` TO dbadmins')
+
+            cls.driver.scheme_client.modify_permissions(
+                cls.database,
+                ydb.ModifyPermissionsSettings().change_owner("dbadmins")
+            )
+
+        create_table_with_data(cls.driver.table_client.session().create(), "db1/table")
+
 
 class TestClusterBackup(BaseTestClusterBackupInFiles):
     def test_cluster_backup(self):
-        session = self.driver.table_client.session().create()
-        create_table_with_data(session, "db1/table")
+        self.setup_sample_data()
 
         self.create_cluster_backup(expected_files=[
             # cluster metadata
@@ -1415,8 +1447,7 @@ def test_cluster_backup(self):
 
 class TestDatabaseBackup(BaseTestClusterBackupInFiles):
     def test_database_backup(self):
-        session = self.driver.table_client.session().create()
-        create_table_with_data(session, "db1/table")
+        self.setup_sample_data()
 
         self.create_database_backup(expected_files=[
             # database metadata
@@ -1431,3 +1462,162 @@ def test_database_backup(self):
             "table/permissions.pb",
             "table/data_00.csv",
         ])
+
+
+class BaseTestMultipleClusterBackupInFiles(BaseTestClusterBackupInFiles):
+    @classmethod
+    def setup_class(cls):
+        super().setup_class()
+
+        cfg = KikimrConfigGenerator(
+            extra_feature_flags=[
+                "enable_strict_acl_check",
+                "enable_strict_user_management",
+                "enable_database_admin"
+            ],
+            domain_name="Root2",
+            domain_login_only=False,
+            enforce_user_token_requirement=True,
+            default_clusteradmin="root@builtin",
+        )
+
+        cls.restore_cluster = KiKiMR(
+            cfg,
+            cluster_name="restore_cluster"
+        )
+
+        cls.restore_cluster.start()
+
+        cls.restore_root_dir = "/Root2"
+        cls.restore_database = os.path.join(cls.restore_root_dir, "db1")
+        cls.restore_database_nodes = cls.restore_cluster.register_and_start_slots(
+            cls.restore_database,
+            count=1
+        )
+
+        restore_driver_config = ydb.DriverConfig(
+            database=cls.restore_database,
+            endpoint="%s:%s" % (
+                cls.restore_cluster.nodes[1].host,
+                cls.restore_cluster.nodes[1].port
+            ),
+            credentials=ydb.AuthTokenCredentials(cls.cluster.config.default_clusteradmin),
+        )
+        cls.restore_driver = ydb.Driver(restore_driver_config)
+
+    @classmethod
+    def teardown_class(cls):
+        cls.restore_cluster.unregister_and_stop_slots(cls.restore_database_nodes)
+        cls.restore_cluster.stop()
+        super().teardown_class()
+
+    @classmethod
+    def restore(cls, command, input, additional_args=[], token=""):
+        backup_files_dir = os.path.join(
+            yatest.common.output_path(),
+            cls.test_name,
+            input
+        )
+
+        yatest.common.execute(
+            [
+                backup_bin(),
+                "--verbose",
+                "--assume-yes",
+                "--endpoint", f"grpc://localhost:{cls.restore_cluster.nodes[1].grpc_port}",
+            ]
+            + command
+            + ["--input", backup_files_dir, "-w", "60s"]
+            + additional_args,
+            env={"YDB_TOKEN": token}
+        )
+
+    @classmethod
+    def restore_cluster_backup(cls, input="backup_files_dir", additional_args=[]):
+        cls.restore(
+            [
+                "--database", cls.restore_root_dir,
+                "admin", "cluster", "restore"
+            ],
+            input,
+            additional_args,
+            token=cls.restore_cluster.config.default_clusteradmin,
+        )
+
+    @classmethod
+    def restore_database_backup(cls, input="backup_files_dir", additional_args=[]):
+        cls.restore(
+            [
+                "--database", cls.restore_database,
+                "--user", "dbadmin1", "--no-password",
+                "admin", "database", "restore"
+            ],
+            input,
+            additional_args
+        )
+
+
+class TestClusterBackupRestore(BaseTestMultipleClusterBackupInFiles):
+    def test_cluster_backup_restore(self):
+        self.setup_sample_data()
+
+        self.create_cluster_backup(expected_files=[
+            # cluster metadata
+            "permissions.pb",
+            "create_user.sql",
+            "create_group.sql",
+            "alter_group.sql",
+
+            # database metadata
+            "Root/db1/database.pb",
+            "Root/db1/permissions.pb",
+            "Root/db1/create_user.sql",
+            "Root/db1/create_group.sql",
+            "Root/db1/alter_group.sql",
+        ])
+
+        self.restore_cluster_backup()
+
+        self.restore_cluster.wait_tenant_up(self.restore_database, self.cluster.config.default_clusteradmin)
+        self.restore_driver.wait(timeout=10)
+
+
+class TestDatabaseBackupRestore(BaseTestMultipleClusterBackupInFiles):
+    def test_database_backup_restore(self):
+        self.setup_sample_data()
+
+        self.create_cluster_backup(output="cluster_backup", expected_files=[
+            # cluster metadata
+            "permissions.pb",
+            "create_user.sql",
+            "create_group.sql",
+            "alter_group.sql",
+
+            # database metadata
+            "Root/db1/database.pb",
+            "Root/db1/permissions.pb",
+            "Root/db1/create_user.sql",
+            "Root/db1/create_group.sql",
+            "Root/db1/alter_group.sql",
+        ])
+
+        self.create_database_backup(output="database_backup", expected_files=[
+            # database metadata
+            "database.pb",
+            "permissions.pb",
+            "create_user.sql",
+            "create_group.sql",
+            "alter_group.sql",
+
+            # database table
+            "table/scheme.pb",
+            "table/permissions.pb",
+            "table/data_00.csv",
+        ])
+
+        self.restore_cluster_backup(input="cluster_backup")
+
+        self.restore_cluster.wait_tenant_up(self.restore_database, self.cluster.config.default_clusteradmin)
+        self.restore_driver.wait(timeout=10)
+
+        self.restore_database_backup(input="database_backup")