Add tests for V2 deploy with auth (#27729)

Author: mregrock

ydb/tests/functional/config/test_distconf.py

@@ -11,7 +11,6 @@
 import ydb.tests.library.common.cms as cms
 from ydb.tests.library.clients.kikimr_http_client import SwaggerClient
 from ydb.tests.library.harness.kikimr_runner import KiKiMR
-from ydb.tests.library.clients.kikimr_config_client import ConfigClient
 from ydb.tests.library.harness.kikimr_config import KikimrConfigGenerator
 from ydb.tests.library.kv.helpers import create_kv_tablets_and_wait_for_start
 from ydb.public.api.protos.ydb_status_codes_pb2 import StatusIds
@@ -46,6 +45,7 @@ class DistConfKiKiMRTest(object):
     use_config_store = True
     separate_node_configs = True
     nodes_count = 0
+    protected_mode = False
     metadata_section = {
         "kind": "MainConfig",
         "version": 0,
@@ -73,17 +73,17 @@ def setup_class(cls):
             separate_node_configs=cls.separate_node_configs,
             simple_config=True,
             use_self_management=True,
+            protected_mode=cls.protected_mode,
             extra_grpc_services=['config'],
             additional_log_configs=log_configs)
 
         cls.cluster = KiKiMR(configurator=cls.configurator)
         cls.cluster.start()
 
-        cms.request_increase_ratio_limit(cls.cluster.client)
+        if not cls.protected_mode:
+            cms.request_increase_ratio_limit(cls.cluster.client)
         host = cls.cluster.nodes[1].host
-        grpc_port = cls.cluster.nodes[1].port
         cls.swagger_client = SwaggerClient(host, cls.cluster.nodes[1].mon_port)
-        cls.config_client = ConfigClient(host, grpc_port)
 
     @classmethod
     def teardown_class(cls):
@@ -135,7 +135,7 @@ def test_cluster_expand_with_distconf(self):
 
         node_port_allocator = self.configurator.port_allocator.get_node_port_allocator(expected_new_node_id)
 
-        fetched_config = fetch_config(self.config_client)
+        fetched_config = fetch_config(self.cluster.config_client)
         dumped_fetched_config = yaml.safe_load(fetched_config)
         config_section = dumped_fetched_config["config"]
 
@@ -170,7 +170,7 @@ def test_cluster_expand_with_distconf(self):
         dumped_fetched_config["metadata"]["version"] = 1
 
         # replace config
-        replace_config_response = self.config_client.replace_config(yaml.dump(dumped_fetched_config))
+        replace_config_response = self.cluster.config_client.replace_config(yaml.dump(dumped_fetched_config))
         logger.debug(f"replace_config_response: {replace_config_response}")
         assert_that(replace_config_response.operation.status == StatusIds.SUCCESS)
         # start new node
@@ -219,7 +219,7 @@ def test_cluster_expand_with_seed_nodes(self):
 
         node_port_allocator = self.configurator.port_allocator.get_node_port_allocator(expected_new_node_id)
 
-        fetched_config = fetch_config(self.config_client)
+        fetched_config = fetch_config(self.cluster.config_client)
         dumped_fetched_config = yaml.safe_load(fetched_config)
         config_section = dumped_fetched_config["config"]
 
@@ -264,7 +264,7 @@ def test_cluster_expand_with_seed_nodes(self):
         dumped_fetched_config["metadata"]["version"] = 1
 
         # replace config
-        replace_config_response = self.config_client.replace_config(yaml.dump(dumped_fetched_config))
+        replace_config_response = self.cluster.config_client.replace_config(yaml.dump(dumped_fetched_config))
         logger.debug(f"replace_config_response: {replace_config_response}")
         assert_that(replace_config_response.operation.status == StatusIds.SUCCESS)
         # start new node
@@ -299,24 +299,24 @@ def test_cluster_expand_with_seed_nodes(self):
                 os.unlink(seed_nodes_file.name)
 
     def test_invalid_host_config_id(self):
-        fetched_config = fetch_config(self.config_client)
+        fetched_config = fetch_config(self.cluster.config_client)
         dumped_fetched_config = yaml.safe_load(fetched_config)
 
         # replace config with invalid host config id
         dumped_fetched_config['metadata']['version'] = 1
         dumped_fetched_config["config"]["host_configs"][0]["host_config_id"] = 1000
-        replace_config_response = self.config_client.replace_config(yaml.dump(dumped_fetched_config))
+        replace_config_response = self.cluster.config_client.replace_config(yaml.dump(dumped_fetched_config))
         logger.debug(f"replace_config_response: {replace_config_response}")
         assert_that(replace_config_response.operation.status == StatusIds.INTERNAL_ERROR)
 
     def test_invalid_change_host_config_disk(self):
-        fetched_config = fetch_config(self.config_client)
+        fetched_config = fetch_config(self.cluster.config_client)
         dumped_fetched_config = yaml.safe_load(fetched_config)
 
         # replace config with invalid host config disk path
         dumped_fetched_config["config"]["host_configs"][0]["drive"].append(dumped_fetched_config["config"]["host_configs"][0]["drive"][0])
         dumped_fetched_config['metadata']['version'] = 1
-        replace_config_response = self.config_client.replace_config(yaml.dump(dumped_fetched_config))
+        replace_config_response = self.cluster.config_client.replace_config(yaml.dump(dumped_fetched_config))
         logger.debug(f"replace_config_response: {replace_config_response}")
         assert_that(replace_config_response.operation.status == StatusIds.INTERNAL_ERROR)
 
@@ -348,3 +348,18 @@ def test_bootstrap_selector_validation(self):
         with pytest.raises(Exception) as ei:
             cluster.start()
         assert 'YAML validation failed' in str(ei.value)
+
+
+class TestDistConfWithAuth(DistConfKiKiMRTest):
+    protected_mode = True
+
+    def test_auth_v2_initialization(self):
+        table_path = '/Root/mydb/mytable_with_auth'
+        self.cluster.create_database(
+            table_path,
+            storage_pool_units_count={
+                'rot': 1
+            },
+            timeout_seconds=60,
+            token=self.cluster.root_token
+        )

ydb/tests/library/clients/kikimr_config_client.py

@@ -14,24 +14,30 @@
 logger = logging.getLogger()
 
 
-def config_client_factory(server, port, cluster=None, retry_count=1):
+def config_client_factory(server, port, cluster=None, retry_count=1, ca_path=None, cert_path=None, key_path=None):
     return ConfigClient(
         server, port, cluster=cluster,
-        retry_count=retry_count
+        retry_count=retry_count,
+        ca_path=ca_path, cert_path=cert_path, key_path=key_path,
     )
 
 
 def channels_list():
     return os.getenv('CHANNELS_LIST', '')
 
 
+def read_file(path):
+    with open(path, 'rb') as f:
+        return f.read()
+
+
 class ConfigClient(object):
     class FetchTransform(Enum):
         NONE = 1
         DETACH_STORAGE_CONFIG_SECTION = 2
         ATTACH_STORAGE_CONFIG_SECTION = 3
 
-    def __init__(self, server, port, cluster=None, retry_count=1):
+    def __init__(self, server, port, cluster=None, retry_count=1, ca_path=None, cert_path=None, key_path=None):
         self.server = server
         self.port = port
         self._cluster = cluster
@@ -42,7 +48,17 @@ def __init__(self, server, port, cluster=None, retry_count=1):
             ('grpc.max_receive_message_length', 64 * 10 ** 6),
             ('grpc.max_send_message_length', 64 * 10 ** 6)
         ]
-        self._channel = grpc.insecure_channel("%s:%s" % (self.server, self.port), options=self._options)
+
+        if cert_path and key_path:
+            credentials = grpc.ssl_channel_credentials(
+                root_certificates=read_file(ca_path) if ca_path else None,
+                private_key=read_file(key_path),
+                certificate_chain=read_file(cert_path)
+            )
+            self._channel = grpc.secure_channel("%s:%s" % (self.server, self.port), credentials, options=self._options)
+        else:
+            self._channel = grpc.insecure_channel("%s:%s" % (self.server, self.port), options=self._options)
+
         self._stub = grpc_server.ConfigServiceStub(self._channel)
         self._auth_token = None
 

ydb/tests/library/harness/kikimr_cluster_interface.py

@@ -8,7 +8,6 @@
 from ydb.tests.library.clients.kikimr_client import kikimr_client_factory
 from ydb.tests.library.clients.kikimr_keyvalue_client import keyvalue_client_factory
 from ydb.tests.library.clients.kikimr_scheme_client import scheme_client_factory
-from ydb.tests.library.clients.kikimr_config_client import config_client_factory
 from ydb.tests.library.common.protobuf_console import (
     CreateTenantRequest, AlterTenantRequest, GetTenantStatusRequest,
     RemoveTenantRequest, GetOperationRequest)
@@ -115,14 +114,16 @@ def scheme_client(self):
     @property
     def config_client(self):
         if self.__config_client is None:
-            self.__config_client = config_client_factory(
-                server=self.nodes[1].host,
-                port=self.nodes[1].grpc_port,
-                cluster=self,
-                retry_count=10,
-            )
+            self.__config_client = self._create_config_client()
         return self.__config_client
 
+    @abc.abstractmethod
+    def _create_config_client(self):
+        """
+        Factory method for ConfigClient, must be implemented by subclasses.
+        """
+        pass
+
     def _send_get_tenant_status_request(self, database_name, token=None):
         req = GetTenantStatusRequest(database_name)
 

ydb/tests/library/harness/kikimr_config.py

@@ -186,7 +186,8 @@ def __init__(
             verbose_memory_limit_exception=False,
             enable_static_auth=False,
             cms_config=None,
-            explicit_statestorage_config=None
+            explicit_statestorage_config=None,
+            protected_mode=False
     ):
         if extra_feature_flags is None:
             extra_feature_flags = []
@@ -213,7 +214,8 @@ def __init__(
         self.app_config = config_pb2.TAppConfig()
         self.port_allocator = KikimrPortManagerPortAllocator() if port_allocator is None else port_allocator
         erasure = Erasure.NONE if erasure is None else erasure
-        self.__grpc_ssl_enable = grpc_ssl_enable
+        self.protected_mode = protected_mode
+        self.__grpc_ssl_enable = grpc_ssl_enable or protected_mode
         self.__grpc_tls_data_path = None
         self.__grpc_tls_ca = None
         self.__grpc_tls_key = None
@@ -453,12 +455,52 @@ def __init__(
         if os.getenv("YDB_ALLOW_ORIGIN") is not None:
             self.yaml_config["monitoring_config"] = {"allow_origin": str(os.getenv("YDB_ALLOW_ORIGIN"))}
 
-        if enforce_user_token_requirement:
+        if enforce_user_token_requirement or protected_mode:
             security_config_root["security_config"]["enforce_user_token_requirement"] = True
 
         if default_user_sid:
             security_config_root["security_config"]["default_user_sids"] = [default_user_sid]
 
+        # protected mode is described in the YDB documentation for cluster deployment: it uses both certificate and token authentication.
+        # see https://ydb.tech/docs/en/devops/deployment-options/manual/initial-deployment?version=main
+        if protected_mode:
+            security_config = security_config_root.setdefault("security_config", {})
+            if "default_users" in security_config:
+                del security_config["default_users"]
+
+            base_sids = ["root", "root@builtin", "ADMINS", "DATABASE-ADMINS", "clusteradmins@cert"]
+            security_config["monitoring_allowed_sids"] = base_sids
+            security_config["viewer_allowed_sids"] = base_sids
+            security_config["bootstrap_allowed_sids"] = base_sids
+            security_config["administration_allowed_sids"] = base_sids
+
+            self.yaml_config["interconnect_config"] = {
+                "start_tcp": True,
+                "encryption_mode": "OPTIONAL",
+                "path_to_certificate_file": self.grpc_tls_cert_path,
+                "path_to_private_key_file": self.grpc_tls_key_path,
+                "path_to_ca_file": self.grpc_tls_ca_path,
+            }
+
+            self.yaml_config['grpc_config']['services_enabled'] = ['legacy']
+            if 'services' in self.yaml_config['grpc_config']:
+                del self.yaml_config['grpc_config']['services']
+
+            self.yaml_config['client_certificate_authorization'] = {
+                "request_client_certificate": True,
+                "client_certificate_definitions": [
+                    {
+                        "member_groups": ["clusteradmins@cert"],
+                        "subject_terms": [
+                            {
+                                "short_name": "O",
+                                "values": ["YDB"]
+                            }
+                        ]
+                    }
+                ]
+            }
+
         if memory_controller_config:
             self.yaml_config["memory_controller_config"] = memory_controller_config
 
@@ -528,14 +570,19 @@ def __init__(
             self._add_host_config_and_hosts()
             self.yaml_config.pop("nameservice_config")
         if self.use_self_management:
+
             if "security_config" in self.yaml_config["domains_config"]:
-                self.yaml_config["domains_config"].pop("security_config")
+                self.yaml_config["security_config"] = self.yaml_config["domains_config"].pop("security_config")
+
+            if "services" in self.yaml_config["grpc_config"] and "config" not in self.yaml_config["grpc_config"]["services"]:
+                self.yaml_config["grpc_config"]["services"].append("config")
+
             self.yaml_config["default_disk_type"] = "ROT"
             self.yaml_config["fail_domain_type"] = "rack"
             self.yaml_config["erasure"] = self.yaml_config.pop("static_erasure")
 
-            for name in ['blob_storage_config', 'domains_config', 'system_tablets', 'grpc_config',
-                         'channel_profile_config', 'interconnect_config']:
+            for name in ['blob_storage_config', 'domains_config', 'system_tablets',
+                         'channel_profile_config']:
                 del self.yaml_config[name]
         if self.simple_config:
             self.yaml_config.pop("feature_flags")