Support SASL authentication in AIOKafkaAdminClient (#890)

* Support SASL authentication in AIOKafkaAdminClient [#889]

* Add missing deps into README

* Added a foundation for admin sasl tests

* Bring back gssapi_consumer_factory() method

* Add forgotten await-s

* Add more admin tests

* Add CHANGES

* Swap method back to minimize diff

---------

Co-authored-by: Sergey Levitin <[email protected]>
Co-authored-by: Denis Otkidach <[email protected]>

Author: 3 people

CHANGES/889.bugfix

@@ -0,0 +1 @@
+Added SASL authentication support to `AIOKafkaAdminClient`.

README.rst

@@ -81,7 +81,7 @@ generate ssh keys for some tests.
 
 Setting up tests requirements (assuming you're within virtualenv on ubuntu 14.04+)::
 
-    sudo apt-get install -y libsnappy-dev libzstd-dev
+    sudo apt-get install -y libsnappy-dev libzstd-dev libkrb5-dev krb5-user
     make setup
 
 Running tests with coverage::

aiokafka/admin.py

@@ -81,7 +81,13 @@ def __init__(self, *, loop=None,
                  metadata_max_age_ms: int = 300000,
                  security_protocol: str = "PLAINTEXT",
                  ssl_context: Optional[SSLContext] = None,
-                 api_version: str = "auto"):
+                 api_version: str = "auto",
+                 sasl_mechanism: str = 'PLAIN',
+                 sasl_plain_username: Optional[str] = None,
+                 sasl_plain_password: Optional[str] = None,
+                 sasl_kerberos_service_name: str = 'kafka',
+                 sasl_kerberos_domain_name: Optional[str] = None,
+                 sasl_oauth_token_provider: Optional[str] = None):
         self._closed = False
         self._started = False
         self._version_info = {}
@@ -94,7 +100,13 @@ def __init__(self, *, loop=None,
             api_version=api_version,
             ssl_context=ssl_context,
             security_protocol=security_protocol,
-            connections_max_idle_ms=connections_max_idle_ms)
+            connections_max_idle_ms=connections_max_idle_ms,
+            sasl_mechanism=sasl_mechanism,
+            sasl_plain_username=sasl_plain_username,
+            sasl_plain_password=sasl_plain_password,
+            sasl_kerberos_service_name=sasl_kerberos_service_name,
+            sasl_kerberos_domain_name=sasl_kerberos_domain_name,
+            sasl_oauth_token_provider=sasl_oauth_token_provider)
 
     async def close(self):
         """Close the KafkaAdminClient connection to the Kafka broker."""

tests/test_sasl.py

@@ -6,6 +6,7 @@
 
 from aiokafka.producer import AIOKafkaProducer
 from aiokafka.consumer import AIOKafkaConsumer
+from aiokafka.admin import AIOKafkaAdminClient
 
 from aiokafka.errors import (
     TopicAuthorizationFailedError, GroupAuthorizationFailedError,
@@ -74,6 +75,19 @@ async def consumer_factory(self, user="test", **kw):
         await consumer.start()
         return consumer
 
+    async def admin_client_factory(self, user="test", **kw):
+        admin_client = AIOKafkaAdminClient(
+            bootstrap_servers=[self.sasl_hosts],
+            security_protocol="SASL_PLAINTEXT",
+            sasl_mechanism="PLAIN",
+            sasl_plain_username=user,
+            sasl_plain_password=user,
+            **kw
+        )
+        self.add_cleanup(admin_client.close)
+        await admin_client.start()
+        return admin_client
+
     async def gssapi_producer_factory(self, **kw):
         if self.kafka_version == "0.9.0.1":
             kw['api_version'] = "0.9"
@@ -110,6 +124,21 @@ async def gssapi_consumer_factory(self, **kw):
         await consumer.start()
         return consumer
 
+    async def gssapi_admin_client_factory(self, **kw):
+        if self.kafka_version == "0.9.0.1":
+            kw['api_version'] = "0.9"
+
+        admin_client = AIOKafkaAdminClient(
+            bootstrap_servers=[self.sasl_hosts],
+            security_protocol="SASL_PLAINTEXT",
+            sasl_mechanism="GSSAPI",
+            sasl_kerberos_domain_name="localhost",
+            **kw
+        )
+        self.add_cleanup(admin_client.close)
+        await admin_client.start()
+        return admin_client
+
     async def scram_producer_factory(self, user="test", **kw):
         producer = AIOKafkaProducer(
 
@@ -142,6 +171,19 @@ async def scram_consumer_factory(self, user="test", **kw):
         await consumer.start()
         return consumer
 
+    async def scram_admin_client_factory(self, user="test", **kw):
+        admin_client = AIOKafkaAdminClient(
+            bootstrap_servers=[self.sasl_hosts],
+            security_protocol="SASL_PLAINTEXT",
+            sasl_mechanism="SCRAM-SHA-256",
+            sasl_plain_username=user,
+            sasl_plain_password=user,
+            **kw
+        )
+        self.add_cleanup(admin_client.close)
+        await admin_client.start()
+        return admin_client
+
     @kafka_versions('>=0.10.0')
     @run_until_complete
     async def test_sasl_plaintext_basic(self):
@@ -181,6 +223,29 @@ async def test_sasl_plaintext_scram(self):
         msg = await consumer.getone()
         self.assertEqual(msg.value, b"Super scram msg")
 
+    @kafka_versions('>=0.10.0')
+    @run_until_complete
+    async def test_admin_client_sasl_plaintext_basic(self):
+        admin_client = await self.admin_client_factory()
+        cluster_info = await admin_client.describe_cluster()
+        self.assertGreaterEqual(len(cluster_info["brokers"]), 1)
+
+    @kafka_versions('>=0.10.0')
+    @run_until_complete
+    async def test_admin_client_sasl_plaintext_gssapi(self):
+        self.kerberos_utils.kinit("client/localhost")
+        admin_client = await self.gssapi_admin_client_factory()
+        cluster_info = await admin_client.describe_cluster()
+        self.assertGreaterEqual(len(cluster_info["brokers"]), 1)
+
+    @kafka_versions('>=0.10.0')
+    @run_until_complete
+    async def test_admin_client_sasl_plaintext_scrum(self):
+        self.kafka_config.add_scram_user("test", "test")
+        admin_client = await self.scram_admin_client_factory()
+        cluster_info = await admin_client.describe_cluster()
+        self.assertGreaterEqual(len(cluster_info["brokers"]), 1)
+
     ##########################################################################
     # Topic Resource
     ##########################################################################