kafka_consumer: add ca certificate path option (DataDog#22602)

* Add option to pass ca certificate for oauth provider

* add changelog

* don't default to kafka ca certificate

* remove default fallback

* Update kafka_consumer/datadog_checks/kafka_consumer/data/conf.yaml.example

Co-authored-by: Bryce Eadie <bryce.eadie@datadoghq.com>

* Update kafka_consumer/assets/configuration/spec.yaml

Co-authored-by: Bryce Eadie <bryce.eadie@datadoghq.com>

---------

Co-authored-by: Bryce Eadie <bryce.eadie@datadoghq.com>

Author: piochelepiotr

kafka_consumer/assets/configuration/spec.yaml

@@ -261,6 +261,15 @@ files:
             Format: 'logicalCluster=lkc-xxxxx' or 'logicalCluster=lkc-xxxxx,identityPoolId=pool-xxxxx'
           value:
             type: string
+        - name: tls_ca_cert
+          enabled: false
+          description: |
+            Path to a CA certificate file used to verify the OIDC token endpoint's TLS certificate.
+            This maps to librdkafka's `https.ca.location` property, which is separate from
+            the broker connection's `ssl.ca.location` (configured through the top-level `tls_ca_cert`).
+          value:
+            type: string
+            example: /path/to/oauth_ca_cert.pem
       - template: instances/tls
       - name: tls_crlfile
         description: |

kafka_consumer/changelog.d/22602.added

@@ -0,0 +1 @@
+Kafka consumer connects to oauth provider with ca certificate in custom path

kafka_consumer/datadog_checks/kafka_consumer/client.py

@@ -79,6 +79,8 @@ def __get_authentication_config(self):
             )
             extras_parameters["sasl.oauthbearer.scope"] = self.config._sasl_oauth_token_provider.get("scope")
             extras_parameters["sasl.oauthbearer.extensions"] = self.config._sasl_oauth_token_provider.get("extensions")
+            if self.config._sasl_oauth_tls_ca_cert:
+                extras_parameters["https.ca.location"] = self.config._sasl_oauth_tls_ca_cert
 
         for key, value in extras_parameters.items():
             # Do not add the value if it's not specified

kafka_consumer/datadog_checks/kafka_consumer/config.py

@@ -80,6 +80,10 @@ def __init__(self, init_config, instance, log) -> None:
         ):
             self._tls_ca_cert = '/opt/datadog-agent/embedded/ssl/certs/cacert.pem'
 
+        self._sasl_oauth_tls_ca_cert = (
+            self._sasl_oauth_token_provider.get("tls_ca_cert") if self._sasl_oauth_token_provider else None
+        )
+
         # Data Streams live messages
         self.live_messages_configs = instance.get('live_messages_configs', [])
 

kafka_consumer/datadog_checks/kafka_consumer/config_models/instance.py

@@ -38,6 +38,7 @@ class SaslOauthTokenProvider(BaseModel):
     client_secret: Optional[str] = None
     extensions: Optional[str] = None
     scope: Optional[str] = None
+    tls_ca_cert: Optional[str] = None
     url: Optional[str] = None
 
 

kafka_consumer/datadog_checks/kafka_consumer/data/conf.yaml.example

@@ -217,6 +217,13 @@ instances:
         #
         # extensions: <EXTENSIONS>
 
+        ## @param tls_ca_cert - string - optional - default: /path/to/oauth_ca_cert.pem
+        ## Path to a CA certificate file used to verify the OIDC token endpoint's TLS certificate.
+        ## This maps to librdkafka's `https.ca.location` property, which is separate from
+        ## the broker connection's `ssl.ca.location` (configured through the top-level `tls_ca_cert`).
+        #
+        # tls_ca_cert: /path/to/oauth_ca_cert.pem
+
     ## @param tls_verify - boolean - optional - default: true
     ## Instructs the check to validate the TLS certificate(s) of the service(s).
     #