<a id="co-troubleshoot-authentication"></a>

# Common Authentication Issues

Resolve common authentication issues in Confluent for Kubernetes using the following
steps.

<a id="co-kraft-mds-ssl-auth"></a>

## Issue: KRaft crash-loops with an SSL authentication error

When the KRaftController CR uses mTLS to authenticate with the Metadata Service (MDS)
Kafka cluster, but the `sslClientAuthentication` field is missing or set to
`false`, KRaftController doesn’t present its client certificate during
the mTLS handshake with MDS. After reconciliation, the KRaftController pod
enters a crash loop and displays the following error:

```text
org.apache.kafka.common.errors.SslAuthenticationException: Failed to process
post-handshake messages, SNI host name: empty
```

#### NOTE
This is a known limitation in all current CFK versions and affects both
new installations and upgrades. A resolution is in progress.

**Solution:** Set `sslClientAuthentication: true` under
`spec.dependencies.mdsKafkaCluster.authentication` in the KRaftController
CR, alongside `spec.dependencies.mdsKafkaCluster.tls.enabled: true`.
For details, see [Enable RBAC for KRaft controller](co-rbac.md#co-rbac-kraft) and
[Client-side mTLS authentication for Kafka and KRaft](co-authenticate-kafka.md#co-authenticate-kafka-client-mtls).

## Issue: Unable to connect to identity provider with self-signed certificates when using OAuth/OIDC authentication for Kafka

The current set of SSL properties in Kafka listeners doesn’t let you
connect to your identity provider with self-signed certificates. You
receive an error in the Kafka pod:

```text
[2024-04-17 13:33:53,712] ERROR Exiting Kafka due to fatal exception during startup. (kafka.Kafka$)
org.apache.kafka.common.KafkaException: org.apache.kafka.common.KafkaException:
The OAuth validator configuration encountered an error when initializing the VerificationKeyResolver
```

**Workaround:** Use one of the following options.

* Specify a custom truststore through JVM arguments in the
  `spec.configOverrides.jvm` section of the Kafka custom resource:
  ```yaml
  kind: Kafka
  spec:
    configOverrides:
      jvm:
        - "-Djavax.net.ssl.trustStoreType=JKS"
        - "-Djavax.net.ssl.trustStore=/mnt/jvmtruststore/truststore.jks"
        - "-Djavax.net.ssl.trustStorePassword=mystorepassword"
  ```
* Add the truststore to the Kafka listeners as JAAS configuration, using
  the parameter `unsecuredLoginStringClaim_sub="thePrincipalName"`.
  ```yaml
  kind: Kafka
  spec:
    configOverrides:
      server:
        - listener.name.controller.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required clientId="${file:/mnt/secrets/oauth-jass/oauth.txt:clientId}" clientSecret="${file:/mnt/secrets/oauth-jass/oauth.txt:clientSecret}" refresh_ms="3000" ssl.truststore.location="/mnt/sslcerts/truststore.jks" ssl.truststore.password="mystorepassword" unsecuredLoginStringClaim_sub="thePrincipalName";
  ```

  Alternatively, you can use a secret to pass the JAAS config passthrough
  information as described in [Create server-side SASL/PLAIN credentials using JAAS config pass-through](co-authenticate-kafka.md#co-sasl-plain-server-jaas-passthrough).

## Issue: Kafka brokers fail to start when configuring OAuth/OIDC authentication

OAuth listeners require `unsecuredLoginStringClaim_sub` in the SASL JAAS
configuration to communicate with the token endpoint. CFK automatically creates
the setting only for the token listener.

**Workaround:**

When you configure OAuth/OIDC for a different Kafka listener than the token
listener, you must manually update the `configOverrides` in the Kafka CR as
below:

```yaml
kind: Kafka
spec:
  configOverrides:
    server:
    - listener.name.<listenerName>.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required scope=<scope> unsecuredLoginStringClaim_sub=\"thePrincipalName\"
```

`<listenerName>` can be a Kafka listener name, such as `external`,
`internal`, or `replication`.

**Fixed Versions:** Upgrade to CFK 2.11.0, 2.10.1, or 2.9.5, which fixes this
issue for all Kafka listeners.

## Issue: Use different authentication for internal and external listeners

When you set up SASL/PLAIN and SASL/PLAIN with LDAP listeners in the Kafka CR,
you get the following error:

```text
WARN [Producer clientId=producer-client] Connection to node -1
(<broker_hostname>/<broker_ip>:9092) terminated during authentication. This
may happen due to any of the following reasons: (1) Authentication failed due
to invalid credentials with brokers older than 1.0.0, (2) Firewall blocking
Kafka TLS traffic (eg it may only allow HTTPS traffic), (3) Transient network
issue. (org.apache.kafka.clients.NetworkClient)
```

**Workaround:**

To implement both a SASL/PLAIN listener and a SASL/PLAIN with LDAP listener in
the Kafka cluster, you must configure the SASL/PLAIN listener with
`authentication.jaasConfigPassThrough`.

You can use the following example configuration to set up an internal listener
with SASL/PLAIN and an external listener with SASL/PLAIN with LDAP:

1. Create a file, `creds-kafka-sasl-users.conf` with the following content.
   ```text
   sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
    username="kafka" \
    password="kafka-secret" \
    user_kafka="kafka-secret";
   ```
2. Create a secret credential.
   ```bash
   kubectl create secret generic credential \
     --from-file=plain-jaas.conf=./creds-kafka-sasl-users.conf \
     --namespace <namespace>
   ```
3. Specify the JAAS configuration passthrough in the Kafka CR and apply the
   change with the `kubectl apply -f` command.
   ```yaml
   kind: Kafka
   spec:
     listeners:
       internal:
         authentication:
           type: plain
           jaasConfigPassThrough:
             secretRef: credential
         tls:
           enabled: true
       external:
         authentication:
           type: ldap
         tls:
           enabled: true
   ```

## Issue: Cluster Link creation failed on mTLS with RBAC enabled

When creating a Cluster Link with mTLS authentication to an Kafka cluster
with RBAC enabled, you get the following error:

```text
kafka rest API request failed: https://kafka.src-dev-1-cluster-<my-cluster-link>: The server experienced an unexpected error when processing the request. (40002)
```

The Kafka broker logs would show:

```text
java.lang.IllegalArgumentException: Could not find a 'KafkaServer' or 'tokensasl.KafkaServer' entry in the JAAS configuration. System property 'java.security.auth.login.config' is not set`
```

The root cause of the problem is that the Cluster Link would default to the
TOKENSASL listener. To override this behavior, force Cluster Link creation on
the INTERNAL listener.

**Solution:** Add the below to the source ClusterLink CR.

```yaml
kind: ClusterLink
spec:
  configs:
    local.security.protocol: SSL
    local.listener.name: INTERNAL
```