Add a tls e2e env

kafka_consumer/datadog_checks/kafka_consumer/client/confluent_kafka_client.py

@@ -14,13 +14,24 @@ class ConfluentKafkaClient(KafkaClient):
     @property
     def kafka_client(self):
         if self._kafka_client is None:
-            self._kafka_client = AdminClient(
-                {
-                    "bootstrap.servers": self.config._kafka_connect_str,
-                    "socket.timeout.ms": self.config._request_timeout_ms,
-                    "client.id": "dd-agent",
-                }
-            )
+            config = {
+                "bootstrap.servers": self.config._kafka_connect_str,
+                "socket.timeout.ms": self.config._request_timeout_ms,
+            }
+
+            if self.config._use_tls:
+                config.update(
+                    {
+                        "security.protocol": "ssl",
+                        "ssl.ca.location": self.config._tls_ca_cert,
+                        "ssl.certificate.location": self.config._tls_cert,
+                        "ssl.key.location": self.config._tls_private_key,
+                        "ssl.key.password": self.config._tls_private_key_password,
+                    }
+                )
+
+            self._kafka_client = AdminClient(config)
+
         return self._kafka_client
 
     def create_kafka_admin_client(self):

kafka_consumer/datadog_checks/kafka_consumer/config.py

@@ -38,6 +38,11 @@ def __init__(self, init_config, instance) -> None:
         self._sasl_kerberos_service_name = instance.get('sasl_kerberos_service_name', 'kafka')
         self._sasl_kerberos_domain_name = instance.get('sasl_kerberos_domain_name')
         self._sasl_oauth_token_provider = instance.get('sasl_oauth_token_provider')
+        self._use_tls = instance.get('use_tls', False)
+        self._tls_ca_cert = instance.get("tls_ca_cert")
+        self._tls_cert = instance.get("tls_cert")
+        self._tls_private_key = instance.get("tls_private_key")
+        self._tls_private_key_password = instance.get("tls_private_key_password")
         self.use_legacy_client = is_affirmative(instance.get('use_legacy_client', False))
 
     def validate_config(self):

kafka_consumer/hatch.toml

@@ -20,10 +20,10 @@ impl = ["legacy"]
 python = ["3.8"]
 version = ["1.1", "2.3", "3.3"]
 
-#[[envs.default.matrix]]
-#python = ["3.8"]
-#version = ["3.3"]
-#auth = ["ssl"]
+[[envs.default.matrix]]
+python = ["3.8"]
+version = ["3.3"]
+auth = ["ssl"]
 
 [envs.default.overrides]
 matrix.version.e2e-env = { value = true, if = ["3.3"] }

kafka_consumer/tests/conftest.py

@@ -13,14 +13,14 @@
 from datadog_checks.dev import WaitFor, docker_run
 from datadog_checks.kafka_consumer import KafkaCheck
 
-from .common import DOCKER_IMAGE_PATH, HERE, HOST_IP, KAFKA_CONNECT_STR, LEGACY_CLIENT, TOPICS
+from .common import AUTHENTICATION, DOCKER_IMAGE_PATH, HERE, HOST_IP, KAFKA_CONNECT_STR, LEGACY_CLIENT, TOPICS
 from .runners import KConsumer, Producer
 
-# Dummy TLS certs
-CERTIFICATE_DIR = os.path.join(os.path.dirname(__file__), 'certificate')
-cert = os.path.join(CERTIFICATE_DIR, 'cert.cert')
-private_key = os.path.join(CERTIFICATE_DIR, 'server.pem')
-
+CERTIFICATE_DIR = os.path.join(os.path.dirname(__file__), 'docker', 'ssl', 'certificate')
+ROOT_CERTIFICATE = os.path.join(CERTIFICATE_DIR, 'caroot.pem')
+CERTIFICATE = os.path.join(CERTIFICATE_DIR, 'cert.pem')
+PRIVATE_KEY = os.path.join(CERTIFICATE_DIR, 'key.pem')
+PRIVATE_KEY_PASSWORD = 'secret'
 
 if LEGACY_CLIENT:
     E2E_METADATA = {
@@ -37,13 +37,28 @@
         'start_commands': ['bash /tmp/start_commands.sh'],
     }
 
-INSTANCE = {
-    'kafka_connect_str': KAFKA_CONNECT_STR,
-    'tags': ['optional:tag1'],
-    'consumer_groups': {'my_consumer': {'marvel': [0]}},
-    'broker_requests_batch_size': 1,
-    'use_legacy_client': LEGACY_CLIENT,
-}
+if AUTHENTICATION == "ssl":
+    INSTANCE = {
+        'kafka_connect_str': "kafka1:9092",
+        'tags': ['optional:tag1'],
+        'consumer_groups': {'my_consumer': {'marvel': [0]}},
+        'broker_requests_batch_size': 1,
+        'use_tls': True,
+        'tls_validate_hostname': True,
+        'tls_cert': CERTIFICATE,
+        'tls_private_key': PRIVATE_KEY,
+        'tls_private_key_password': PRIVATE_KEY_PASSWORD,
+        'tls_ca_cert': ROOT_CERTIFICATE,
+        'use_legacy_client': LEGACY_CLIENT,
+    }
+else:
+    INSTANCE = {
+        'kafka_connect_str': KAFKA_CONNECT_STR,
+        'tags': ['optional:tag1'],
+        'consumer_groups': {'my_consumer': {'marvel': [0]}},
+        'broker_requests_batch_size': 1,
+        'use_legacy_client': LEGACY_CLIENT,
+    }
 
 
 @pytest.fixture(scope='session')
@@ -80,22 +95,6 @@ def kafka_instance():
     return copy.deepcopy(INSTANCE)
 
 
-@pytest.fixture
-def kafka_instance_tls():
-    return {
-        'kafka_connect_str': KAFKA_CONNECT_STR,
-        'tags': ['optional:tag1'],
-        'consumer_groups': {'my_consumer': {'marvel': [0]}},
-        'broker_requests_batch_size': 1,
-        'use_tls': True,
-        'tls_validate_hostname': True,
-        'tls_cert': cert,
-        'tls_private_key': private_key,
-        'tls_ca_cert': CERTIFICATE_DIR,
-        'use_legacy_client': LEGACY_CLIENT,
-    }
-
-
 def create_topics():
     client = _create_admin_client()
 
@@ -106,10 +105,8 @@ def create_topics():
 
 
 def initialize_topics():
-    consumer = KConsumer(TOPICS)
-
-    with Producer():
-        with consumer:
+    with Producer(INSTANCE):
+        with KConsumer(INSTANCE, TOPICS):
             time.sleep(5)
 
 
@@ -121,9 +118,20 @@ def mock_local_kafka_hosts_dns():
 
 
 def _create_admin_client():
-    return AdminClient(
-        {
-            "bootstrap.servers": KAFKA_CONNECT_STR,
-            "socket.timeout.ms": 1000,
-        }
-    )
+    config = {
+        "bootstrap.servers": INSTANCE['kafka_connect_str'],
+        "socket.timeout.ms": 1000,
+    }
+
+    if INSTANCE.get('use_tls', False):
+        config.update(
+            {
+                "security.protocol": "ssl",
+                "ssl.ca.location": INSTANCE.get("tls_ca_cert"),
+                "ssl.certificate.location": INSTANCE.get("tls_cert"),
+                "ssl.key.location": INSTANCE.get("tls_private_key"),
+                "ssl.key.password": INSTANCE.get("tls_private_key_password"),
+            }
+        )
+
+    return AdminClient(config)

kafka_consumer/tests/docker/ssl/cert.pem

@@ -0,0 +1,27 @@
+-----BEGIN CERTIFICATE-----
+MIIEnjCCAoYCCQCg3s7NDJiDPDANBgkqhkiG9w0BAQsFADARMQ8wDQYDVQQDDAZj
+bGllbnQwHhcNMjMwMzA4MTQyODIxWhcNMzMwMzA1MTQyODIxWjARMQ8wDQYDVQQD
+DAZjbGllbnQwggIiMA0GCSqGSIb3DQEBAQUAA4ICDwAwggIKAoICAQCvTeOWMVla
+oePAw4FywBSe22dGXhvHxCzYQ8cE2rWb4NkLfr+nq428hPd2wtI+lhnXy+XpSoRk
+r+D02oNtkk/U/tBQxutnMvo4jy6g+N5HltfTwLrNgPROVbDZRRonv3CJp96fMdFc
+2fpdD5KKRLgqGogL1aZIib7nxdadRC5oHRGJLszxGrTygYsep9QrJWY7S4c9LxTS
+DbS7Dtg9VaCuu97U71Dg8fQo/1hGWcqQRn8OvBz649TZ3cRyPph8iLlj/ZAwcMcE
+rTGK0Bd3wR7nvVpqCbpx1Sn4z8dYaB+0/7Oiq30EzW9XHTKyuI2L3aqpK8j4Jt4H
+gM+L4nobM6ja68yu5b/5Ddms5SK81oVlXp7SnLA3E6A+HIAiAN7LCGBsbucPE2BI
+oAuYxJuha23I+gBa8ZOF4p/j7EQK4au3Hx3HLpk11tE+qMikYQzYnGblzb6pWbzv
+KS37zkkNhlwUn7Dr4YxMwB/usuaJEck/CI7QaJ7Jy7lpZq/6qSUfN5iHDcZqDpTF
+16Rvf+2BsZ3D7/xlmcBGc5BE3aR/XkJehK9vT42/2IUesxEk6ZNrjqQVy5fXtVuu
+uhlB+q3A1Wwq2fy/j4pU4nuYG6a9+XB2845vGoxk/KbTZ2Fg1h5OGXynlfgd6nSC
+hJrxya12ykeIkPoL2Q4Jnai+M347U8fdQQIDAQABMA0GCSqGSIb3DQEBCwUAA4IC
+AQAFSt19Pp1Jo3fd/GYPBdNtXd1rLIVoDlDyLFnouylExMu0UxNwOxVu5qRsWoNr
+FSmecY2YuCjgyy0QmuR51zcfi2pV7HiUh00SJgPwa71yBRQD5nlI2aXdnWj1k+mW
+ygJWfGw6wIQ1WiE4qDgHpcdlbzLry03kts8cN/Kryos+F5TZ1QEBGQREPLY06aTZ
+fZwYPOofUs5Ckq4JKQkcAvOoZjjy1IuPlvKjQFVfFfoHCDmiAzshrrxxdJr1Vz7u
+P9pA6do+MdIsLHa6E00hANTO4AaoXJx32IGCFwQb5vNw8Ak2JexIby4Xtjvigz+x
+HFojc0wjGkKLA4WlqvLhd2USDBgUnsFum9ifjSJ0L8aqTo2LDcXFlp/uH3whIzOi
+gY0kptE/ucdGLErMff2qdticPSiPxtHVstiZjsDS0HdpHJgsI4RDnU9Cxx7sObUV
+ox0finvBCbYdSxo8vaS21LoXg2tC7cU+jlSBOD82Cz2dH3Q/2AiugSda5Zwpakpy
+IrOj1xbxJZum/koRp+L5dAIFcAU3U50Lv1Ykqa3cu25J+dnqSBEKbmwMlcNs2c5C
+kTfVmprALKinZgZEmNcL9lWQcB7ekGgpfe65cDQSQYYNPzeS/BSOliBEXyR2pQmE
+3QfQzsDbbe9AIS0Q+4bRC3KhH0dB1IX2EZHtwf210nuYnQ==
+-----END CERTIFICATE-----

kafka_consumer/tests/docker/ssl/certificate/caroot.pem

@@ -0,0 +1,18 @@
+-----BEGIN CERTIFICATE-----
+MIIC8DCCAdgCCQD1NF3JgzmSbDANBgkqhkiG9w0BAQsFADA6MQswCQYDVQQGEwJV
+UzEVMBMGA1UECgwMU2VydmljZVVzZXJzMRQwEgYDVQQDDAtrYWZrYS1hZG1pbjAe
+Fw0yMzAzMDkxMDQ0NTRaFw0zMzAzMDYxMDQ0NTRaMDoxCzAJBgNVBAYTAlVTMRUw
+EwYDVQQKDAxTZXJ2aWNlVXNlcnMxFDASBgNVBAMMC2thZmthLWFkbWluMIIBIjAN
+BgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA3j6xm1bkA8rt8V+ZhXosc0zljA3w
+HilR6ApToaGjdM9BMM2qgQwmZlIr/fjanL4KYNKNGazn+3hw8mm+JSmBPwczjgjy
+EfXxJvI8XA+lDwoM7LVv9vIR4X0/QvAHvFLQuYC9kNHs2kN0xTqw+ZdztKkWFZoJ
+OLygp/qglbN+0eVaS6w91PlX6URaCpjrB1QPWzUcAfvYU4D4lPeSlB5HYxSNr9Cx
+RY6S66xdCk0LvkTJUEPb9/whckuJGjGOJNK8QMASPYDv/bF7U8bq5eIal6hAqVXP
+91BSYCWVHs+Bb6Xt9g812vMUjLPtejlu2c4GCZ6xA/yMJ7s1Uxth9ThAVwIDAQAB
+MA0GCSqGSIb3DQEBCwUAA4IBAQC9zDMd0KxuP7puG5HTWH0Uv8P/8g5b7l8xP2sW
+GgEtVlMFQUS4nm6WxZrSkYbEsuEvG8URjbqFRli+2SSV8UVB3HxlkA1wYX90ForJ
+/yyINj1ohqzKdsMMco6jhK9UP0opEXp4sMKBl359mjnPbu9RqjwangkhPrLZNZNF
+pbo6HIY0vFlUqc0yWYAFmY8ixi3zAUKAr94eFWJwpFH7rncFYx+3Q1y5JgHvlQKT
+wjjTCr7sKfAsTqUC8OE7HguS6SVKx7RRHV9C0OTL5ekdEHURC4uifgGV3WtkVQGH
+5wdC0nNY/Wjuc66TKkqzLoyydfRSXpWuEpOah0FcQJX9DfLy
+-----END CERTIFICATE-----

kafka_consumer/tests/docker/ssl/certificate/cert.pem

@@ -0,0 +1,21 @@
+-----BEGIN CERTIFICATE-----
+MIIDhDCCAmwCCQDY21uQbeDVTTANBgkqhkiG9w0BAQsFADA6MQswCQYDVQQGEwJV
+UzEVMBMGA1UECgwMU2VydmljZVVzZXJzMRQwEgYDVQQDDAtrYWZrYS1hZG1pbjAe
+Fw0yMzAzMDkxMDQ1MDFaFw0zMzAzMDYxMDQ1MDFaME4xEjAQBgNVBAMTCWxvY2Fs
+aG9zdDEVMBMGA1UEChMMU2VydmljZVVzZXJzMQkwBwYDVQQHEwAxCTAHBgNVBAgT
+ADELMAkGA1UEBhMCVVMwggGiMA0GCSqGSIb3DQEBAQUAA4IBjwAwggGKAoIBgQCz
+KM9rFto8r4XD8iEasbSx0/hpWxNYc9I3vtasLPsedt6MY0DbXigEXHX92nx1e+Cl
+qVW9EGkIzD0Jgz55pqwi0wctgTJNTmUjSkvefMxggWIUN4QTeCV9jtqZw5jN1tgS
+eIT0uCkYtUvGPajEAOjWWLjmChFJkNDnI2tIQMXU/W2Ob1YnZeGfxG8qlK53bpIJ
+XQY1l65LSQtUI6aDsrysjTDmYnVqHF4/FZzym2IhESIWrZQGE1/A0DJaZEO9hyn2
+//GWtnGzkt4s2+7bbuiSNaA6af9YbeDroKt+ByHXB9YAyxind1tVhNAIv4MQ6GE7
+yco01VV+P+vj9LSSCjU3b8pUrZDJEKYUth/5XSEfAxLaej74TVk6bPrZR6s03C9n
+V41DKTKqPFb9Ql+s5XewRnRLp7slX2TVzbHUuLTA/ID9U68lDxOQc9mRQWt10X4o
+yxqpUmsXf7u2PedO7Wl5fTRZpQsidtukYBLezYEZPpPiSP8G2FMP1f2VgvG9YyUC
+AwEAATANBgkqhkiG9w0BAQsFAAOCAQEAmohuyZjCxPw2mdZrvxDMf6Rcli6DyyKL
+IrPq/EH3kNylxhYix86QUNggnJZv+HSOEYVf++cPf9VamGVQZapYLMCzJVrOdNgT
+hDdy9+a8Q2heLHPfJ7u11cjO6V5q4WOLEmCx4pI2cJ+Y+e3wxSKjas9A6VQaJuFm
+2uHd3T6ny0m7HLkUAoM1/LqnMvGmG7wXBrsK/AwaMjeyW04skkhGks1GbmJsx/G1
+QvhnBcSJvXVjBa6PQ52e3BDpoqLAs4BuOeWuZ03D55/Cp5bYkGbo3QpWIArLuikN
+yktIeUgnBvSKRTgI9GDP+uKesG/Ny8Mle0t5BbU1fXgvw/LfJNslCQ==
+-----END CERTIFICATE-----

kafka_consumer/tests/docker/ssl/certificate/key.pem

@@ -0,0 +1,41 @@
+-----BEGIN ENCRYPTED PRIVATE KEY-----
+MIIHKjAcBgoqhkiG9w0BDAEEMA4ECGwj+87fxFROAgIDJASCBwgRUvWC2+3lrnJ7
+LyZmkR1sc9Dt+EVZC3H46nIvU65PCbXotOenQ2l0RcXsWSmAXr0IQkrKU2e/TRlG
+VqfTNYPmysxNv39dGjOL7aaa/Fe2i5//40Z2DaZ+OhXhg3eVYthbfsokngiyWg5O
+CMez0JjHD3ZgvajUs/rK4NDDpSyRG7iGnROGTaoyOhNb3RFXqvGCf8AlR1uEGB5V
+fh4CIuvgOx26YtnNtmRHTwctFbOSRmZd1eMOYWrRYAhAS+NAP1f58tezcjkYaDTf
+texJurkY8hjNdFCpmhWqxVKaorcQk3nqehc8ij9ayskBqQmf5cWwiW+MVOXBkArx
+mmN4LS63VKyPeTdV312a5kuEYZurdMETN9O41h3PDaLMfayqAinQVdUv1GVdPYs0
+BIuHpa+FRaqVM5RZugjrj3AzmVtwfKsAdM/r9OnauFpdiWrWvHIhfKCc1ZibwmSY
+X4DagVMEs9t/jPSxBgEfb0RKc9UbHatIrm1hG6UDj5fbhPvMOz7xlaDxiOgXwUB0
+6HyDc1xQqDLLgN2mFwFJ0okZUaW4gfpF7EjCzGFfMImTkkive8GbH0rbaUFIaRqK
+SzcoP4hDUyxC1Iiz1k4OeiOG/fKai9RNeU4v3muyRoXE8FW5JtgT8yRKTgilrC/C
+yn6cpI6wawgn62FGn4r9slXlbbLd0Q0MI1LEbPSvYo14EDBokfmZrosAHHzoC7xT
+dy1tDQEK66UGbZ/CDI55dHXasGPAajkTa0tPjPFcoJjqkZs/9uBpgGT2b3Y9FVyH
+nI5cO907ST8rPJNmHKYulf0nV4VtQiMvewSFbDQiz8/Qyb3KxkwY93WC4ia/wAZH
++r1eUyZM6cRR3dug+vHWqmX1Amw4ogLJax+rJbdCVXHA3SQlsPe2YcbX8axdKrTq
+EWPzYqx8ZEgtdi/aVcyRh/sfoMUv6bUmJg6eVH+3ph1KKlBT2AZU9zPuJu2tkJUW
+Ols+gscp1TU6BVON+vRRLHyhu7yz2jx0WbyNLtaCDaEHTEaXk8McCWp70kimqz/T
+jLHahVZ8YJprtksSb9mXSqgiJzTOFMYv03cf3F1PNB928KZWGhU/trDis2WJB5FC
+9E7WCTFVKr21+onoQFcZ86mBHzlATUu8BXd0LTWvjSqlHhfynyNyk0P0QjMwkin4
+9V9SWHSGimD6YENM9o+zyp9DgGvEuRMUztvrxZ2lVVsBZMCCQhja/wakeIOVwQDx
+S8VRjKm8/DgAYZHUCJTvOIBxLnodImO7VdnpGpaknT9v5E2aWyWFs8RjpeP7X7Pa
+95r62IKMstvRam4QvXX7ascO7uI4vNpl/RkGSw9gKjj+rqH7IZ37JuDskR7HZTsp
+vFYkUtcdoBkZUeQoaabzj8zCXTvhUvbdWHep+IOGnEJ54cYDzxmsDYsdveTHncT2
+GkPZGWFQfbh2wX1V7AHxdKsbwY8kWWaGj1HdNzKtoCqFQmBj9TGqHuX6FX1SzWzk
+Fw1eHLJ53I9OHQOyRtpmKknQbNB5AASVSoMYQx7Qj4/BKGNsUYt6ymyarNsW4e0g
+k/e14lH4uKhRvUuUaJUe7nNVjm4Q888lOWQTMmawYaobtOv+wAh9FD1E5ThvQ4d2
+dtWaNLvGost69TWm+BqUai4JJ/Ws/t8uZzws2sbbwFO7FdMEOSniYT460HlxHWfo
+Zl3BFvxbSmsFwKCCAhtuGEHi56WC27wD+yVEEBDdOpqkoltBQw8hRjhNMozDb1we
+EzGyaCHfXtOlKMomJzODW0Erb8PbR+m50CHYLhu5E49CoJ6uhWyFJzKVppYxhQC6
+kbv2b25TONLtwW0/EuIPtIpcI3k4MewJbiEFtOxzhpU8p0fosPJvJLmbbiM5oKsv
+Z7J09LMxQfn+rILlHeywhjaYtCodqnP/1lfvX6Cwzc1QSI6j5A6pe4IFMPfarlY6
+XX4+0eqpsFQSrJGvXh6+lfwZ1SKvjQi2+LOYz2+Ia0wsqlr+RxhjyrrHLN3hheCN
+GFC7zXldMaa62Gk9BGnEF626yRnDU3MSe3WyOPl1KLCZLjFhGGo1h56kf6bQGpKj
+d/Ai7YMdHJo3d0eOo+w/3cgIdHbdILIXpBRXwxk8lSMfkkkQXinvhUF2XMsR2kNl
+djSByS1Dh7oBibM+llN3Ole36GVU2QAxXZImnKZctJMi+7dy1cXtVfX5Tdie3To3
+w2vcwVCa81iUAAv+WjwlB2Oit5OIVpvRoCm6JX07u5UiL7iWmnhyW/Zuh6Rojbl3
+FgRezdkSCwWxMCPXF+MU3O5EUHSbqrIFmIcMUEydeMJakzKFmugjwkdSObllnbUu
+WiGAjmrcYu7x7Fk59kjk20BIsrwBCH29+HiSj/eDmEu+zeuOV/ntJoZ8hf/td+al
+g4B0DD3hTUdDJo+Cd70=
+-----END ENCRYPTED PRIVATE KEY-----

kafka_consumer/tests/docker/ssl/client.config

@@ -0,0 +1,7 @@
+bootstrap.servers=kafka1:9092
+security.protocol=SSL
+ssl.truststore.location=/bitnami/kafka/config/certs/kafka.truststore.jks
+ssl.truststore.password=secret
+ssl.keystore.location=/bitnami/kafka/config/certs/kafka.keystore.jks
+ssl.keystore.password=secret
+ssl.key.password=secret

kafka_consumer/tests/docker/ssl/command.properties

@@ -0,0 +1,7 @@
+bootstrap.servers=kafka1:19092
+security.protocol=SSL
+ssl.truststore.location=/bitnami/kafka/config/certs/kafka.truststore.jks
+ssl.truststore.password=secret
+ssl.keystore.location=/bitnami/kafka/config/certs/kafka.keystore.jks
+ssl.keystore.password=secret
+ssl.key.password=secret