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
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>";
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 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 this 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:
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>";
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
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.