...

Kafka currently supports non-configurable SASL extensions in its SCRAM authentication protocol for delegation token validation. It would be useful to provide configurable SASL extensions for the OAuthBearer authentication mechanism as well, such that clients could attach arbitrary data for the principal authenticating into Kafka.

This way, a custom principal can hold information derived from the authentication mechanism, which could prove useful for better tracing and troubleshooting, for example. This  This can be done in a way which allows for easier extendability in future SASL mechanisms.

It is worth noting that these extensions would lack a digital signature and therefore should not be used for critical use-cases where security is a concern.

Public Interfaces

`SaslExtensions` class - encapsulates extensions and parses them to/from strings

...

Create a new `SaslExtensions` class that takes most of the generalizable logic from `ScramExtensions`. `ScramExtensions` will extend `SaslExtensions`
Create a new `SaslExtensionsCallback` which will be exactly the same as `ScramExtensionsCallback`. `ScramExtensionsCallback` cannot be deleted since it is a public class - it will extend `SaslExtensionsCallback` to preserve backwards-compatibility.
Pass `SaslExtensionsCallback` to the callback handler of `OAuthBearerSaslClient` so that the handler can populate the extensions in the callback. `OAuthBearerSaslClient` will then attach the extensions (if any) to the first client message.
Have `OAuthBearerServer` parse sent extensions and expose them via its `OAuthBearerServer#getNegotiatedProperty()` method. It will use a strict regex, parsing only letters for keys and only ASCII characters for values without commas (","), as they are used for separators in the OAuth message

This will allow custom principals to access them through the `SaslServer` instance in `SaslAuthenticationContext#server()`

Note that the default callback handler `OAuthBearerSaslClientCallbackHandler` will not attach any extensions - it is up to the custom user-defined callback handler to attach the appropriate extensions.

Example

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

1. Create a custom LoginCallbackHandler and configure the client to use it
   1. This handler should handle the `SaslExtensionsCallback` in his `handle()` method and attach custom extensions

      
      

      Code Block

      @Override public void handle(Callback[] callbacks) { for (Callback callback : callbacks) { if (callback instanceof OAuthBearerTokenCallback) handleTokenCallback((OAuthBearerTokenCallback) callback); else if (callback instanceof SaslExtensionsCallback) { Map<String, String> extensions = new HashMap<>(); extensions.put("trace-id", "123"); ((SaslExtensionsCallback) callback).extensions(extensions); } else throw new UnsupportedCallbackException(callback); } }

2. The extensions can be fetched in the server-side through `OAuthBearerSaslServer#getNegotiatedProperty("trace-id")`

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

   1. 
      

      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("trace-id"); return new CustomPrincipal("", saslServer, traceId); } throw new Exception(); } }

      
      

Compatibility, Deprecation, and Migration Plan

...