Setting up Jaas config extensions for kafka connect.

[firdaustahir]

Hi,

I'm setting up my kafka connect with OAuth configuration against Confluent Cloud platform.

I configured Kafka-connect with OAuth authentication by referring to Strimzi KafkaClientAuthenticationOAuth guide. The following is the authentication config for my kafka-connect.

authentication:
    type: oauth
    clientId: client-01
    clientSecret:
      key: secret
      secretName: client-secret
    tokenEndpointUri: https://login.microsoftonline.com/e741d71c-c6b6-47b0-803c-0f3b32b07556/oauth2/v2.0/token
tls:  
  trustedCertificates: 
  - secretName: cert_secret
    certificate: certfile.crt

Based on Confluent documentation, to set OAuth authentication, the following needs to be configured in kafka client.

security.protocol=SASL_SSL
sasl.oauthbearer.token.endpoint.url=https://myidp.example.com/oauth2/default/v1/token
sasl.login.callback.handler.class=org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerLoginCallbackHandler
sasl.mechanism=OAUTHBEARER
sasl.jaas.config= \
  org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
    clientId='<client ID>' \
    scope='<Requested Scope>' \
    clientSecret='<Client Secret>' \
    extension_logicalCluster='<Cluster ID>' \
    extension_identityPoolId='<Pool ID>';

I can see two additional configurations are required for the sasl.jaas.config - extension_logicalCluster and extension_identityPoolId. However, in strimzi kafka client, I am not able to configure those two extensions. Thus when deploying kafka-connect, I encountered the following error :

2024-02-15 22:19:52,640 ERROR Stopping due to error (org.apache.kafka.connect.cli.AbstractConnectCli) [main]
org.apache.kafka.connect.errors.ConnectException: Failed to connect to and describe Kafka cluster. Check worker's broker connection and security properties.
	at org.apache.kafka.connect.runtime.WorkerConfig.lookupKafkaClusterId(WorkerConfig.java:283)
	at org.apache.kafka.connect.runtime.WorkerConfig.lookupKafkaClusterId(WorkerConfig.java:263)
	at org.apache.kafka.connect.runtime.WorkerConfig.kafkaClusterId(WorkerConfig.java:393)
	at org.apache.kafka.connect.cli.AbstractConnectCli.startConnect(AbstractConnectCli.java:124)
	at org.apache.kafka.connect.cli.AbstractConnectCli.run(AbstractConnectCli.java:94)
	at org.apache.kafka.connect.cli.ConnectDistributed.main(ConnectDistributed.java:116)
Caused by: java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.SaslAuthenticationException: Authentication failed: 1 extensions are invalid! They are: logicalCluster: CLUSTER_ID_MISSING_OR_EMPTY
	at java.base/java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:396)
	at java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2073)
	at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:165)
	at org.apache.kafka.connect.runtime.WorkerConfig.lookupKafkaClusterId(WorkerConfig.java:277)
	... 5 more
Caused by: org.apache.kafka.common.errors.SaslAuthenticationException: Authentication failed: 1 extensions are invalid! They are: logicalCluster: CLUSTER_ID_MISSING_OR_EMPTY

I noticed in strimzi kafka-connect crds, it doesn't allow to set sasl.jaas.config in the config: section.

config:
   x-kubernetes-preserve-unknown-fields: true
   type: object
   description: "The Kafka Connect configuration. Properties with the following prefixes cannot be set: ssl., sasl., security., listeners, plugin.path, rest., bootstrap.servers, consumer.interceptor.classes, producer.interceptor.classes (with the exception of: ssl.endpoint.identification.algorithm, ssl.cipher.suites, ssl.protocol, ssl.enabled.protocols)."

Basically, my question here is there a way to set the jaas extensions (logicalCluster, identityPoolId) in kafka-connect yaml file?

Thanks.
Firdaus

[scholzj]

I don't think we support these extension_ options. You might need to switch to a different authentication method.

[Karthik-Rangasamy]

Hi @scholzj

As Confluent cloud is being adopted by many, Strimzi Kafka Connect to Confluent cloud is becoming a common deployment setup.
Wondering should Strimzi Kafka Connect allow options to configure towards confluent cloud? should this be a new feature request?

Confluent cloud doc: https://docs.confluent.io/cloud/current/access-management/authenticate/oauth/overview.html

Thanks

[scholzj]

I think you can just use a different authentication mechanism to connect to Confluent Cloud.

[Karthik-Rangasamy]

Hi @scholzj

Unfortunately OAuth is the chosen authentication mechanism given the benefits of it and i'm looking for options on how this can be enabled on the Strimzi Kafka connect side?

[rao2100]

Hi @scholzj,

Is there a way to set the values directly for kafka-connect, example below:

sasl.mechanism=PLAIN
security.protocol=SASL_SSL
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="connect" \
  password="connect-secret";

Like how confluent connect allows:
https://docs.confluent.io/platform/current/kafka/authentication_sasl/authentication_sasl_plain.html#kafka-connect

[scholzj]

I don't think you need to set any value directly for what you have in your example. You can just use the type: plain authentication in the KafkaConnect CR.

[tbellisiv-mediware]

Hi @scholzj I'm facing a similar requirement for configuring Strimzi connect to authenticate to a managed Kafka service. In my case, the Kafka cluster is a Google-managed cluster and Strimzi connect is hosted in Google's managed Kubernetes service (GKE).

In order to configure Strimzi connect to authenticate, I need to specify the following properties for the connect workers:

security.protocol: SASL_SSL
sasl.mechanism: OAUTHBEARER
sasl.login.callback.handler.class: com.google.cloud.hosted.kafka.auth.GcpLoginCallbackHandler
sasl.jaas.config: org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required;

Google has implemented a unique JWT-based auth scheme that does not support OIDC-based auth flows. Instead, they have developed a custom JAAS login handler that constructs a JWT using the Google credentials of the context that the Kafka Client is running under.

Since the sasl.* properties are not configurable via the KafkaConnect resource, is there any other way these properties can configured? If not, would love to know if support for this could be added in a future release?

[scholzj]

I don't think this is supported in Strimzi. You will need to use different authentication type if available in your Google cluster.

[tbellisiv-mediware]

The only other supported method for Google's managed Kafka service is SASL/Plain. Unfortunately, because this relies on GCP service account keys, it introduces additional management overhead to prevent security risks. Google's own documentation acknowledges this:

https://cloud.google.com/managed-service-for-apache-kafka/docs/authentication-kafka

For this reason we are avoiding SASL/Plain.

[scholzj]

But I'm afraid that would be the only way to get it working with Strimzi.