# Configure Authentication for Kafka, KRaft, and MDS Using Confluent for Kubernetes

This document presents the supported authentication methods and describes how
to configure authentication for Kafka, KRaft, and MDS using Confluent for Kubernetes (CFK).

Any configuration difference between KRaft and ZooKeeper-based deployments are
noted where applicable.

Kafka, KRaft, and MDS are configured without authentication by default.

For more details on security concepts in Confluent Platform, see [Security in Confluent
Platform](https://docs.confluent.io/platform/current/kafka/overview-authentication-methods.html).

For a comprehensive tutorial scenario for configuring authentication, see
[Deploy Secure Confluent Platform](https://github.com/confluentinc/confluent-kubernetes-examples/tree/master/security/secure-authn-encrypt-deploy).

<a id="co-authenticate-kafka"></a>

## Configure authentication to access Kafka and KRaft

This section describes the following methods for the server and client-side Kafka
and KRaft authentication:

- [SASL/PLAIN authentication](#co-authenticate-kafka-plain)
- [SASL/PLAIN with LDAP authentication](#co-authenticate-kafka-plain-ldap)
- [OAuth/OIDC authentication](#co-authenticate-kafka-oauth)
- [Passwordless OAuth/OIDC authentication](#co-authenticate-kafka-client-assertion)
- [mTLS authentication](#co-authenticate-kafka-mtls)

### Notes and considerations

* When you enable RBAC with LDAP or OAuth, CFK adds a token listener with the
  authentication based upon the Metadata Service (MDS). Other Confluent Platform components connect to
  Kafka using the *token listener authentication*, ignoring the authentication
  defined in the component custom resource (CR) under
  `spec.dependencies.kafka.authentication`.
* When you enable RBAC with mTLS, the internal listener is not overridden. And
  Confluent Platform components connect to Kafka using the *internal listener authentication*
  as defined in the component CR under
  `spec.dependencies.kafka.authentication`.
* When you enable RBAC and use the custom listener instead of the internal listener for
  communication, the *custom listener authentication* setting is not overridden
  and is used in place of token listener. The custom listener must support the MDS
  authentication method for this setup to work.

<a id="co-authenticate-kafka-plain"></a>

### SASL/PLAIN authentication

SASL/PLAIN is a simple username/password mechanism that is typically used with
TLS network encryption to implement secure authentication.

The username is used as the authenticated principal, which can then be used in
authorization.

#### Server-side SASL/PLAIN authentication for Kafka and KRaft

Configure the server-side SASL/PLAIN authentication for Kafka and KRaft.

1. Set up credentials.

   You can use the JAAS and JAAS pass-through mechanisms to set up SASL/PLAIN
   credentials.
   * [Create server-side SASL/PLAIN credentials using JAAS config](#co-sasl-plain-server-jaas)
   * [Create server-side SASL/PLAIN credentials using JAAS config pass-through](#co-sasl-plain-server-jaas-passthrough)
2. In the Kafka and KRaftController custom resources (CRs), configure the listener to use
   SASL/PLAIN as the authentication mechanism:
   ```yaml
   kind: <Kafka or kRaftController>
   spec:
     listeners:
       <listener type>:
         authentication:
           type: plain                 --- [1]
           jaasConfig:                 --- [2]
             secretRef:                --- [3]
           jaasConfigPassThrough:      --- [4]
             secretRef:                --- [5]
             directoryPathInContainer: --- [6]
   ```

   * [1] Required. Set to `plain`.
   * [2] When you use `jaasConfig`, you provide the user names and passwords, and
     CFK automates configuration. For example, when you add, remove, or update
     users, CFK automatically updates the JAAS config. This is the recommended
     way to configure SASL/PLAIN for Kafka.
   * One of [3], [5], or [6] is required. Only specify one.
   * [3] Provide the name of the Kubernetes secret that you created in the
     [previous section](#co-sasl-plain-server-jaas).
   * [4] If you have customizations, such as using a custom login handler, you
     can bypass the CFK automation and provide the configuration directly using
     `jaasConfigPassThrough`.
   * [5] Provide a Kubernetes secret that you created in the
     [previous section](#co-sasl-plain-server-jaas-passthrough) with the
     expected key and the value.
   * [6] Provide the directory path in the container that you set up for the
     credentials in the [previous section](#co-sasl-plain-server-jaas-passthrough).

     See [CFK GitHub examples](https://github.com/confluentinc/confluent-kubernetes-examples/tree/master/security/configure-with-vault)
     for more information on using the `directoryPathInContainer` property with
     Vault.

<a id="co-authenticate-kafka-client-plain"></a>

#### Client-side SASL/PLAIN authentication for Kafka and KRaft

Configure the client-side SASL/PLAIN authentication for other Confluent
components to authenticate to Kafka and KRaft.

1. Set up credentials.

   You can use the JAAS and JAAS pass-through mechanisms to set up SASL/PLAIN
   credentials.
   * [Create client-side SASL/PLAIN credentials using JAAS config](#co-sasl-plain-client-jaas)
   * [Create client-side SASL/PLAIN credentials using JAAS config pass-through](#co-sasl-plain-client-jaas-passthrough)
2. For each of the Confluent components that communicates with Kafka or KRaft, configure
   SALS/PLAIN authentication in the component CR as below:
   ```yaml
   kind: <Confluent component>
   spec:
     dependencies:
       <kafka or kRaftController>:
         authentication:
           type: plain                 --- [1]
           jaasConfig:                 --- [2]
             secretRef:                --- [3]
           jaasConfigPassThrough:      --- [4]
             secretRef:                --- [5]
             directoryPathInContainer: --- [6]
   ```

   * [1] Required. Set to `plain`.
   * [2] When you use `jaasConfig`, you provide the user names and passwords, and
     CFK automates configuration. For example, when you add, remove, or update
     users, CFK automatically updates JAAS config.
   * [3], [5] or [6] is required. Specify only one.
   * [3] Provide a Kubernetes secret you created in the [previous section](#co-sasl-plain-client-jaas) for this Confluent component to authenticate to
     Kafka and KRaft.
   * [4] An alternate way to configure JAAS is to use `jaasConfigPassThrough`.
     If you have customizations, such as using custom login handlers, you can
     bypass the CFK automation and provide the configuration directly.
   * [5] Provide a Kubernetes secret that you created in the [previous section](#co-sasl-plain-client-jaas-passthrough).
   * [6] Provide the directory path in the container that you set up in the
     [previous section](#co-sasl-plain-client-jaas-passthrough).

<a id="co-authenticate-kafka-plain-ldap"></a>

### SASL/PLAIN with LDAP authentication

SASL/PLAIN with LDAP callback handler is a variation of SASL/PLAIN. When you use
SASL/PLAIN with LDAP for authentication, the username principals and passwords
are retrieved from an LDAP server.

<a id="co-authenticate-kafka-plain-ldap-server-side"></a>

#### Server-side SASL/PLAIN with LDAP for Kafka and KRaft

You must set up an LDAP server, for example, Active Directory (AD), before
configuring and starting up a Kafka cluster with the SASL/PLAIN with LDAP
authentication. For more information, see [Configuring Kafka Client
Authentication with LDAP](https://docs.confluent.io/platform/current/kafka/authentication_sasl/client-authentication-ldap.html).

1. Set up credentials.

   You can use the JAAS and JAAS pass-through mechanisms to set up SASL/PLAIN
   credentials.
   * [Create server-side SASL/PLAIN LDAP credentials using JAAS config](#co-sasl-ldap-server-jaas)
   * [Create server-side SASL/PLAIN LDAP credentials using JAAS config pass-through](#co-sasl-ldap-server-jaas-passthrough)

   #### NOTE
   To implement both a SASL/PLAIN listener and a SASL/PLAIN with LDAP listener
   in your Kafka cluster, the SASL/PLAIN listener must be configured with
   `authentication.jaasConfigPassThrough`. Configuring
   `authentication.jaasConfig` for the SASL/PLAIN will cause the clients
   connection to the SASL/PLAIN with the LDAP listener to fail with a
   `java.io.EOFException`.
2. Configure the listeners in the Kafka or KRaft custom resource (CR):
   ```yaml
   kind: <Kafka or kRaftController>
   spec:
     listeners:
       <listener type>:
         authentication:
           type: ldap                 --- [1]
           jaasConfig:                --- [2]
             secretRef:               --- [3]
           jaasConfigPassThrough:     --- [4]
             secretRef:               --- [5]
             directoryPathInContainer:--- [6]
   ```

   * [1] Required for the SASL/PLAIN with LDAP authentication for the specific
     Kafka or KRaft listener type.
   * [2] When you use `jaasConfig` to pass credentials, you provide the user
     name and password, and CFK automates configuration. When you
     add, remove, or update the user, CFK automatically updates the JAAS
     configuration. This is the recommended way to configure SASL/PLAIN LDAP for Kafka.
   * [3] Provide the name of the Kubernetes secret that you created in the
     [previous section](#co-sasl-ldap-server-jaas) for inter-broker authentication.
   * [4] An alternate way to configure JAAS is to use `jaasConfigPassThrough`.
     If you have customizations, such as using a custom login handler, you can
     bypass the CFK automation and provide the configuration directly.
   * [2] [4] To configure authentication type `ldap` on external or custom
     listener, you do not need to specify `jaasConfig` or
     `jaasConfigPassThrough`.
   * [5] Provide the name of the Kubernetes secret that you created in the
     [previous section](#co-sasl-ldap-server-jaas-passthrough) for
     inter-broker authentication.
   * [6] Provide the directory path in the container that you set up in the
     [previous section](#co-sasl-ldap-server-jaas-passthrough).
3. Configure the identity provider in the Kafka or KRaft CR, and apply the CR with the
   `kubectl apply` command:
   ```yaml
   kind: <Kafka or kRaftController>
   spec:
     identityProvider:                --- [1]
       type: ldap                     --- [2]
       ldap:                          --- [3]
         address:                     --- [4]
         authentication:              --- [5]
           type:                      --- [6]
           simple:                    --- [7]
         tls:
           enabled:                   --- [8]
         configurations:              --- [9]
   ```

   * [1] Required for the Kafka authentication type `ldap`. Specifies the
     identity provider configuration.

     When the MDS is enabled, this property is ignored, and the LDAP
     configuration in `spec.services.mds.provider` is used.
   * [2] Required.
   * [3] This block includes the same properties used in the
     `spec.services.mds.provider.ldap` block in this CR.
   * [4] Required. The address of the LDAP server, for example,
     `ldaps://ldap.confluent.svc.cluster.local:636`.
   * [5] Required. The authentication method to access the LDAP server.
   * [6] Required. Specify `simple` or `mtls`.
   * [7] Required if the authentication type ([6]) is set to `simple`.
   * [8] Required if the authentication type ([6]) is set to `mtls`. Set to
     `true`.
   * [9] Required. The LDAP configuration settings.
4. Mixing `FileBasedLoginModule` (used by SASL/PLAIN) and `PlainLoginModule`
   (used by SASL/PLAIN with LDAP) in the JAAS configs is not supported and can
   cause the Kafka broker to fail. Therefore, to use both SASL/PLAIN listener and
   a SASL/PLAIN with LDAP listener in a Kafka cluster, you must configure the
   SASL/PLAIN listener with JAAS config pass-through which enforces the use of
   the `PlainLoginModule` class.

   To implement both a SASL/PLAIN listener and a SASL/PLAIN with LDAP listener
   for a Kafka cluster in KRaft mode, in addition to using JAAS config
   pass-through, you must also provide the config override setting,
   `listener.name.controller.plain.sasl.jaas.config` for the controller
   listener in the Kafka CR.

   An example Kafka CR snippet:
   ```yaml
   kind: Kafka
   spec:
     configOverrides:
       server:
       - listener.name.controller.plain.sasl.jaas.config=${file:/mnt/secrets/<internal_listener_SASLPLAIN_secret_name>/plain-jaas.conf:sasl.jaas.config}
   ```

   See an [end-to-end sample scenario](https://github.com/confluentinc/confluent-kubernetes-examples/tree/master/security/kraft_sasl_ldap)
   in the CFK example repository in GitHub.

An example kRaftController CR:

```yaml
kind: KRaftController
spec:
  identityProvider:
    type: ldap
    ldap:
      address: ldaps://ldap.operator.svc.cluster.local:636
      authentication:
        simple:
          secretRef: credential
        type: simple
      configurations:
        groupMemberAttribute: member
        groupMemberAttributePattern: CN=(.*),DC=test,DC=com
        groupNameAttribute: cn
        groupObjectClass: group
        groupSearchBase: dc=test,dc=com
        userMemberOfAttributePattern: CN=(.*),DC=test,DC=com
        userNameAttribute: cn
        userObjectClass: organizationalRole
        userSearchBase: dc=test,dc=com
      tls:
        enabled: true
```

<a id="co-authenticate-kafka-client-plain-ldap"></a>

#### Client-side SASL/PLAIN with LDAP for Kafka and KRaft

When Kafka is configured with SASL/PLAIN with LDAP, Confluent components and
clients authenticate to Kafka as SASL/PLAIN clients. The clients must
authenticate as users in LDAP.

See [Client-side SASL/PLAIN authentication for Kafka and KRaft](#co-authenticate-kafka-client-plain) for configuration details.

<a id="co-authenticate-kafka-mtls"></a>

### mTLS authentication

#### Server-side mTLS authentication for Kafka and KRaft

mTLS utilizes TLS certificates as an authentication mechanism. The certificate
provides the identity.

The certificate Common Name (CN) is used as the authenticated principal, which
can then  be used in authorization.

Staring in CFK 2.10, you can configure and use two authentication methods (mTLS
and OAuth or mTLS and LDAP) for Kafka listeners, one with mTLS and another
supported method.

Configure the listeners as below, in the Kafka or KRaftController CR, to use mTLS as the
authentication mechanism:

```yaml
kind: Kafka
spec:
  listeners:
    <listener type>:
      authentication:
        type:                                          --- [1]
        principalMappingRules:                         --- [2]
        - RULE:.*CN[\\s]?=[\\s]?([a-zA-Z0-9.]*)?.*/$1/
        mtls:                                            --- [3]
          sslClientAuthentication:                       --- [4]
          principalMappingRules:                         --- [5]
          - RULE:.*CN[\\s]?=[\\s]?([a-zA-Z0-9.]*)?.*/$1/
      tls:
        enabled: true                                  --- [6]
```

* [1] Required. Set to `mTLS`.
  * Set to `mtls` for setting up mTLS as the only authentication method.
  * Set to another methods other than `mtls` described in this topic. If set
    to `mtls`, the listener will be configure to only use the mTLS
    authentication.
* [2] Optional. This specifies a mapping rule that extracts the principal name
  from the certificate Common Name (CN) when using mTLS only authentication.

  The regular expression (regex) used in the mapping rule is [Java mapping API](https://docs.oracle.com/javase/8/docs/api/java/util/regex/Matcher.html#matches).

  Shorthand character classes need to be escaped with another backslash. For
  example, to use a whitespace (`\s`), specify `\\s`.
* [3] Required for mTLS authentication in the server-side mTLS authentication
  mode.
* [4] Required. Set to `required` to enable mandatory mTLS authentication from
  the client-side and to enforce certificate presentation.

  Valid values are `requested` and `required`. Use `requested` for
  optional mTLS.
* [5] Optional. This specifies a principal mapping rules list from the
  certificate of the client when using mTLS authentication. See [2] for
  details.
* [6] Required for mTLS authentication. Set to `true`.

<a id="co-authenticate-kafka-client-mtls"></a>

#### Client-side mTLS authentication for Kafka and KRaft

For each of the Confluent components that communicates with Kafka or KRaft, configure the
mTLS authentication mechanism in the component CR as below:

```yaml
kind: <Confluent component>
spec:
  dependencies:
    <kafka or kRaftController>:
      authentication:
        type: mtls               --- [1]
      tls:
        enabled: true            --- [2]
```

* [1] Required. Set to `mtls`.
* [2] Required for mTLS authentication. Set to `true`.

<a id="co-authenticate-kafka-oauth"></a>

### OAuth/OIDC authentication

Open Authentication (OAuth) 2.0 is an open-standard authorization protocol that
provides applications the ability for securely designated access. You can
leverage your own identity provider and centralize identity management across
your Confluent Platform and other service deployments on the cloud and on-premises.

Starting with CFK 2.9 and Confluent Platform 7.7, you can configure Confluent components with
OAuth/OpenID Connect (OIDC), an OAuth-based authentication mechanism.

For the OAuth overview in Confluent Platform, see [OAuth 2.0 for Confluent Platform](https://docs.confluent.io/platform/current/security/authentication/oauth-oidc/overview.html).

<a id="co-authenticate-kafka-server-oauth"></a>

#### Server-side OAuth/OIDC authentication for Kafka and KRaft

Configure a Kafka or KRaft listener as below in the Kafka or KraftController CR
to use OAuth/OIDC as the authentication mechanism.

<a id="co-authenticate-kafka-oauth-settings"></a>
```yaml
kind: <Kafka or kRaftController>
spec:
  listeners:
    <listener type>:
      authentication:
        type: oauth                       --- [1]
        jaasConfig/jaasConfigPassThrough:
          secretRef:                      --- [2]
        oauthSettings:
          groupsClaimName:                --- [3]
          subClaimName:                   --- [4]
          audience:                       --- [5]
          expectedIssuer:                 --- [6]
          jwksEndpointUri:                --- [7]
          tokenEndpointUri:               --- [8]
          scope:                          --- [9]
          loginConnectTimeoutMs:          --- [10]
          loginReadTimeoutMs:             --- [11]
          loginRetryBackoffMs:            --- [12]
          loginRetryMaxBackoffMs:         --- [13]
  tls:                                    --- [14]
```

* [1] Required. Set to `oauth`.
* [2] The secret that contains an OIDC client ID and the client secret for
  authorization and token request to the identity provider (IdP).

  Create the secret that contains two keys with their respective values,
  `clientId` and  `clientSecret` as following:
  ```text
  clientId=<client-id>
  clientSecret=<client-secret>
  ```
* [3] Required. The name of the claim in token for identifying the groups of
  subject in the JSON Web Tokens (JWT). The default value is `groups`.
* [4] The subject name of the JWT (session token). The default value is `sub`.
  Used in SSO.
* [5] Required. The intended consumer of the access token. You can specify a
  comma-delimited list for multiple audiences.
* [6] The issuer URL, which is typically the authorization server’s URL. This
  value is used to compare to the issuer claim in the JWT for verification.
* [7] The JSON Web Key Set (JWKS) URI. It is used to verify any JWT issued by
  the IdP.
* [8] The base URI for the token endpoint. This is required for OAuth
  for inter-broker communication along with `clientId` and `clientSecret`
  in JassConfig or JassConfigPassthrough.
* [9] Required only when your identity provider does not have a default scope or
  your groups claim is linked to a scope.
* [10] Connect timeout with IdP in ms.
* [11] Read timeout with IdP in ms.
* [12] Retry backoff with IdP in ms.
* [13] Max retry backoff with IdP in ms.
* [14] If the IdP is secured with HTTPS, CFK uses this setting to fetch OAuth
  token from the IdP server. Set this setting as describe in
  [Configure Network Encryption for Confluent Platform Using Confluent for Kubernetes](co-network-encryption.md#co-network-encryption).

  If the Kafka cluster does not have TLS  enabled while the IdP server does, you
  may run into a `tls: failed to verify  certificate` error.

<a id="co-authenticate-kafka-idp-self-signed-certs"></a>

##### Connect to IdP with self-signed certificates

The current set of SSL properties in Kafka listeners does not let you connect to your
IdP with self signed certificates. You will 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
```

Use one of the following workaround options:

* Specify a custom trust store 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 trust store to the Kafka listeners as JAAS config, 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-sasl-plain-server-jaas-passthrough).

<a id="co-authenticate-kafka-client-oauth"></a>

#### Client-side OAuth/OIDC authentication for Kafka and KRaft

For each of the Confluent components that communicates with Kafka or KRaft,
configure the OAuth/OIDC authentication mechanism in the component CR as below.

```yaml
kind: <Confluent component>
spec:
  dependencies:
    <kafka or kRaftController>:
      authentication:
        type: oauth                       --- [1]
        jaasConfig/jaasConfigPassThrough:
          secretRef:                      --- [2]
        oauthSettings:
          tokenEndpointUri:               --- [3]
```

* [1] Required. Set to `oauth`.
* [2] The secret that contains a OIDC client ID and the client secret for
  authorization and token request to IdP.

  Create the secret that contains two keys with their respective values,
  `clientId` and  `clientSecret` as following:
  ```text
  clientId=<client-id>
  clientSecret=<client-secret>
  ```
* [3] The base URI for the token endpoint.

For the full list of the OAuth settings, see
[OAuth configuration](#co-authenticate-kafka-oauth-settings).

<a id="co-authenticate-kafka-client-assertion"></a>

### Passwordless OAuth/OIDC authentication with client assertion

Starting with version 8.0, Confluent Platform supports OAuth client assertion, a secure
credential management with passwordless authentication. It uses asymmetric
encryption-based authentication, extending Confluent Platform OAuth, and allows you to:

* Avoid deploying username and password while securing Confluent Platform.
* Streamline and automate client credential rotation on a periodic basis without
  manual intervention for the client applications.

A client assertion is a JSON Web Token (JWT) with a collection of information
for sharing identity and security information, and it is presented as proof of
the client’s identity.

The following client assertion flows are supported in CFK:

* JSON Web Token (JWT) assertion retrieval from file flow

  JWT assertion retrieval from file flow is not recommended for production use cases.
  Instead, you should use local client assertion flow for production.
* Local client assertion flow

In CFK 3.0, OAuth client assertion is supported for the following resources:

* Day 1 components: Kafka, KRaft, MDS, Schema Registry
* Day 2 application resources: KafkaTopic, Kafka REST Class, ConfluentRoleBinding, Schema,
  SchemaExporter, ClusterLinking

<a id="co-authenticate-kafka-server-client-assertion"></a>

#### Server-side OAuth client assertion for Kafka and KRaft

Configure a Kafka listener as below in the Kafka CR to use OAuth client assertion
as the authentication mechanism.

To set up client assertion, first, you must complete the [server-side OAuth
configuration](#co-authenticate-kafka-server-oauth).

<a id="co-authenticate-client-assertion-settings"></a>

For client assertion, configure the additional properties on top of the existing
OAuth configurations:

```yaml
kind: <Kafka or kRaftController>
spec:
  listeners:
    <listener type>:
      authentication:
        type: oauth
        jaasConfig/jaasConfigPassThrough:
          secretRef:
        oauthSettings:
          clientAssertion:
            clientId:                    --- [1]
            jwtTokenFile:                --- [2]
              secretRef:
              directoryPathInContainer:
            localClientAssertion         --- [3]
              issuer:                    --- [4]
              subject:                   --- [5]
              audience:                  --- [6]
              tokenLifetime:             --- [7]
              includeNbfClaim:           --- [8]
              includeJtiClaim:           --- [9]
              privateKeyFile             --- [10]
                secretRef:
                directoryPathInContainer:
                encryptedPrivateKey:     --- [11]
                includeAssertionTemplate:--- [12]
```

* [1] The client ID needed for getting the access token. This field is required
  only if a clientId is mandatory for your IdP.
* For client assertion flow, either `jwtTokenFile` ([2]) or
  `localClientAssertion` ([3]) must be provided.

To set up *JWT assertion retrieval from file*, use the following properties:

* [2] Required for JWT assertion retrieval from file. The location of the file from
  which token(assertion) will be read. Use `SecretRef` or
  `DirectoryPathInContainer` to specify the value.

To set up *local client assertion*, use the following properties:

* [3] Required for local client assertion.
* [4] Required. The name of the claim in token for identifying the issuer in the client assertion JSON Web Tokens (JWT).
* [5] Required. The subject name of the client assertion JWT.
* [6] Required. The intended consumer of the client assertion JWT.
* [7] Default lifetime for the client assertion. The default is 5 minutes.
* [8] Determines whether to include “not before” claim in the client assertion JWT.
* [9] Determines whether to include the JWT ID (JTI) claim in the client assertion JWT.
* [10] Required. If local assertion creation is required, private key used to
  sign the client assertion JWT. Use `secretRef` or
  `directoryPathInContainer` to provide the key.
* [11] Required. Set to `true` if the private key is an encrypted PKCS#8 key
  and if passphrase is used to decrypt the private key.
* [12] Set to `true` if the assertion template JSON is provided which contains
  additional header and payload claims to be included in the client assertion.

<a id="co-authenticate-kafka-client-client-assertion"></a>

#### Client-side OAuth client assertion for Kafka and KRaft

For each of the Confluent component (currently in Confluent Platform 8.0, Schema Registry only) to
authenticate with Kafka or KRaft using OAuth client assertion, configure the
client-side OAuth client assertion in the component CR as below.

For KRaft, the `authentication` object is under
`dependencies.kRaftController.controllerListener.authentication`.

To set up client assertion, first, you must complete the [client-side OAuth
configuration](#co-authenticate-kafka-client-oauth).

For client assertion, configure the additional properties on top of the existing
OAuth configurations:

```yaml
kind: <Confluent component>
spec:
  dependencies:
    <kafka or kRaftController>:
      authentication:
        type: oauth        --- [1]
        oauthSettings:
          clientAssertion: --- [2]
```

* [1] Required.
* [2] See [the client assertion properties](#co-authenticate-client-assertion-settings) for a list of properties you can use.

The following is a sample snippet of Schema Registry to authenticate with Kafka using local
client assertion:

```yaml
  kind: SchemaRegistry
  spec:
    dependencies:
      kafka:
        bootstrapEndpoint: kafka.operator.svc.cluster.local:9071
        authentication:
          type: oauth
          oauthSettings:
            tokenEndpointUri: http://keycloak:8080/realms/sso_test/protocol/openid-connect/token
            clientAssertion:
              clientId: private-key-client
#             jwtTokenFile:
#               secretRef: jwt-secret
              localClientAssertion:
                issuer: private-key-client
                subject: private-key-client
                audience: http://keycloak:8080/realms/sso_test/protocol/openid-connect/token
                privateKeyFile:
                  secretRef: private-key
                includeJtiClaim: tru
```

<a id="co-authenticate-mds"></a>

## Configure authentication to access MDS

You can use the following authentication methods to communicate with MDS.

- [Bearer authentication](#co-authenticate-mds-bearer): You use the bearer
  token issued by MDS.
- [OAuth/OIDC authentication](#co-authenticate-mds-oauth): You use the
  token, provided by your own identity provider and accepted by MDS.
- [mTLS authentication](#co-authenticate-mds-mtls): You use the principal
  extracted from the certificate data that MDS server gets when authenticating
  client.

<a id="co-authenticate-mds-bearer"></a>

### Bearer authentication

#### Client-side Bearer authentication for MDS

1. [Set up client-side Bearer credentials for MDS](#co-bearer-server-creds).
2. For each of the Confluent components that communicate with MDS, configure
   Bearer authentication mechanism in the component CR. And then apply the CR
   with the `kubectl apply` command.
   ```yaml
   kind: <Confluent component>
   spec:
     dependencies:
       mds:
         authentication:
           type: bearer               --- [1]
           bearer:
             secretRef:               --- [2]
             directoryPathInContainer --- [3]
   ```

   * [1] Required. Set to `bearer`.
   * [2] or [3] Required. Do not specify both.
   * [2] To load username and secret, set to the Kubernetes secret
     you created in the previous section.
   * [3] Provide the path in the container in the
     [previous section](#co-bearer-server-creds). The expected file,
     `bearer.txt`, must exist in the specified directory path.

<a id="co-authenticate-mds-oauth"></a>

### OAuth/OIDC authentication

<a id="co-authenticate-mds-server-oauth"></a>

#### Server-side OAuth/OIDC authentication for MDS

Configure the MDS as below in the Kafka CR to use OAuth/OIDC as the
authentication mechanism:

```yaml
kind: Kafka
spec:
  services:
    mds:
      provider:
        oidc:
          clientCredentials:
            secretRef:                  --- [1]
        oauth:
          configurations:
            groupsClaimName:            --- [2]
            subClaimName:               --- [3]
            expectedIssuer:             --- [4]
            jwksEndpointUri:            --- [5]
```

* [1] Required only when enabling SSO. The secret that contains a OIDC client ID
  and the client secret for authorization and token request to IdP.

  Create the secret that contains two keys with their respective values,
  `clientId` and  `clientSecret` as following:
  ```text
  clientId=<client-id>
  clientSecret=<client-secret>
  ```
* [2] Groups in JSON Web Tokens (JWT). The default value is `groups`.

  JWT is issued by Confluent (MDS) to maintain the session. This contains
  information passed by IdP in id token along with custom claims (if any) added
  by the MDS.
* [3] The subject name of the JWT. The default value is `sub`.
* [4] The issuer URL, which is typically the authorization server’s URL. This
  value is used to compare to the issuer claim in the JWT for verification.
* [5] The JSON Web Key Set (JWKS) URI. It is used to verify any JWT issued by
  the IdP.

For the full list of the OAuth settings, see
[OAuth configuration](#co-authenticate-kafka-oauth-settings).

<a id="co-authenticate-mds-client-oauth"></a>

#### Client-side OAuth/OIDC authentication for MDS

For each of the Confluent components that communicate with MDS, configure
OAuth/OIDC authentication mechanism in the component CR:

```yaml
kind: <Confluent component>
spec:
  dependencies:
    mds:
      authentication:
        type: oauth                --- [1]
        oauth:
          secretRef:               --- [2a]
          directoryPathInContainer --- [2b]
          configuration:
            tokenEndpointUri:      --- [3]
```

* [1] Required. Set to `oauth`.
* [2a] or [2b] Specify only one setting.
* [2a] The secret that contains the OIDC client ID and the client secret for
  authorization and token request to the identity provider.

  Create the secret that contains two keys with their respective values,
  `clientId` and  `clientSecret` as following:
  ```text
  clientId=<client-id>
  clientSecret=<client-secret>
  ```
* [2b] The OIDC client ID and the client secret for authorization and token
  request to the identity provider (IdP).

  See  [Provide secrets for Confluent Platform component CR](co-credentials.md#co-vault-category-1) for providing the credential and required
  annotations when using Vault.
* [3] The base URI for the token endpoint.

For the full list of the OAuth settings, see
[OAuth configuration](#co-authenticate-kafka-oauth-settings).

<a id="co-authenticate-mds-client-assertion"></a>

### Passwordless OAuth/OIDC authentication with client assertion

Starting with version 8.0, Confluent Platform supports OAuth client assertion. For the brief
overview of client assertion in CFK, see
[OAuth Client Assertion](#co-authenticate-kafka-client-assertion).

<a id="co-authenticate-mds-server-client-assertion"></a>

#### Server-side OAuth client assertion for MDS

No client assertion specific configuration is needed for MDS beyond the
[server-side OAuth configurations](#co-authenticate-mds-server-oauth).

<a id="co-authenticate-mds-client-client-assertion"></a>

#### Client-side OAuth client assertion for MDS

For each of the Confluent components (currently in Confluent Platform 8.0, only for Schema Registry) to
authenticate with MDS using OAuth client assertion, configure the client-side
OAuth client assertion in the component CR as below.

To set up client assertion, first, you must complete the [client-side OAuth
configuration](#co-authenticate-mds-client-oauth).

For client assertion, configure the additional properties on top of the existing
OAuth configurations:

```yaml
kind: <Confluent component>
spec:
  dependencies:
    mds:
      authentication:
        type: oauth        --- [1]
        oauth:
          clientAssertion: --- [2]
```

* [1] Required.
* [2] See [the client assertion properties](#co-authenticate-client-assertion-settings) for a list of properties you can
  use.

<a id="co-authenticate-mds-mtls"></a>

### mTLS authentication

#### Server-side mTLS authentication for MDS

Starting in CFK 2.10 and Confluent Platform 7.8, MDS supports mTLS authentication either in
single authentication mode or dual authentication mode.

#### NOTE
Updating an existing RBAC setup to use mTLS RBAC is not currently supported in
CFK.

Configure the MDS as below in the Kafka CR to use the mTLS authentication
mechanism:

```yaml
kind: Kafka
spec:
  services:
    mds:
      provider:
        mtls:                               --- [1]
          sslClientAuthentication:          --- [2]
          principalMappingRules:            --- [3]
            - ^CN=([a-zA-Z0-9.]*).*$/$1/L
```

* [1] Required.
* [2] Required. Set to `required` to enable mandatory mTLS authentication from
  the client-side and to enforce certificate presentation.

  Valid values are `requested` and `required`. Use `requested` for
  optional mTLS.
* [3] Mapping rules for extracting principal from the certificate.

<a id="co-authenticate-mds-mtls-client"></a>

#### Client-side mTLS authentication for MDS

1. For each of the Confluent components that communicates with MDS, configure
   mTLS authentication mechanism in the component CR:
   ```yaml
   kind: <Confluent component>
   spec:
     dependencies:
       mds:
         authentication:
           type:                    --- [1]
           sslClientAuthentication: --- [2]
   ```

   * [1] If only using mTLS, set to `mtls`.

     If configuring two authentication methods for MDS, set to `bearer` or `oauth`.
   * [2] can be a boolean true or false and will enable mTLS authentication from
     the client side.
2. Configure the Kafka CR to specify that `kafkaRest` will connect with MDS over
   mTLS.

   `kafkaRest` provides the REST client configuration for the MDS when RBAC is
   enabled.
   ```yaml
   kind: kafka
   spec:
     dependencies:
       kafkaRest:
         authentication:
           type: mtls
           sslClientAuthentication: true
         tls:
           enabled: true
           secretRef: tls-kafka
   ```

For an example scenario configuration, see the [CFK examples repository in
GitHub](https://github.com/confluentinc/confluent-kubernetes-examples/tree/master/security/mds-mtls).

<a id="co-authenticate-mds-file"></a>

### File-based authentication

When you set up RBAC only with mTLS authentication, you need to use the
file-based authentication provider for Control Center (Legacy) and Confluent CLI.

File-based provider is not supported when MDS uses LDAP or OAuth authentication.

#### Server-side file-based authentication for MDS

1. Create a file with user credentials in the following format:
   ```text
   <username-1>:<password-1>
   <username-2>:<password-2>
   ```
2. Create a Kubernetes secret that contains file-based username/password you
   created in the previous step. They secret should have key `userstore.txt`.
   For example:
   ```bash
   kubectl create secret generic file-secret \
     --from-file=userstore.txt=./fileUserPassword.txt \
     --namespace confluent
   ```
3. Configure the MDS as below in the Kafka CR to use the file-based authentication
   mechanism:
   ```yaml
   kind: Kafka
   spec:
     services:
       mds:
         provider:
           file:        --- [1]
             secretRef: --- [2]
   ```

   * [1] Required.
   * [2] Create a Kubernetes secret that contains the credential file.

If you need to use a vault to set up file-based authentication for MDS, use the
[Configuration Override](co-configure-misc.md#co-config-overrides) feature. For a sample
configuration, see the [CFK example repo scenario](https://github.com/confluentinc/confluent-kubernetes-examples/tree/master/security/mds-mtls-with-vault).

## Create authentication credentials for Kafka, KRaft, and MDS

#### NOTE
Authentication secret files must use the Linux style line ending that only
uses line feed (`\n`). The Windows style line ending that uses carriage
return and line feed (`\r\n`) does not work in Kubernetes secret files.

### SASL/PLAIN credentials for Kafka and KRaft

<a id="co-sasl-plain-server-jaas"></a>

#### Create server-side SASL/PLAIN credentials using JAAS config

When you use `jaasConfig` to provide required credentials for Kafka and
KRaft, CFK automates configuration. For example, when you add, remove, or
update  users, CFK automatically updates the JAAS config. This is the
recommended  way to configure SASL/PLAIN for Kafka.

The expected key for `jaasConfig` is `plain-users.json`.

1. Create a `.json` file and add the expected value, in the following
   format. This file is used for client authentication.
   ```json
   {
   "username1": "password1",
   "username2": "password2",
   ...
   "usernameN": "passwordN"
   }
   ```
2. Create a `.txt` file and add the expected value, in the following
   format. This file is used for inter-broker authentication.
   ```text
   username=<username>
   password=<password>
   ```
3. Create a Kubernetes secret using the expected keys (`plain-users.json` and
   `plain-interbroker.txt`) and the value files you created in the previous
   steps.

   The following example command creates a Kubernetes secret, using the
   `./creds-kafka-sasl-users.json` and `./creds-kafka-sasl-interbroker.txt`
   files that contain the credentials:
   ```bash
   kubectl create secret generic credential \
     --from-file=plain-users.json=./creds-kafka-sasl-users.json \
     --from-file=plain-interbroker.txt=./creds-kafka-sasl-interbroker.txt \
     --namespace confluent
   ```

<a id="co-sasl-plain-server-jaas-passthrough"></a>

#### Create server-side SASL/PLAIN credentials using JAAS config pass-through

If you have customizations, such as using a custom login handler, you can
bypass the CFK automation and provide the configuration directly using
`jaasConfigPassThrough`.

The expected key for `jaasConfigPassThrough` is `plain-jaas.conf`.

The expected value for the key (the data in the file) is your JAAS config text.
See [this Confluent Platform doc](https://docs.confluent.io/platform/current/security/authentication/sasl/plain/overview.html#jaas)
for understanding JAAS configs.

1. Create a `.conf` file and add the expected value, in the following
   format.
   ```bash
   sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
     username="<admin username>" \
     password="<admin user password>" \
     user_admin="<admin user password>" \
     user_<additional user1>="<additional user1 password>" \
     ...
     user_<additional userN>=”<additional userN password>”;
   ```

   * The `username` and `password` properties are used by the broker to
     initiate connections to other brokers.
   * The `user_<username_N>` properties define the passwords for the users,
     `<username_N>`, that connect to the broker. The broker validates all
     client connections, including those from other brokers using these
     properties.

   The following example uses the standard login module and specifies two
   additional users, `user1` and `user2`.
   ```bash
   sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
     username="admin" \
     password="admin-secret" \
     user_admin="admin-secret" \
     user_user1="user1-secret" \
     user_user2=”user2-secret”;
   ```
2. You can use a Kubernetes secret or a directory path in the container to store
   the credentials.
   * Create a Kubernetes secret using the expected key (`plain-jaas.conf`)
     and the value file you created in the previous step.

     The following example command creates a Kubernetes secret, using the
     `./creds-kafka-sasl-users.conf` file that contains the credentials:
     ```bash
     kubectl create secret generic credential \
       --from-file=plain-jaas.conf=./creds-kafka-sasl-users.conf \
       --namespace confluent
     ```
   * Use a directory path in the container to provide the required credentials.

     If `jaasConfigPassThrough.directoryPathInContainer` is configured as
     `/vaults/secrets` in the Kafka CR, the expected file, `plain-jaas.conf`,
     must exist in the directory path.

     See  [Provide secrets for Confluent Platform component CR](co-credentials.md#co-vault-category-1) for providing the credential and required
     annotations when using Vault.

     See [CFK GitHub examples](https://github.com/confluentinc/confluent-kubernetes-examples/tree/master/security/configure-with-vault)
     for more information on using the `directoryPathInContainer` property
     with Vault.

<a id="co-sasl-plain-client-jaas"></a>

#### Create client-side SASL/PLAIN credentials using JAAS config

When you use `jaasConfig`, you provide the user names and passwords, and  CFK
automates configuration. For example, when you add, remove, or update  users,
CFK automatically updates JAAS config.

The expected client-side key for `jaasConfig` is `plain.txt`.

1. Create a `.txt` file and add the expected value, in the following
   format:
   ```text
   username=<username>
   password=<password>
   ```

   You specify the name of this value file in the next step when you create a
   secret.
2. Create a Kubernetes secret using the expected key (`plain.txt`) and the
   value file you created in the previous step.

   The following example command creates a Kubernetes secret, using the
   `./creds-kafka-sasl-users.txt` file that contains the credentials:
   ```bash
   kubectl create secret generic credential \
     --from-file=plain.txt=./creds-kafka-sasl-users.txt \
     --namespace confluent
   ```

   If the user name or password changes in the future, you need to update the
   credentials in the value file (e.g. `./creds-kafka-sasl-users.txt` file),
   update the secret for the `plain.txt` key, and manually restart the Confluent Platform
   components that depends on the `plain.txt` key. For details, see
   [Update client-side SASL/PLAIN users using JAAS config](co-manage-authentication.md#co-update-client-sasl-plain-users).

<a id="co-sasl-plain-client-jaas-passthrough"></a>

#### Create client-side SASL/PLAIN credentials using JAAS config pass-through

If you have customizations, such as using a custom login handler, you can bypass
the CFK automation and provide the configuration directly using
`jaasConfigPassThrough`.

The expected client-side key for `jaasConfigPassThrough` is
`plain-jaas.conf`.

1. Create a `.conf` file and add the expected value, in the following
   format.

   For example:
   ```text
   sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
     username="kafka" \
     password="kafka-secret";
   ```

   You specify the name of this value file in the next step when you create a
   secret.
2. You can use a Kubernetes secret or a directory path in the container to
   store the credentials.
   * Create a Kubernetes secret using the expected key (`plain-jaas.conf`)
     and the value file you created in the previous step.

     The following example command creates a Kubernetes secret, using the
     `./creds-kafka-sasl-users.conf` file that contains the credentials:
     ```bash
     kubectl create secret generic credential \
       --from-file=plain-jaas.conf=./creds-kafka-sasl-users.conf \
       --namespace confluent
     ```
   * Use a directory path in the container to provide the required credentials.

     If `jaasConfigPassThrough.directoryPathInContainer` is configured as
     `/vaults/secrets` in the component CR, the expected file,
     `plain-jaas.conf`, must exist in that directory path.

     See  [Provide secrets for Confluent Platform component CR](co-credentials.md#co-vault-category-1) for providing the credential and required
     annotations when using Vault.

### SASL/PLAIN LDAP credentials for Kafka and KRaft

<a id="co-sasl-ldap-server-jaas"></a>

#### Create server-side SASL/PLAIN LDAP credentials using JAAS config

The expected server-side key for `jaasConfig` is `plain-interbroker.txt`.

1. Create a `.txt` file and add the expected value, in the following
   format. You specify the name of this value file in the next step when you
   create a secret.
   ```text
   username=<user>
   password=<password>
   ```

   The username and password must belong to a user that exists in LDAP. This
   is the user that each Kafka broker authenticates when the cluster starts.
2. Create a Kubernetes Secret with the user name and password for inter-broker
   authentication.

   The following example command creates a Kubernetes secret, using the
   `./creds-kafka-ldap-users.txt` file that contains the credentials:
   ```bash
   kubectl create secret generic credential \
     --from-file=plain-interbroker.txt=./creds-kafka-ldap-users.txt \
     --namespace confluent
   ```

<a id="co-sasl-ldap-server-jaas-passthrough"></a>

#### Create server-side SASL/PLAIN LDAP credentials using JAAS config pass-through

The expected server-side key for `jaasConfigPassThrough` is
`plain-jaas.conf`.

1. Create a `.conf` file and add the expected value. For example:
   ```text
   sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
      username="kafka" \
      password="kafka-secret";
   ```

   You specify the name of this value file in the next step when you create a
   secret.
2. You can use a Kubernetes secret or a directory path in the container to
   store the credentials.
   * Create a Kubernetes secret using the expected key (`plain-jaas.conf`)
     and the value file you created in the previous step.

     The following example command creates a Kubernetes secret, using the
     `./creds-kafka-sasl-users.conf` file that contains the credentials:
     ```bash
     kubectl create secret generic credential \
       --from-file=plain-jaas.conf=./creds-kafka-sasl-users.conf \
       --namespace confluent
     ```
   * Use a directory path in the container to provide the required credentials.

     If `jaasConfigPassThrough.directoryPathInContainer` is configured as
     `/vaults/secrets` in the Kafka CR, the expected file, `plain-jaas.conf`,
     must exist in the directory path.

     See  [Provide secrets for Confluent Platform component CR](co-credentials.md#co-vault-category-1) for providing the credential and required
     annotations when using Vault.

### Bearer credentials for MDS

<a id="co-bearer-server-creds"></a>

#### Create client-side Bearer credentials for MDS

Provide the required Bearer credentials.

The expected key is `bearer.txt`.

1. Create a `.txt` file with the expected value in the following format:
   ```text
   username=<username>
   password=<password>
   ```
2. You can use a Kubernetes secret or a directory path in the container to
   store the bearer credentials.
   * Create a Kubernetes secret using the expected key (`bearer.txt`) and the
     value.

     For example, using the `./c3-mds-client.txt` file that you
     create for Control Center (Legacy) credentials:
     ```bash
     kubectl create secret generic c3-mds-client \
       --from-file=bearer.txt=./c3-mds-client.txt \
       --namespace confluent
     ```
   * Use a directory path in the container to provide the required credentials.

     See  [Provide secrets for Confluent Platform component CR](co-credentials.md#co-vault-category-1) for providing the credential and required
     annotations when using Vault.

     See  [CFK GitHub examples](https://github.com/confluentinc/confluent-kubernetes-examples/tree/master/security/configure-with-vault)
     for more information on using the `directoryPathInContainer` property
     with Vault.