Package org.apache.kafka.common.security.oauthbearer

Class OAuthBearerLoginModule

java.lang.Object
org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule
All Implemented Interfaces:
LoginModule
public class OAuthBearerLoginModule
extends Object
implements LoginModule
The LoginModule for the SASL/OAUTHBEARER mechanism. When a client (whether a non-broker client or a broker when SASL/OAUTHBEARER is the inter-broker protocol) connects to Kafka the OAuthBearerLoginModule instance asks its configured AuthenticateCallbackHandler implementation to handle an instance of OAuthBearerTokenCallback and return an instance of OAuthBearerToken. A default, builtin AuthenticateCallbackHandler implementation creates an unsecured token as defined by these JAAS module options:

JAAS Module Option for Unsecured Token Retrieval Documentation
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'.

You can also add custom unsecured SASL extensions when using the default, builtin AuthenticateCallbackHandler implementation through using the configurable option unsecuredLoginExtension_<extensionname>. Note that there are validations for the key/values in order to conform to the SASL/OAUTHBEARER standard (https://tools.ietf.org/html/rfc7628#section-3.1), including the reserved key at OAuthBearerClientInitialResponse.AUTH_KEY. The OAuthBearerLoginModule instance also asks its configured AuthenticateCallbackHandler implementation to handle an instance of SaslExtensionsCallback and return an instance of SaslExtensions. The configured callback handler does not need to handle this callback, though -- any UnsupportedCallbackException that is thrown is ignored, and no SASL extensions will be associated with the login.

Production use cases will require writing an implementation of AuthenticateCallbackHandler that can handle an instance of OAuthBearerTokenCallback and declaring it via either the sasl.login.callback.handler.class configuration option for a non-broker client or via the listener.name.sasl_ssl.oauthbearer.sasl.login.callback.handler.class configuration option for brokers (when SASL/OAUTHBEARER is the inter-broker protocol).

This class stores the retrieved OAuthBearerToken in the Subject's private credentials where the SaslClient can retrieve it. An appropriate, builtin SaslClient implementation is automatically used and configured such that it can perform that retrieval.

Here is a typical, basic JAAS configuration for a client leveraging unsecured SASL/OAUTHBEARER authentication: 

 KafkaClient {
      org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule Required
      unsecuredLoginStringClaim_sub="thePrincipalName";
 };
An implementation of the Login interface specific to the OAUTHBEARER mechanism is automatically applied; it 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.

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

When a broker accepts a SASL/OAUTHBEARER connection the instance of the builtin SaslServer implementation asks its configured AuthenticateCallbackHandler implementation to handle an instance of OAuthBearerValidatorCallback constructed with the OAuth 2 Bearer Token's compact serialization and return an instance of OAuthBearerToken if the value validates. A default, builtin AuthenticateCallbackHandler implementation validates an unsecured token as defined by these JAAS module options:

JAAS Module Option for Unsecured Token Validation Documentation
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).

Here is a typical, basic JAAS configuration for a broker leveraging unsecured SASL/OAUTHBEARER validation: 

 KafkaServer {
      org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule Required
      unsecuredLoginStringClaim_sub="thePrincipalName";
 };
Production use cases will require writing an implementation of AuthenticateCallbackHandler that can handle an instance of OAuthBearerValidatorCallback and declaring it via the listener.name.sasl_ssl.oauthbearer.sasl.server.callback.handler.class broker configuration option.

The builtin SaslServer implementation for SASL/OAUTHBEARER in Kafka makes the instance of OAuthBearerToken available upon successful authentication via the negotiated property "OAUTHBEARER.token"; the token could be used in a custom authorizer (to authorize based on JWT claims rather than ACLs, for example).

This implementation's logout() method will logout the specific token that this instance logged in if it's Subject instance is shared across multiple LoginContexts and there happen to be multiple tokens on the Subject. This functionality is useful because it means a new token with a longer lifetime can be created before a soon-to-expire token is actually logged out. Otherwise, if multiple simultaneous tokens were not supported like this, the soon-to-be expired token would have to be logged out first, and then if the new token could not be retrieved (maybe the authorization server is temporarily unavailable, for example) the client would be left without a token and would be unable to create new connections. Better to mitigate this possibility by leaving the existing token (which still has some lifetime left) in place until a new replacement token is actually retrieved. This implementation supports this.

See Also:
SaslConfigs.SASL_LOGIN_REFRESH_WINDOW_FACTOR_DOC, SaslConfigs.SASL_LOGIN_REFRESH_WINDOW_JITTER_DOC, SaslConfigs.SASL_LOGIN_REFRESH_MIN_PERIOD_SECONDS_DOC, SaslConfigs.SASL_LOGIN_REFRESH_BUFFER_SECONDS_DOC