Keycloak authentication and authorization dosn't work correctly

[geniymaxdem]

Hello!
I deployed kafka cluster with Keycloak authentication and authorization
When I try to test а authorization and authentication mechanism via keylock using a kafka-cli, I recive error like - PKIX path building failed: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target
This is my kafka-cluster configuration

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: kafka-cluster
  namespace: kafka
spec:
  kafka:
    jvmOptions:
      javaSystemProperties:
        - name: javax.net.debug
          value: ssl
    version: 3.0.0
    replicas: 3
    listeners:
      - name: plain
        port: 9092
        type: internal
        tls: false
      - name: tls
        port: 9093
        type: internal
        tls: true
      - name: internal
        port: 9094
        tls: true
        type: internal
        authentication:
          type: oauth
          enableOauthBearer: true
          disableTlsHostnameVerification: true
          validIssuerUri: https://keycloak.service.rowi.tech/auth/realms/kafka-authz
          jwksEndpointUri: https://keycloak.service.rowi.tech/auth/realms/kafka-authz/protocol/openid-connect/certs
          userNameClaim: preferred_username
          maxSecondsWithoutReauthentication: 3600
          tlsTrustedCertificates:
          - secretName: oauth-server-cert
            certificate: sso.crt
    authorization:
      type: keycloak
      clientId: kafka
      delegateToKafkaAcls: false
      tokenEndpointUri: https://keycloak.service.rowi.tech/auth/realms/kafka-authz/protocol/openid-connect/token
      disableTlsHostnameVerification: true
      tlsTrustedCertificates:
      - secretName: oauth-server-cert
        certificate: sso.crt
    config:
      # start manual config
      ssl.cipher.suites: "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384"
      ssl.endpoint.identification.algorithm: ""
      ssl.enabled.protocols: "TLSv1.2"
      ssl.protocol: "TLSv1.2"
      log.retention.hours: 24
      log.segment.bytes: 104857600
      log.retention.bytes: 1073741824
      log.retention.check.interval.ms: 300000
      log.cleaner.enable: true
      log.cleanup.policy: "delete"
      fetch.max.bytes: 52428800
      max.partition.fetch.bytes: 1048576
      replica.fetch.max.bytes: 1000000000.
      max.message.bytes: 1000000000
      # end of manual config
      offsets.topic.replication.factor: 3
      transaction.state.log.replication.factor: 3
      transaction.state.log.min.isr: 2
      default.replication.factor: 3
      min.insync.replicas: 2
      inter.broker.protocol.version: "3.0"

The file alice.properties was created according to the official documentation

security.protocol=SASL_SSL
ssl.truststore.location=/tmp/truststore.p12
ssl.truststore.password=storepass
ssl.truststore.type=PKCS12
sasl.mechanism=OAUTHBEARER
sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required   oauth.refresh.token="eyJhbGciOiJIUzI1NiIsInR5cCIgOiAiSldUIiwia2lkIiA6ICI4ZDUyNmExNC1hNDFhLTQ3M2QtODVmYS1jMDY0NzYwMDUzZDAifQ.eyJpYXQiOjE2NDUwNDc1MDUsImp0aSI6Ijk2ZjRmOWFjLTFlNzctNDk2Ny05M2IxLTIzYTU0NGQ4ZWJkYSIsImlzcyI6Imh0dHBzOi8va2V5Y2xvYWsuc2VydmljZS5yb3dpLnRlY2gvYXV0aC9yZWFsbXMva2Fma2EtYXV0aHoiLCJhdWQiOiJodHRwczovL2tleWNsb2FrLnNlcnZpY2Uucm93aS50ZWNoL2F1dGgvcmVhbG1zL2thZmthLWF1dGh6Iiwic3ViIjoiMjVhMDY4M2EtYjRjNy00NjUzLWJiZDYtMDVhZmE0Yzg3MjcwIiwidHlwIjoiT2ZmbGluZSIsImF6cCI6ImthZmthLWNsaSIsInNlc3Npb25fc3RhdGUiOiIwOTQwYjBiNy05OGQ3LTRmNzYtODk5NC0xMTAwMmNmZjc3NjAiLCJzY29wZSI6Im9mZmxpbmVfYWNjZXNzIGVtYWlsIHByb2ZpbGUiLCJzaWQiOiIwOTQwYjBiNy05OGQ3LTRmNzYtODk5NC0xMTAwMmNmZjc3NjAifQ.tGhox61HpM9bmJ0yONo_AehOzvPVsgOEZKYUZgjnEbI"   oauth.client.id="kafka-cli"   oauth.ssl.truststore.location="/tmp/truststore.p12"   oauth.ssl.truststore.password="storepass"   oauth.ssl.truststore.type="PKCS12"   oauth.token.endpoint.uri="https://keycloak.service.rowi.tech/auth/realms/kafka-authz/protocol/openid-connect/token" ;
sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler

I check authentication and authorization using simple command - list topic

[root@kafka-cli tmp]# /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka-cluster-kafka-bootstrap:9094 --command-config /tmp/alice.properties --list
[2022-02-16 22:52:48,632] ERROR [AdminClient clientId=adminclient-1] Connection to node -1 (kafka-cluster-kafka-bootstrap/10.233.38.100:9094) failed authentication due to: SSL handshake failed (org.apache.kafka.clients.NetworkClient)
[2022-02-16 22:52:48,634] WARN [AdminClient clientId=adminclient-1] Metadata update failed due to authentication error (org.apache.kafka.clients.admin.internals.AdminMetadataManager)
org.apache.kafka.common.errors.SslAuthenticationException: SSL handshake failed
Caused by: javax.net.ssl.SSLHandshakeException: PKIX path building failed: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target
        at java.base/sun.security.ssl.Alert.createSSLException(Alert.java:131)
        at java.base/sun.security.ssl.TransportContext.fatal(TransportContext.java:349)
        at java.base/sun.security.ssl.TransportContext.fatal(TransportContext.java:292)
        at java.base/sun.security.ssl.TransportContext.fatal(TransportContext.java:287)
        at java.base/sun.security.ssl.CertificateMessage$T12CertificateConsumer.checkServerCerts(CertificateMessage.java:654)
        at java.base/sun.security.ssl.CertificateMessage$T12CertificateConsumer.onCertificate(CertificateMessage.java:473)
        at java.base/sun.security.ssl.CertificateMessage$T12CertificateConsumer.consume(CertificateMessage.java:369)
        at java.base/sun.security.ssl.SSLHandshake.consume(SSLHandshake.java:392)
        at java.base/sun.security.ssl.HandshakeContext.dispatch(HandshakeContext.java:443)
        at java.base/sun.security.ssl.SSLEngineImpl$DelegatedTask$DelegatedAction.run(SSLEngineImpl.java:1074)
        at java.base/sun.security.ssl.SSLEngineImpl$DelegatedTask$DelegatedAction.run(SSLEngineImpl.java:1061)
        at java.base/java.security.AccessController.doPrivileged(Native Method)
        at java.base/sun.security.ssl.SSLEngineImpl$DelegatedTask.run(SSLEngineImpl.java:1008)
        at org.apache.kafka.common.network.SslTransportLayer.runDelegatedTasks(SslTransportLayer.java:430)
        at org.apache.kafka.common.network.SslTransportLayer.handshakeUnwrap(SslTransportLayer.java:514)
        at org.apache.kafka.common.network.SslTransportLayer.doHandshake(SslTransportLayer.java:368)
        at org.apache.kafka.common.network.SslTransportLayer.handshake(SslTransportLayer.java:291)
        at org.apache.kafka.common.network.KafkaChannel.prepare(KafkaChannel.java:178)
        at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:543)
        at org.apache.kafka.common.network.Selector.poll(Selector.java:481)
        at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:551)
        at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.processRequests(KafkaAdminClient.java:1389)
        at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.run(KafkaAdminClient.java:1320)
        at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: sun.security.validator.ValidatorException: PKIX path building failed: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target
        at java.base/sun.security.validator.PKIXValidator.doBuild(PKIXValidator.java:439)
        at java.base/sun.security.validator.PKIXValidator.engineValidate(PKIXValidator.java:306)
        at java.base/sun.security.validator.Validator.validate(Validator.java:264)
        at java.base/sun.security.ssl.X509TrustManagerImpl.validate(X509TrustManagerImpl.java:313)
        at java.base/sun.security.ssl.X509TrustManagerImpl.checkTrusted(X509TrustManagerImpl.java:276)
        at java.base/sun.security.ssl.X509TrustManagerImpl.checkServerTrusted(X509TrustManagerImpl.java:141)
        at java.base/sun.security.ssl.CertificateMessage$T12CertificateConsumer.checkServerCerts(CertificateMessage.java:632)
        ... 19 more
Caused by: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target
        at java.base/sun.security.provider.certpath.SunCertPathBuilder.build(SunCertPathBuilder.java:141)
        at java.base/sun.security.provider.certpath.SunCertPathBuilder.engineBuild(SunCertPathBuilder.java:126)
        at java.base/java.security.cert.CertPathBuilder.build(CertPathBuilder.java:297)
        at java.base/sun.security.validator.PKIXValidator.doBuild(PKIXValidator.java:434)
        ... 25 more
Error while executing topic command : SSL handshake failed
[2022-02-16 22:52:48,642] ERROR org.apache.kafka.common.errors.SslAuthenticationException: SSL handshake failed
Caused by: javax.net.ssl.SSLHandshakeException: PKIX path building failed: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target
        at java.base/sun.security.ssl.Alert.createSSLException(Alert.java:131)
        at java.base/sun.security.ssl.TransportContext.fatal(TransportContext.java:349)
        at java.base/sun.security.ssl.TransportContext.fatal(TransportContext.java:292)
        at java.base/sun.security.ssl.TransportContext.fatal(TransportContext.java:287)
        at java.base/sun.security.ssl.CertificateMessage$T12CertificateConsumer.checkServerCerts(CertificateMessage.java:654)
        at java.base/sun.security.ssl.CertificateMessage$T12CertificateConsumer.onCertificate(CertificateMessage.java:473)
        at java.base/sun.security.ssl.CertificateMessage$T12CertificateConsumer.consume(CertificateMessage.java:369)
        at java.base/sun.security.ssl.SSLHandshake.consume(SSLHandshake.java:392)
        at java.base/sun.security.ssl.HandshakeContext.dispatch(HandshakeContext.java:443)
        at java.base/sun.security.ssl.SSLEngineImpl$DelegatedTask$DelegatedAction.run(SSLEngineImpl.java:1074)
        at java.base/sun.security.ssl.SSLEngineImpl$DelegatedTask$DelegatedAction.run(SSLEngineImpl.java:1061)
        at java.base/java.security.AccessController.doPrivileged(Native Method)
        at java.base/sun.security.ssl.SSLEngineImpl$DelegatedTask.run(SSLEngineImpl.java:1008)
        at org.apache.kafka.common.network.SslTransportLayer.runDelegatedTasks(SslTransportLayer.java:430)
        at org.apache.kafka.common.network.SslTransportLayer.handshakeUnwrap(SslTransportLayer.java:514)
        at org.apache.kafka.common.network.SslTransportLayer.doHandshake(SslTransportLayer.java:368)
        at org.apache.kafka.common.network.SslTransportLayer.handshake(SslTransportLayer.java:291)
        at org.apache.kafka.common.network.KafkaChannel.prepare(KafkaChannel.java:178)
        at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:543)
        at org.apache.kafka.common.network.Selector.poll(Selector.java:481)
        at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:551)
        at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.processRequests(KafkaAdminClient.java:1389)
        at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.run(KafkaAdminClient.java:1320)
        at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: sun.security.validator.ValidatorException: PKIX path building failed: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target
        at java.base/sun.security.validator.PKIXValidator.doBuild(PKIXValidator.java:439)
        at java.base/sun.security.validator.PKIXValidator.engineValidate(PKIXValidator.java:306)
        at java.base/sun.security.validator.Validator.validate(Validator.java:264)
        at java.base/sun.security.ssl.X509TrustManagerImpl.validate(X509TrustManagerImpl.java:313)
        at java.base/sun.security.ssl.X509TrustManagerImpl.checkTrusted(X509TrustManagerImpl.java:276)
        at java.base/sun.security.ssl.X509TrustManagerImpl.checkServerTrusted(X509TrustManagerImpl.java:141)
        at java.base/sun.security.ssl.CertificateMessage$T12CertificateConsumer.checkServerCerts(CertificateMessage.java:632)
        ... 19 more
Caused by: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target
        at java.base/sun.security.provider.certpath.SunCertPathBuilder.build(SunCertPathBuilder.java:141)
        at java.base/sun.security.provider.certpath.SunCertPathBuilder.engineBuild(SunCertPathBuilder.java:126)
        at java.base/java.security.cert.CertPathBuilder.build(CertPathBuilder.java:297)
        at java.base/sun.security.validator.PKIXValidator.doBuild(PKIXValidator.java:434)
        ... 25 more
 (kafka.admin.TopicCommand$)

Kafka brokers logs

javax.net.ssl|ERROR|58|data-plane-kafka-network-thread-0-ListenerName(INTERNAL-9094)-SASL_SSL-12|2022-02-16 22:52:48.627 GMT|TransportContext.java:341|Fatal (UNEXPECTED_MESSAGE): Unexpected handshake message: client_hello (
"throwable" : {
  javax.net.ssl.SSLProtocolException: Unexpected handshake message: client_hello
        at java.base/sun.security.ssl.Alert.createSSLException(Alert.java:129)
        at java.base/sun.security.ssl.Alert.createSSLException(Alert.java:117)
        at java.base/sun.security.ssl.TransportContext.fatal(TransportContext.java:336)
        at java.base/sun.security.ssl.TransportContext.fatal(TransportContext.java:292)
        at java.base/sun.security.ssl.TransportContext.fatal(TransportContext.java:283)
        at java.base/sun.security.ssl.HandshakeContext.dispatch(HandshakeContext.java:437)
        at java.base/sun.security.ssl.SSLEngineImpl$DelegatedTask$DelegatedAction.run(SSLEngineImpl.java:1074)
        at java.base/sun.security.ssl.SSLEngineImpl$DelegatedTask$DelegatedAction.run(SSLEngineImpl.java:1061)
        at java.base/java.security.AccessController.doPrivileged(Native Method)
        at java.base/sun.security.ssl.SSLEngineImpl$DelegatedTask.run(SSLEngineImpl.java:1008)
        at org.apache.kafka.common.network.SslTransportLayer.runDelegatedTasks(SslTransportLayer.java:430)
        at org.apache.kafka.common.network.SslTransportLayer.handshakeUnwrap(SslTransportLayer.java:514)
        at org.apache.kafka.common.network.SslTransportLayer.doHandshake(SslTransportLayer.java:368)
        at org.apache.kafka.common.network.SslTransportLayer.handshake(SslTransportLayer.java:291)
        at org.apache.kafka.common.network.KafkaChannel.prepare(KafkaChannel.java:178)
        at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:543)
        at org.apache.kafka.common.network.Selector.poll(Selector.java:481)
        at kafka.network.Processor.poll(SocketServer.scala:989)
        at kafka.network.Processor.run(SocketServer.scala:892)
        at java.base/java.lang.Thread.run(Thread.java:829)}

)
2022-02-16 22:52:48,628 INFO [SocketServer listenerType=ZK_BROKER, nodeId=0] Failed authentication with /10.233.104.145 (SSL handshake failed) (org.apache.kafka.common.network.Selector) [data-plane-kafka-network-thread-0-ListenerName(INTERNAL-9094)-SASL_SSL-12]

But what is most interesting is that 1 out of 5 requests works correctly

[root@kafka-cli tmp]# /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka-cluster-kafka-bootstrap:9094 --command-config /tmp/alice.properties --list
__consumer_offsets
__strimzi-topic-operator-kstreams-topic-store-changelog
__strimzi_store_topic
a-replica-topic
replica-topic
strimzi.cruisecontrol.metrics
strimzi.cruisecontrol.modeltrainingsamples
strimzi.cruisecontrol.partitionmetricsamples
x-replica-topic
x_messages

I no longer know which part of the configuration and settings to make changes.
Please help me with to sole the problem or suggest which certificates and storages are used by java in Kafka brokers

Thanks in advance

[scholzj]

From the client and broker error, I think it looks like this is not related to OAuth / Keycloak but to the Kafka connection it self. So that would point to wrongly configured truststore. How did you create / get the truststore? It is a bit weird if one out of 5 requests works, since in most cases the truststore would either work or not work. Might be interesting to run it with javax.net.debug=ssl system property to see where it connects when it fails and when it succeeds. The occasional success could be also related to some networking issues - something intercepting the connection or something like that.

[geniymaxdem]

As you can see in logs, javax.net.debug have a value - ssl

javax.net.ssl|ERROR|58|data-plane-kafka-network-thread-0-ListenerName(INTERNAL-9094)-SASL_SSL-12|2022-02-16 22:52:48.627 GMT|TransportContext.java:341|Fatal (UNEXPECTED_MESSAGE): Unexpected handshake message: client_hello (
"throwable" : {
  javax.net.ssl.SSLProtocolException: Unexpected handshake message: client_hello
        at java.base/sun.security.ssl.Alert.createSSLException(Alert.java:129)
        at java.base/sun.security.ssl.Alert.createSSLException(Alert.java:117)
        at java.base/sun.security.ssl.TransportContext.fatal(TransportContext.java:336)
        at java.base/sun.security.ssl.TransportContext.fatal(TransportContext.java:292)
        at java.base/sun.security.ssl.TransportContext.fatal(TransportContext.java:283)
        at java.base/sun.security.ssl.HandshakeContext.dispatch(HandshakeContext.java:437)
        at java.base/sun.security.ssl.SSLEngineImpl$DelegatedTask$DelegatedAction.run(SSLEngineImpl.java:1074)
        at java.base/sun.security.ssl.SSLEngineImpl$DelegatedTask$DelegatedAction.run(SSLEngineImpl.java:1061)
        at java.base/java.security.AccessController.doPrivileged(Native Method)
        at java.base/sun.security.ssl.SSLEngineImpl$DelegatedTask.run(SSLEngineImpl.java:1008)
        at org.apache.kafka.common.network.SslTransportLayer.runDelegatedTasks(SslTransportLayer.java:430)
        at org.apache.kafka.common.network.SslTransportLayer.handshakeUnwrap(SslTransportLayer.java:514)
        at org.apache.kafka.common.network.SslTransportLayer.doHandshake(SslTransportLayer.java:368)
        at org.apache.kafka.common.network.SslTransportLayer.handshake(SslTransportLayer.java:291)
        at org.apache.kafka.common.network.KafkaChannel.prepare(KafkaChannel.java:178)
        at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:543)
        at org.apache.kafka.common.network.Selector.poll(Selector.java:481)
        at kafka.network.Processor.poll(SocketServer.scala:989)
        at kafka.network.Processor.run(SocketServer.scala:892)
        at java.base/java.lang.Thread.run(Thread.java:829)}

)
2022-02-16 22:52:48,628 INFO [SocketServer listenerType=ZK_BROKER, nodeId=0] Failed authentication with /10.233.104.145 (SSL handshake failed) (org.apache.kafka.common.network.Selector) [data-plane-kafka-network-thread-0-ListenerName(INTERNAL-9094)-SASL_SSL-12]

What kind of truststore do you mean?
If you talk about tlsTrustedCertificates - it's created from ca cert from Keycloak server
If you talk about truststore in file alice.properties - I created it used your official manual

KAFKA_HOST_PORT=kafka-cluster-kafka-bootstrap.kafka.svc:9094 \
SSO_HOST=keycloak.service.rowi.tech \
SSO_HOST_PORT=$SSO_HOST:443 \
STOREPASS=storepass

echo "Q" | openssl s_client -showcerts -connect $KAFKA_HOST_PORT 2>/dev/null | awk ' /BEGIN CERTIFICATE/,/END CERTIFICATE/ { print $0 } ' > /tmp/my-cluster-kafka.crt

echo "Q" | openssl s_client -showcerts -connect $SSO_HOST_PORT 2>/dev/null | awk ' /BEGIN CERTIFICATE/,/END CERTIFICATE/ { print $0 } ' > /tmp/sso.crt

keytool -keystore /tmp/truststore.p12 -storetype pkcs12 -alias sso -storepass $STOREPASS -import -file /tmp/sso.crt -noprompt

keytool -keystore /tmp/truststore.p12 -storetype pkcs12 -alias my-cluster-kafka -storepass $STOREPASS -import -file /tmp/my-cluster-kafka.crt -noprompt

[scholzj]

As you can see in logs, javax.net.debug have a value - ssl

But that is in the broker. I think you should use it in the client in the first place.

What kind of truststore do you mean?
If you talk about tlsTrustedCertificates - it's created from ca cert from Keycloak server
If you talk about truststore in file alice.properties - I created it used your official manual

I'm talking about the truststore created for Kafka. The commands you shared do not look like anything we recommend. You should take the ca.crt or the ca.p12 from the <cluster-name>-cluster-ca-cert secret and use that.