Important

You are viewing documentation for an older version of Confluent Platform. For the latest, click here.

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. 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:

   // Replace all text between ``<>`` with appropriate values.
   listeners=SASL_PLAINTEXT://<host.name:port> (or SASL_SSL if production)
   security.inter.broker.protocol=SASL_PLAINTEXT (or SASL_SSL if production)
   sasl.mechanism.inter.broker.protocol=OAUTHBEARER
   sasl.enabled.mechanisms=OAUTHBEARER
   sasl.jaas.config= \
     org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
     publicKeyPath="<path-to-mds-public-key>";
   

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
   

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 RBAC REST Proxy security configuration.

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

rest.servlet.initializor.classes=io.confluent.common.security.jetty.initializer.InstallBearerOrBasicSecurityHandler
websocket.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=<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

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>";
   

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.