...

New JAAS config option for default, unsecured bearer tokens - `unsecuredLoginExtension_<extensionname>` (as shown in the "Example" paragraph). The name "auth" is not supported as a custom extension name with any SASL/OAUTHBEARER mechanism, including the unsecured one, since it is reserved by the specification for what is normally sent in the HTTP Authorization header. An attempt to use it will result in an exception. There are also additional regex validations for extension name and values to ensure they conform to the OAuth standard. The server can further validate the extensions via its pluggable callback handler.

`OAuthBearerExtensionsValidatorCallback` - callback for OAuth extension validation, providing access to the token

package org.apache.kafka.common.security.oauthbearer;

/**
 * A {@code Callback} for use by the {@code SaslServer} implementation when it
 * needs to validate the SASL extensions for the OAUTHBEARER mechanism
 * Callback handlers should use the
 * {@link #error(String, String)} method to communicate validation errors back to
 * the SASL Server.
 * Callback handlers should communicate other problems by raising an {@code IOException}.
 * <p>
 * The OAuth bearer token is provided in the callback for better context.
 * It is very important that token validation is done in its own {@link OAuthBearerValidatorCallback}
 * irregardless of provided extensions, as they are inherently insecure.
 */
public class OAuthBearerExtensionsValidatorCallback {

    public OAuthBearerExtensionsValidatorCallback(OAuthBearerToken token, SaslExtensions extensions)

    /**
     * @return (potentially null) {@link SaslExtensions} consisting of the extension names and values that were sent by the client
     */
    public SaslExtensions extensions()

    /**
     * @return (potentially null) {@link OAuthBearerToken} the OAuth bearer token of the client
     */
    public OAuthBearerToken token()

    /**
     * @return (potentially null) name of extension which caused validation failure
     */
    public String invalidExtensionName()

    /**
     * @return (potentially null) message further describing reason of validation failure
     */
    public String errorMessage()

    /**
     * Set the error values if extension validation has failed
     *
     * @param invalidExtensionName
     *            the mandatory extension name which caused the validation failure
     * @param errorMessage
     *            optional error message describing why the validation failed
     */
    public void error(String invalidExtensionName, String errorMessage)
}

`SaslExtensionsCallback` - generic callback to hold extensions

...

package org.apache.kafka.common.security.auth;

/**
 * A simple value object class holding customizable SASL extensions
 */
public class SaslExtensions {
    public SaslExtensions(Map<String, String> extensionMap)

    /**
     * Returns an <strong>immutable</strong> map of the extension names and their values
     */
     public Map<String, String> map() {
         return extensionsMap;
     }
}

The default `OAuthBearerLoginModule` and the `OAuthBearerSaslClient` will be changed to request the extensions from their callback handler. For backwards compatibility it is not necessary for the callback handler to support `SaslExtensionsCallback`. Any UnsupportedCallbackException that is thrown will be ignored and no extensions will be added.

...

Describe the new thing you want to do in appropriate detail. This may be fairly extensive and have large subsections of its own. Or it may be a few sentences. Use judgement based on the scope of the change.

Create a new public `SaslExtensions` class that takes most of the generalizable logic from `ScramExtensions`. `ScramExtensions` will extend `SaslExtensions`
Create a new public `SaslExtensionsCallback` class which will be similar to `ScramExtensionsCallback`. `ScramExtensionsCallback` will NOT extend `SaslExtensionsCallback` since it will not support the new `SaslExtensions` class.
Create a new public `OAuthBearerExtensionsValidatorCallback` class.

Client Path

1. Pass `SaslExtensionsCallback` to the callback handler of `OAuthBearerLoginModule`

...

1. . The handler should parses the extensions from the JAAS config (unsecuredLoginExtension_xxx) and populate them in the Subject class.
   1. The default `OAuthBearerUnsecuredLoginCallbackHandler` will be updated with this behavior.
2. Pass `SaslExtensionsCallback` to the callback handler of `OAuthBearerSaslClient`

...

1. . The handler should take the extensions from the Subject and populate them in the callback

...

1. 
   
   1. The default `OAuthBearerSaslClientCallbackHandler` will be updated to handle the callback.
2. `OAuthBearerSaslClient` will then attach the populated extensions (if any) to the first client message

...

...

Server Path - `OAuthBearerServer`

1. Parse sent extensions (if any) from the first client message 
   1. The OAuthBearerServer

...

1. 1. will use a strict regex

...

1. 1. which parses only letters for keys and only ASCII characters for values. This ensures the message conforms to the standard
2. Validate them by passing `OAuthBearerExtensionsValidatorCallback` to its callback handler
   1. If the configured server callback handler does not support `OAuthBearerExtensionsValidatorCallback`, no extensions will be exposed (as per RFC 7628 - "Unknown key/value pairs MUST be ignored by the server")
      
3. Expose them via its `OAuthBearerServer#getNegotiatedProperty()` method. 
   
   1. This will allow custom principals to access

...

1. 1. extensions through the `SaslServer` instance in `SaslAuthenticationContext#server()`

Example

A user would make use of the changes in this KIP in the following way:

1. Add the extension names to your JAAS configuration in the client
   
   

   Code Block
   KafkaClient { org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule Required unsecuredLoginStringClaim_sub="thePrincipalName" unsecuredLoginExtension_traceId="123" unsecuredLoginExtension_logLevel="WARN"; };

   
   

2. A custom principal builder can then make use of the new extension

   Code Block

   public class CustomPrincipalBuilder implements KafkaPrincipalBuilder { @Override public KafkaPrincipal build(AuthenticationContext context) { if (context instanceof SaslAuthenticationContext) { SaslServer saslServer = ((SaslAuthenticationContext) context).server(); String traceId = saslServer.getNegotiatedPropery("traceId"); return new CustomPrincipal("", saslServer, traceId); } throw new Exception(); } }

   
   

Compatibility, Deprecation, and Migration Plan

• What impact (if any) will there be on existing users? 
  None. This is simply allowing allows for more configuration. We will ignore if legacy callback handlers raise `UnsupportedCallbackException` on the new callback classes.
  Other mechanisms like SCRAM should remain unaffected

...