tests: support testenv setup in non-anonymous mode (#15130)

Add support for fully authenticated setup operations to `tests/library/`.
Before that library could only execute cluster setup, configuration and database manipulation in anonymous mode or in the mode when every user is a cluster admin.
Now `tests/library/` can operate when `administration_allowed_lists` is not empty and `enforce_user_token_requirement=True`.

Author: ijon

ydb/tests/library/clients/kikimr_client.py

@@ -254,8 +254,10 @@ def console_request(self, request_text, raise_on_error=True):
             raise RuntimeError('console_request failed: %s: %s' % (response.Status.Code, response.Status.Reason))
         return response
 
-    def add_config_item(self, config, cookie=None, raise_on_error=True):
+    def add_config_item(self, config, cookie=None, raise_on_error=True, token=None):
         request = msgbus.TConsoleRequest()
+        if token is not None:
+            request.SecurityToken = token
         action = request.ConfigureRequest.Actions.add()
         item = action.AddConfigItem.ConfigItem
         if isinstance(config, str) or isinstance(config, bytes):

ydb/tests/library/common/protobuf_console.py

@@ -2,7 +2,7 @@
 # -*- coding: utf-8 -*-
 import ydb.core.protos.msgbus_pb2 as msgbus
 
-from ydb.tests.library.common.protobuf import AbstractProtobufBuilder
+from ydb.tests.library.common.protobuf import AbstractProtobufBuilder, to_bytes
 
 
 class CreateTenantRequest(AbstractProtobufBuilder):
@@ -14,6 +14,10 @@ def __init__(self, path):
         super(CreateTenantRequest, self).__init__(msgbus.TConsoleRequest())
         self.protobuf.CreateTenantRequest.Request.path = path
 
+    def set_user_token(self, token):
+        self.protobuf.SecurityToken = to_bytes(token)
+        self.protobuf.CreateTenantRequest.UserToken = to_bytes(token)
+
     def add_storage_pool(self, pool_type, pool_size):
         pool = self.protobuf.CreateTenantRequest.Request.resources.storage_units.add()
         pool.unit_kind = pool_type
@@ -78,6 +82,10 @@ def __init__(self, path):
         super(AlterTenantRequest, self).__init__(msgbus.TConsoleRequest())
         self.protobuf.AlterTenantRequest.Request.path = path
 
+    def set_user_token(self, token):
+        self.protobuf.SecurityToken = to_bytes(token)
+        self.protobuf.AlterTenantRequest.UserToken = to_bytes(token)
+
     def set_schema_quotas(self, schema_quotas):
         quotas = self.protobuf.AlterTenantRequest.Request.schema_operation_quotas
         quotas.SetInParent()
@@ -106,6 +114,10 @@ def __init__(self, path):
         super(GetTenantStatusRequest, self).__init__(msgbus.TConsoleRequest())
         self.protobuf.GetTenantStatusRequest.Request.path = path
 
+    def set_user_token(self, token):
+        self.protobuf.SecurityToken = to_bytes(token)
+        self.protobuf.GetTenantStatusRequest.UserToken = to_bytes(token)
+
 
 class RemoveTenantRequest(AbstractProtobufBuilder):
     """
@@ -116,6 +128,10 @@ def __init__(self, path):
         super(RemoveTenantRequest, self).__init__(msgbus.TConsoleRequest())
         self.protobuf.RemoveTenantRequest.Request.path = path
 
+    def set_user_token(self, token):
+        self.protobuf.SecurityToken = to_bytes(token)
+        self.protobuf.RemoveTenantRequest.UserToken = to_bytes(token)
+
 
 class SetConfigRequest(AbstractProtobufBuilder):
     """
@@ -158,3 +174,6 @@ class GetOperationRequest(AbstractProtobufBuilder):
     def __init__(self, op_id):
         super(GetOperationRequest, self).__init__(msgbus.TConsoleRequest())
         self.protobuf.GetOperationRequest.id = op_id
+
+    def set_user_token(self, token):
+        self.protobuf.SecurityToken = to_bytes(token)

ydb/tests/library/harness/kikimr_cluster_interface.py

@@ -110,35 +110,41 @@ def scheme_client(self):
             )
         return self.__scheme_client
 
-    def get_database_status(self, database_name):
-        response = self.client.send_request(
-            GetTenantStatusRequest(database_name).protobuf,
-            method='ConsoleRequest'
-        ).GetTenantStatusResponse
+    def _send_get_tenant_status_request(self, database_name, token=None):
+        req = GetTenantStatusRequest(database_name)
+
+        if token is not None:
+            req.set_user_token(token)
+
+        return self.client.send_request(req.protobuf, method='ConsoleRequest').GetTenantStatusResponse
+
+    def get_database_status(self, database_name, token=None):
+        response = self._send_get_tenant_status_request(database_name, token=token)
 
         if response.Response.operation.status != StatusIds.SUCCESS:
             logger.critical("Console response status: %s", str(response.Response.operation.status))
             assert False
-            return False
 
         result = cms_tenants_pb.GetDatabaseStatusResult()
         response.Response.operation.result.Unpack(result)
         return result
 
-    def wait_tenant_up(self, database_name):
+    def wait_tenant_up(self, database_name, token=None):
         self.__wait_tenant_up(
             database_name,
-            expected_computational_units=1
+            expected_computational_units=1,
+            token=token,
         )
 
     def __wait_tenant_up(
             self,
             database_name,
             expected_computational_units=None,
-            timeout_seconds=120
+            timeout_seconds=120,
+            token=None
     ):
         def predicate():
-            result = self.get_database_status(database_name)
+            result = self.get_database_status(database_name, token=token)
 
             if expected_computational_units is None:
                 expected = set([2])
@@ -154,21 +160,25 @@ def predicate():
         )
         assert tenant_running
 
-    def __get_console_op(self, op_id):
+    def __get_console_op(self, op_id, token=None):
         req = GetOperationRequest(op_id)
+
+        if token is not None:
+            req.set_user_token(token)
+
         response = self.client.send_request(req.protobuf, method='ConsoleRequest')
         operation = response.GetOperationResponse.operation
         if not operation.ready and response.Status.Code != StatusIds.STATUS_CODE_UNSPECIFIED:
             raise RuntimeError('get_console_op failed: %s: %s' % (response.Status.Code, response.Status.Reason))
         return operation
 
-    def __wait_console_op(self, op_id, timeout_seconds, step_seconds=0.5):
+    def __wait_console_op(self, op_id, timeout_seconds, step_seconds=0.5, token=None):
         deadline = time.time() + timeout_seconds
         while True:
             time.sleep(step_seconds)
             if time.time() >= deadline:
                 raise RuntimeError('wait_console_op: deadline exceeded')
-            operation = self.__get_console_op(op_id)
+            operation = self.__get_console_op(op_id, token=token)
             if operation.ready:
                 return operation
 
@@ -177,7 +187,8 @@ def create_database(
             database_name,
             storage_pool_units_count,
             disable_external_subdomain=False,
-            timeout_seconds=120
+            timeout_seconds=120,
+            token=None,
     ):
         req = CreateTenantRequest(database_name)
         for storage_pool_type_name, units_count in storage_pool_units_count.items():
@@ -189,27 +200,32 @@ def create_database(
         if disable_external_subdomain:
             req.disable_external_subdomain()
 
+        if token is not None:
+            req.set_user_token(token)
+
         response = self.client.send_request(req.protobuf, method='ConsoleRequest')
         operation = response.CreateTenantResponse.Response.operation
         if not operation.ready and response.Status.Code != StatusIds.STATUS_CODE_UNSPECIFIED:
             raise RuntimeError('create_database failed: %s: %s' % (response.Status.Code, response.Status.Reason))
         if not operation.ready:
-            operation = self.__wait_console_op(operation.id, timeout_seconds=timeout_seconds)
+            operation = self.__wait_console_op(operation.id, timeout_seconds=timeout_seconds, token=token)
         if operation.status != StatusIds.SUCCESS:
             raise RuntimeError('create_database failed: %s, %s' % (operation.status, ydb.issues._format_issues(operation.issues)))
 
         self.__wait_tenant_up(
             database_name,
             expected_computational_units=0,
-            timeout_seconds=timeout_seconds
+            timeout_seconds=timeout_seconds,
+            token=token,
         )
         return database_name
 
     def create_hostel_database(
             self,
             database_name,
             storage_pool_units_count,
-            timeout_seconds=120
+            timeout_seconds=120,
+            token=None,
     ):
         req = CreateTenantRequest(database_name)
         for storage_pool_type_name, units_count in storage_pool_units_count.items():
@@ -218,19 +234,23 @@ def create_hostel_database(
                 units_count,
             )
 
+        if token is not None:
+            req.set_user_token(token)
+
         response = self.client.send_request(req.protobuf, method='ConsoleRequest')
         operation = response.CreateTenantResponse.Response.operation
         if not operation.ready and response.Status.Code != StatusIds.STATUS_CODE_UNSPECIFIED:
             raise RuntimeError('create_hostel_database failed: %s: %s' % (response.Status.Code, response.Status.Reason))
         if not operation.ready:
-            operation = self.__wait_console_op(operation.id, timeout_seconds=timeout_seconds)
+            operation = self.__wait_console_op(operation.id, timeout_seconds=timeout_seconds, token=token)
         if operation.status != StatusIds.SUCCESS:
             raise RuntimeError('create_hostel_database failed: %s' % (operation.status,))
 
         self.__wait_tenant_up(
             database_name,
             expected_computational_units=0,
-            timeout_seconds=timeout_seconds
+            timeout_seconds=timeout_seconds,
+            token=token,
         )
         return database_name
 
@@ -241,10 +261,14 @@ def create_serverless_database(
             timeout_seconds=120,
             schema_quotas=None,
             disk_quotas=None,
-            attributes=None
+            attributes=None,
+            token=None,
     ):
         req = CreateTenantRequest(database_name)
 
+        if token is not None:
+            req.set_user_token(token)
+
         req.share_resources_with(hostel_db)
 
         if schema_quotas is not None:
@@ -262,13 +286,14 @@ def create_serverless_database(
         if not operation.ready and response.Status.Code != StatusIds.STATUS_CODE_UNSPECIFIED:
             raise RuntimeError('create_serverless_database failed: %s: %s' % (response.Status.Code, response.Status.Reason))
         if not operation.ready:
-            operation = self.__wait_console_op(operation.id, timeout_seconds=timeout_seconds)
+            operation = self.__wait_console_op(operation.id, timeout_seconds=timeout_seconds, token=token)
         if operation.status != StatusIds.SUCCESS:
             raise RuntimeError('create_serverless_database failed: %s' % (operation.status,))
 
         self.__wait_tenant_up(
             database_name,
-            timeout_seconds=timeout_seconds
+            timeout_seconds=timeout_seconds,
+            token=token,
         )
         return database_name
 
@@ -278,9 +303,13 @@ def alter_serverless_database(
             schema_quotas=None,
             disk_quotas=None,
             timeout_seconds=120,
+            token=None,
     ):
         req = AlterTenantRequest(database_name)
 
+        if token is not None:
+            req.set_user_token(token)
+
         assert schema_quotas is not None or disk_quotas is not None
 
         if schema_quotas is not None:
@@ -294,33 +323,39 @@ def alter_serverless_database(
         if not operation.ready and response.Status.Code != StatusIds.STATUS_CODE_UNSPECIFIED:
             raise RuntimeError('alter_serverless_database failed: %s: %s' % (response.Status.Code, response.Status.Reason))
         if not operation.ready:
-            operation = self.__wait_console_op(operation.id, timeout_seconds=timeout_seconds)
+            operation = self.__wait_console_op(operation.id, timeout_seconds=timeout_seconds, token=token)
         if operation.status != StatusIds.SUCCESS:
             raise RuntimeError('alter_serverless_database failed: %s' % (operation.status,))
 
         self.__wait_tenant_up(
             database_name,
-            timeout_seconds=timeout_seconds
+            timeout_seconds=timeout_seconds,
+            token=token,
         )
         return database_name
 
     def remove_database(
             self,
             database_name,
-            timeout_seconds=20
+            timeout_seconds=20,
+            token=None,
     ):
         logger.debug(database_name)
 
-        operation_id = self._remove_database_send_op(database_name)
-        self._remove_database_wait_op(database_name, operation_id, timeout_seconds=timeout_seconds)
-        self._remove_database_wait_tenant_gone(database_name, timeout_seconds=timeout_seconds)
+        operation_id = self._remove_database_send_op(database_name, token=token)
+        self._remove_database_wait_op(database_name, operation_id, timeout_seconds=timeout_seconds, token=token)
+        self._remove_database_wait_tenant_gone(database_name, timeout_seconds=timeout_seconds, token=token)
 
         return database_name
 
-    def _remove_database_send_op(self, database_name):
-        logger.debug('%s: send console operation', database_name)
+    def _remove_database_send_op(self, database_name, token=None):
+        logger.debug('%s: send console operation, token %s', database_name, token)
 
         req = RemoveTenantRequest(database_name)
+
+        if token is not None:
+            req.set_user_token(token)
+
         response = self.client.send_request(req.protobuf, method='ConsoleRequest')
         operation = response.RemoveTenantResponse.Response.operation
         logger.debug('%s: response from console: %s', database_name, response)
@@ -330,20 +365,19 @@ def _remove_database_send_op(self, database_name):
 
         return operation.id
 
-    def _remove_database_wait_op(self, database_name, operation_id, timeout_seconds=20):
+    def _remove_database_wait_op(self, database_name, operation_id, timeout_seconds=20, token=None):
         logger.debug('%s: wait console operation done', database_name)
-        operation = self.__wait_console_op(operation_id, timeout_seconds=timeout_seconds)
+        operation = self.__wait_console_op(operation_id, timeout_seconds=timeout_seconds, token=token)
         logger.debug('%s: console operation done', database_name)
 
         if operation.status not in (StatusIds.SUCCESS, StatusIds.NOT_FOUND):
             raise RuntimeError('remove_database failed: %s' % (operation.status,))
 
-    def _remove_database_wait_tenant_gone(self, database_name, timeout_seconds=20):
+    def _remove_database_wait_tenant_gone(self, database_name, timeout_seconds=20, token=None):
         logger.debug('%s: wait tenant gone', database_name)
 
         def predicate():
-            response = self.client.send_request(
-                GetTenantStatusRequest(database_name).protobuf, method='ConsoleRequest').GetTenantStatusResponse
+            response = self._send_get_tenant_status_request(database_name, token=token)
             return response.Response.operation.status == StatusIds.NOT_FOUND
 
         tenant_not_found = wait_for(

ydb/tests/library/harness/kikimr_config.py

@@ -164,6 +164,8 @@ def __init__(
             column_shard_config=None,
             use_config_store=False,
             separate_node_configs=False,
+            default_clusteradmin=None,
+            enable_resource_pools=None,
     ):
         if extra_feature_flags is None:
             extra_feature_flags = []
@@ -270,6 +272,8 @@ def __init__(
 
         # for faster shutdown: there is no reason to wait while tablets are drained before whole cluster is stopping
         self.yaml_config["feature_flags"]["enable_drain_on_shutdown"] = False
+        if enable_resource_pools is not None:
+            self.yaml_config["feature_flags"]["enable_resource_pools"] = enable_resource_pools
         for extra_feature_flag in extra_feature_flags:
             self.yaml_config["feature_flags"][extra_feature_flag] = True
         if enable_alter_database_create_hive_first:
@@ -457,6 +461,16 @@ def __init__(
         self.use_config_store = use_config_store
         self.separate_node_configs = separate_node_configs
 
+        self.__default_clusteradmin = default_clusteradmin
+        if self.__default_clusteradmin is not None:
+            security_config = self.yaml_config["domains_config"]["security_config"]
+            security_config.setdefault("administration_allowed_sids", []).append(self.__default_clusteradmin)
+            security_config.setdefault("default_access", []).append('+F:{}'.format(self.__default_clusteradmin))
+
+    @property
+    def default_clusteradmin(self):
+        return self.__default_clusteradmin
+
     @property
     def pdisks_info(self):
         return self._pdisks_info