Configuring Token Authentication

The OAuth 2 Authorization Framework “enables a third-party application to obtain limited access to an HTTP service, either on behalf of a resource owner by orchestrating an approval interaction between the resource owner and the HTTP service, or by allowing the third-party application to obtain access on its own behalf.” The SASL OAUTHBEARER mechanism enables the use of the framework in a SASL (non-HTTP) context; it is defined in RFC 7628. The default OAUTHBEARER implementation in Apache Kafka® creates and validates Unsecured JSON Web Tokens and is only suitable for use in non-production Kafka installations. In addition to Kafka clients, neither the Confluent Metrics Reporter nor interbroker protocols should use OAUTHBEARER. For more information, see Security Considerations for SASL/OAUTHBEARER.

MDS extends upon this with the introduction of cryptographically signed tokens, which allow services to communicate proof of authentication for a user without forwarding along user secrets. After this feature is generally available for use in production systems, cryptographically signed tokens provide the most suitable solution. For details about MDS configuration, see Configure Metadata Service (MDS).

Configuring Kafka Brokers

1. Configure SASL port and SASL mechanisms in server.properties as described here.

   For example:

   # Add named listener RBAC to existing listeners and advertised.listeners
   listeners=RBAC://:9092,existing_listener://:9093
   advertised.listeners=RBAC://localhost:9092,existing_listener://localhost:9093
   
   # Add protocol mapping for newly-added named listener RBAC
   listener.security.protocol.map=existinglistener:existingsecurity,RBAC:SASL_PLAINTEXT (or SASL_SSL if production)
   listener.name.rbac.sasl.enabled.mechanisms=OAUTHBEARER
   
   # Configure the public key used to verify RBAC Metadata Service signatures
   # username, password and metadataServerUrls must be set if used for inter-broker communication
   listener.name.rbac.oauthbearer.sasl.jaas.config= \
     org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
     publicKeyPath=“<path-to-mds-public-key>“;
   
   # Set SASL callback handler to verify authentication token signatures
   listener.name.rbac.oauthbearer.sasl.server.callback.handler.class=io.confluent.kafka.server.plugins.auth.token.TokenBearerValidatorCallbackHandler
   
   # Set interbroker protocol to SASL_PLAINTEXT
   sasl.mechanism.inter.broker.protocol=PLAIN
   security.inter.broker.protocol=SASL_PLAINTEXT
   

2. Configure OAuthBearerLoginModule callback handlers for verifying client tokens.

   sasl.server.callback.handler.class=io.confluent.kafka.server.plugins.auth.token.TokenBearerValidatorCallbackHandler
   sasl.login.callback.handler.class=io.confluent.kafka.server.plugins.auth.token.TokenBearerServerLoginCallbackHandler
   

For details on the token listener configuration options shown above, refer to Configure the token listener.

Configuring Confluent Schema Registry

Configure the login service with bearer token authentication in your schema-registry.properties. For more information, see Configuring Role-Based Access Control for Schema Registry.

For example:

// Replace all text between ``<>`` with appropriate values.
schema.registry.resource.extension.class=io.confluent.kafka.schemaregistry.security.SchemaRegistrySecurityResourceExtension
confluent.schema.registry.authorizer.class=io.confluent.kafka.schemaregistry.security.authorizer.rbac.RbacAuthorizer

rest.servlet.initializor.classes=io.confluent.common.security.jetty.initializer.InstallBearerOrBasicSecurityHandler

confluent.schema.registry.auth.mechanism=JETTY_AUTH
confluent.metadata.bootstrap.server.urls=<http/s>://<host>:<port>
confluent.metadata.http.auth.credentials.provider=BASIC
confluent.metadata.basic.auth.user.info=<sr-user-name>:<sr-password>

confluent.metadata.public.key.path=<path-to-mds-public-key>

Configuring Confluent REST Proxy

Configure the login service with bearer token authentication in your kafka-rest.properties. For more information, see Role-Based Access Control (RBAC).

For example:

// Replace all text between ``<>`` with appropriate values.
kafka.rest.resource.extension.class=io.confluent.kafkarest.security.KafkaRestSecurityResourceExtension

rest.servlet.initializor.classes=io.confluent.common.security.jetty.initializer.InstallBearerOrBasicSecurityHandler

confluent.metadata.bootstrap.server.urls=<http/s>://<host>:<port>
confluent.metadata.http.auth.credentials.provider=BASIC
confluent.metadata.basic.auth.user.info=<test-user-name>:<rest-password>

confluent.metadata.public.key.path=<path-to-mds-public-key>

client.security.protocol=SASL_PLAINTEXT

Configuring Kafka Connect

Configure the login service with bearer token authentication in your connect-distributed.properties. For more information, see Kafka Connect and RBAC.

For example:

// Replace all text between ``<>`` with appropriate values.
rest.extension.classes=io.confluent.connect.security.ConnectSecurityExtension

rest.servlet.initializor.classes=io.confluent.common.security.jetty.initializer.InstallBearerOrBasicSecurityHandler

confluent.metadata.bootstrap.server.urls=<http/s>://<host>:<port>
confluent.metadata.http.auth.credentials.provider=BASIC
confluent.metadata.basic.auth.user.info=<connect-user-name>:<connect-password>

public.key.path=<path-to-mds-public-key>

Configuring ksqlDB

Configure the login service with bearer token authentication in your ksql-server.properties. For more information, see ksqlDB and role-based access control (RBAC).

For example:

// Replace all text between ``<>`` with appropriate values.
ksql.security.extension.class=io.confluent.ksql.security.KsqlConfluentSecurityExtension

ksql.authentication.plugin.class=io.confluent.ksql.security.VertxBearerOrBasicAuthenticationPlugin

confluent.metadata.bootstrap.server.urls=<http/s>://<host>:<port>
confluent.metadata.http.auth.credentials.provider=BASIC
confluent.metadata.basic.auth.user.info=<ksql-user-name>:<ksql-password>

public.key.path=/path/to/public-pem-key

Configuring Confluent Control Center

Configure the login service with bearer token authentication in your control-center.properties. For more information, see Configure RBAC for Control Center.

For example:

// Replace all text between ``<>`` with appropriate values.
confluent.controlcenter.rest.authentication.method=BEARER
confluent.controlcenter.streams.security.protocol=SASL_PLAINTEXT (or SASL_SSL if production)
confluent.controlcenter.auth.bearer.public.key.path=/etc/confluent-control-center/public.pem
confluent.controlcenter.metadata.username=<c3 user>
confluent.controlcenter.metadata.password=<c3 password>
confluent.controlcenter.metadata.urls=<http/s>://<host>:<port>

Configuring Kafka Clients

Important

If you are configuring Kafka clients for the Schema Registry or REST Proxy security plugins, you must prefix each parameter with confluent.license. For example, security.protocol becomes confluent.license.security.protocol.

To configure SASL authentication on the clients:

1. Configure the JAAS configuration property for each client in producer.properties or consumer.properties. The login module describes how the clients like producer and consumer can connect to the Kafka Broker. The following is an example configuration for a client for the OAUTHBEARER mechanisms:

   // Replace all text between ``<>`` with appropriate values
   sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
     username="<username>" \
     password="<password>" \
     metadataServerUrls="<http/s>://<host>:<port>";
   

   Important

   Do not use token services or the OAUTHBEARER SASL mechanism (listener.name.rbac.sasl.enabled.mechanisms=OAUTHBEARER) for external client communications. With RBAC enabled, token services are intended for internal communication between Confluent Platform components only (for example, it is valid for a Schema Registry licensed client), and not for long-running service principals or client authentication. The OAUTHBEARER setting is for internal use and subject to change, and does not implement a full-featured OAuth protocol. Therefore, use one of the supported authentication methods like SASL or mTLS (mutual TLS) for long-lived or client use cases. For details, refer to Authentication Methods Overview.

2. Configure clients with the following properties (either in client properties files, code, or flags; for example, --producer-property flag on kafka-console-producer):

   security.protocol=SASL_SSL (or SASL_PLAINTEXT if non-production)
   sasl.mechanism=OAUTHBEARER
   sasl.login.callback.handler.class=io.confluent.kafka.clients.plugins.auth.token.TokenUserLoginCallbackHandler
   

3. The default implementation of SASL/OAUTHBEARER depends on the jackson-databind library. Because it’s an optional dependency, you must configure it as a dependency in your build tool.

Token Refresh for SASL/OAUTHBEARER

Kafka periodically refreshes any token before it expires so that the client can continue to make connections to brokers. The parameters that impact how the refresh algorithm operates are specified as part of the producer/consumer/broker configuration and are as follows. The default values should not need to be explicitly set.

Producer/Consumer/Broker Configuration Property	Default
sasl.login.refresh.window.factor	.8
sasl.login.refresh.window.jitter	.05
sasl.login.refresh.min.period.seconds	60
sasl.login.refresh.min.buffer.seconds	300

Security Considerations for SASL/OAUTHBEARER

• OAUTHBEARER should be used in production environments only with TLS-encryption to prevent interception of tokens.
• For more details on OAuth 2 security considerations in general, refer to RFC 6749, Section 10.