Configuring OAUTHBEARER

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 (i.e. a 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.

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.

Configuring Kafka Brokers

Note

While use of separate JAAS files is supported, it is not the recommended approach. Instead, use the listener configuration specified in Recommended Kafka Broker Configuration to replace steps 1 and 2 below. Note that step 3 is still required.

  1. Add a suitably modified JAAS file similar to the one below to each Kafka broker’s configuration directory. In this example it’s named kafka_server_jaas.conf:

    KafkaServer {
    org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required
    unsecuredLoginStringClaim_sub="admin";
};

    The property unsecuredLoginStringClaim_sub in the KafkaServer section is used by the broker when it initiates connections to other brokers. In this example, admin will appear in the subject (sub) claim and will be the user for inter-broker communication.

  2. Pass the JAAS config file location as JVM parameter to each Kafka broker:

    -Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf

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

    For example:

    listeners=SASL_SSL://host.name:port (or SASL_PLAINTEXT if non-production)
security.inter.broker.protocol=SASL_SSL (or SASL_PLAINTEXT if non-production)
sasl.mechanism.inter.broker.protocol=OAUTHBEARER
sasl.enabled.mechanisms=OAUTHBEARER

OAuthBearerLoginModule Login Options When Using a Custom Callback Handler

All client connections within the same Java process share the same login, unless they are using different configurations. When using the OAuthBearerLoginModule with a custom callback handler (sasl.login.callback.handler.class), login options are not included. This results in all clients sharing the same authentication token, which can be problematic when when you have multiple Kafka clients created with different logins (for example, when using Kafka connectors) within the same JVM.

To avoid having clients share the same authentication token in such circumstances, configure different logins using sasl.jaas.config (do not use the JVM system property java.security.auth.login.config). In other words, different clients that require separate logins must have different values specified in sasl.jaas.config. This ensures that the login callback handler returns the correct token based on the different configurations.

For example, the following login configuration explicitly overrides the SASL configuration in each connector by specifying different login options:

"producer.override.sasl.jaas.config":"org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required kid=\"ABCDefgh11Pi7jKLMNOpq-R0s3TuVWxyzAbcDEFgHI0\"  login.id=\"kafka_producer_prem\" ;"

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:

    sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
     unsecuredLoginStringClaim_sub="alice";

    The option unsecuredLoginStringClaim_sub is used by clients to configure the subject (sub) claim, which determines the user for client connections. In this example, clients connect to the broker as user alice. Different clients within a JVM may connect as different users by specifying different subject (sub) claims in sasl.jaas.config.

    You can also specify the JAAS configuration for clients as a JVM parameter similar to brokers as described here. Clients use the login section named KafkaClient. This option allows only one user for all client connections from a JVM.

  2. Configure the following properties in producer.properties or consumer.properties:

    security.protocol=SASL_SSL (or SASL_PLAINTEXT if non-production)
sasl.mechanism=OAUTHBEARER

  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 via your build tool.

Unsecured Client Side Token Creation Options for SASL/OAUTHBEARER

  • The default implementation of SASL/OAUTHBEARER in Kafka creates and validates Unsecured JSON Web Tokens. While suitable only for non-production use, it does provide the flexibility to create arbitrary tokens in a DEV or TEST environment.

  • Here are the various supported JAAS module options on the client side (and on the broker side if OAUTHBEARER is the inter-broker protocol):

    Client Side JAAS Module Options Description
    unsecuredLoginStringClaim_<claimname>="value" Creates a String claim with the given name and value. Any valid claim name can be specified except iat and exp (these are automatically generated).
    unsecuredLoginNumberClaim_<claimname>="value" Creates a Number claim with the given name and value. Any valid claim name can be specified except iat and exp (these are automatically generated).
    unsecuredLoginListClaim_<claimname>="value" Creates a String List claim with the given name and values parsed from the given value where the first character is taken as the delimiter. For example: unsecuredLoginListClaim_fubar="value1 value2". Any valid claim name can be specified except iat and exp (these are automatically generated).
    unsecuredLoginPrincipalClaimName Set to a custom claim name if you wish the name of the String claim holding the principal name to be something other than ‘sub’.
    unsecuredLoginLifetimeSeconds Set to an integer value if the token expiration is to be set to something other than the default value of 3600 seconds (which is 1 hour). The exp claim will be set to reflect the expiration time.
    unsecuredLoginScopeClaimName Set to a custom claim name if you wish the name of the String or String List claim holding any token scope to be something other than ‘scope’.

Unsecured Broker Side Token Validation Options for SASL/OAUTHBEARER

  • Here are the various supported JAAS module options on the broker side for Unsecured JSON Web Token validation:

    Broker Side JAAS Module Options Description
    unsecuredValidatorPrincipalClaimName="value" Set to a non-empty value if you wish a particular String claim holding a principal name to be checked for existence; the default is to check for the existence of the ‘sub’ claim.
    unsecuredValidatorScopeClaimName="value" Set to a custom claim name if you wish the name of the String or String List claim holding any token scope to be something other than ‘scope’.
    unsecuredValidatorRequiredScope="value" Set to a space-delimited list of scope values if you wish the String/String List claim holding the token scope to be checked to make sure it contains certain values.
    unsecuredValidatorAllowableClockSkewMs="value" Set to a positive integer value if you wish to allow up to some number of positive milliseconds of clock skew (the default is 0).

  • You can override the default unsecured SASL/OAUTHBEARER implementation using custom login and SASL Server callback handlers.

  • For more details on security considerations, see RFC 6749, Section 10.

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. See the documentation for these properties elsewhere for details. The default values are usually reasonable, in which case these configuration parameters would not need to be explicitly set.

Producer/Consumer/Broker Configuration Property
sasl.login.refresh.window.factor
sasl.login.refresh.window.jitter
sasl.login.refresh.min.period.seconds
sasl.login.refresh.min.buffer.seconds

Production Use of SASL/OAUTHBEARER

To use in production, you must write an implementation of a login callback handler to create and obtain the OAUTHBEARER token and a server callback handler to securely validate the OAUTHBEARER token:

Login Callback Handler for Token Retrieval

You must provide an implementation of org.apache.kafka.common.security.auth.AuthenticateCallbackHandler that handles an instance of org.apache.kafka.common.security.oauthbearer.OAuthBearerTokenCallback. You can declare it using either the sasl.login.callback.handler.class configuration option for a non-broker client, or using the prefixed listener.name.sasl_ssl.oauthbearer.sasl.login.callback.handler.class configuration option for brokers (when SASL/OAUTHBEARER is the inter-broker protocol).

Server Callback Handler for Token Validation

You must provide an implementation of org.apache.kafka.common.security.auth.AuthenticateCallbackHandler that handles an instance of org.apache.kafka.common.security.oauthbearer.OAuthBearerValidatorCallback. You can declare it using the prefixed listener.name.sasl_ssl.oauthbearer.sasl.server.callback.handler.class broker configuration option.

Security Considerations for SASL/OAUTHBEARER

  • The default implementation of SASL/OAUTHBEARER in Kafka creates and validates Unsecured JSON Web Tokens. This is suitable only for non-production use.
  • OAUTHBEARER should be used in production enviromnments only with TLS-encryption to prevent interception of tokens.
  • The default unsecured SASL/OAUTHBEARER implementation may be overridden (and must be overridden in production environments) using custom login and SASL Server callback handlers as described above.
  • For more details on OAuth 2 security considerations in general, refer to RFC 6749, Section 10.