:_mod-docs-content-type: CONCEPT

Configures custom authentication for listeners.

To configure custom authentication, set the `type` property to `custom`.
Custom authentication allows for any type of Kafka-supported authentication to be used.

.Example custom OAuth authentication configuration
[source,yaml,subs="attributes+"]
----
spec:
  kafka:
    config:
      principal.builder.class: SimplePrincipal.class
    listeners:
      - name: oauth-bespoke
        port: 9093
        type: internal
        tls: true
        authentication:
          type: custom
          sasl: true
          listenerConfig:
            oauthbearer.sasl.client.callback.handler.class: client.class
            oauthbearer.sasl.server.callback.handler.class: server.class
            oauthbearer.sasl.login.callback.handler.class: login.class
            oauthbearer.connections.max.reauth.ms: 999999999
            sasl.enabled.mechanisms: oauthbearer
            oauthbearer.sasl.jaas.config: |
              org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required ;
    template:
      pod:
        volumes:
          - name: example-secret
            secret:
              secretName: example
      kafkaContainer:
        volumeMounts:
          - name: example-secret
            mountPath: /mnt/secret-volume
----

A protocol map is generated that uses the `sasl` and `tls` values to determine which protocol to map to the listener.

* SASL = True, TLS = True -> SASL_SSL
* SASL = False, TLS = True -> SSL
* SASL = True, TLS = False -> SASL_PLAINTEXT
* SASL = False, TLS = False -> PLAINTEXT

Secrets are mounted to the `/mnt` directory in the Kafka broker nodes' containers.
For example, the mounted secret (`example`) in the example configuration would be located at `/mnt/secret-volume`.

= Configuring customized TLS Client Authentication

You can also use the `custom` authentication to configure customized TLS client authentication.
This allows configuration options that are not permissible with `type: tls` authentication.
For example, it's possible to configure a custom truststore with multiple trusted CAs or options such as `ssl.principal.mapping.rules`.

.Example custom TLS Client Authentication configuration
[source,yaml,subs="attributes+"]
----
spec:
  kafka:
    listeners:
      - name: tls
        port: 9093
        tls: true
        type: internal
        authentication:
          type: custom
          sasl: false
          listenerConfig:
            ssl.client.auth: required
            ssl.principal.mapping.rules: RULE:^CN=(.*?),(.*)$/$1@my-cluster.com/
            ssl.truststore.location: /mnt/my-truststore/ca.crt
            ssl.truststore.type: PEM
    template:
      pod:
        volumes:
          - name: my-truststore
            secret:
              secretName: custom-truststore
      kafkaContainer:
        volumeMounts:
          - name: my-truststore
            mountPath: /mnt/my-truststore
----

= Setting a custom principal builder

You can set a custom principal builder in the Kafka cluster configuration.
However, the principal builder is subject to the following requirements:

* The specified principal builder class must exist on the image.
_Before_ building your own, check if one already exists.
You'll need to rebuild the Strimzi images with the required classes.
* No other listener is using `oauth` type authentication.
This is because an OAuth listener appends its own principle builder to the Kafka configuration.
* The specified principal builder is compatible with Strimzi.

Custom principal builders must support peer certificates for authentication, as Strimzi uses these to manage the Kafka cluster.

ifdef::Section[]
A custom OAuth principal builder might be identical or very similar to the Strimzi https://github.com/strimzi/strimzi-kafka-oauth/blob/main/oauth-server/src/main/java/io/strimzi/kafka/oauth/server/OAuthKafkaPrincipalBuilder.java[OAuth principal builder].
endif::Section[]

NOTE: link:https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/common/security/authenticator/DefaultKafkaPrincipalBuilder.java#L73-L79[Kafka's default principal builder class] supports the building of principals based on the names of peer certificates.
The custom principal builder should provide a principal of type `user` using the name of the SSL peer certificate.

The following example shows a custom principal builder that satisfies the OAuth requirements of Strimzi.

.Example principal builder for custom OAuth configuration
[source,java,subs="attributes+"]
----
public final class CustomKafkaPrincipalBuilder implements KafkaPrincipalBuilder {

    public KafkaPrincipalBuilder() {}

    @Override
    public KafkaPrincipal build(AuthenticationContext context) {
        if (context instanceof SslAuthenticationContext) {
            SSLSession sslSession = ((SslAuthenticationContext) context).session();
            try {
                return new KafkaPrincipal(
                    KafkaPrincipal.USER_TYPE, sslSession.getPeerPrincipal().getName());
            } catch (SSLPeerUnverifiedException e) {
                throw new IllegalArgumentException("Cannot use an unverified peer for authentication", e);
            }
        }

        // Create your own KafkaPrincipal here
        ...
    }
}
----
