Chapter 12. KafkaListenerAuthenticationCustom schema reference Used in: GenericKafkaListener Full list of KafkaListenerAuthenticationCustom schema properties 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 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 ; secrets: - name: example 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 12.1. listenerConfig Listener configuration specified using listenerConfig is prefixed with listener.name. <listener_name>-<port> . For example, sasl.enabled.mechanisms becomes listener.name. <listener_name>-<port> .sasl.enabled.mechanisms . 12.2. secrets Secrets are mounted to /opt/kafka/custom-authn-secrets/custom-listener- <listener_name>-<port> / <secret_name> in the Kafka broker nodes' containers. For example, the mounted secret ( example ) in the example configuration would be located at /opt/kafka/custom-authn-secrets/custom-listener-oauth-bespoke-9093/example . 12.3. 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 AMQ Streams 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 AMQ Streams. Custom principal builders must support peer certificates for authentication, as AMQ Streams uses these to manage the Kafka cluster. Note 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 AMQ Streams. Example principal builder for custom OAuth configuration 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 ... } } 12.4. KafkaListenerAuthenticationCustom schema properties The type property is a discriminator that distinguishes use of the KafkaListenerAuthenticationCustom type from KafkaListenerAuthenticationTls , KafkaListenerAuthenticationScramSha512 , KafkaListenerAuthenticationOAuth . It must have the value custom for the type KafkaListenerAuthenticationCustom . Property Description listenerConfig Configuration to be used for a specific listener. All values are prefixed with listener.name. <listener_name> . map sasl Enable or disable SASL on this listener. boolean secrets Secrets to be mounted to /opt/kafka/custom-authn-secrets/custom-listener- <listener_name>-<port> / <secret_name> . GenericSecretSource array type Must be custom . string

Commands:
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 ; secrets: - name: example
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 } }