...

Once the login has occurred for this client, the returned access token can be reused by other connections from this client.

Validator Callback Handler (Broker)

The validation callback handler is invoked as part of the SASL Kafka server authentication process when a JWT is received. This callback handler will parse and validate the JWT using its contents along with provided configuration.

Validation Logic

While these additional connections will not issue the token retrieval HTTP call on the client, the broker will still validate the token repeatedly each time it is received.

Per KIP-368, the OAuth token re-authentication logic from the existing implementation is automatically "inherited" by this implementation, so no additional work is needed to support that feature.

Validator Callback Handler (Broker)

The validation callback handler is invoked as part of the SASL Kafka server authentication process when a JWT is received. This callback handler will parse and validate the JWT using its contents along with provided configuration.

Validation Logic

The basic overview of the JWT validation that will The basic overview of the JWT validation that will be performed is:

1. Parse the JWT into separate header, payload, and signature sections

2. Base-64 decode the header and payload

3. Extract the necessary claims for validation

4. Match the key ID (kid) specified in the JWT header to a JWK ID from the JWK Set

5. Ensure the encoding algorithm (alg from the header) isn't none and matches the expected algorithm for the JWK ID

6. Encode the header and payload sections of the original encoded JWT access token using the public key from the JWK and ensure it matches the signature section of the JWT

7. Extract the scope, iat, exp, and sub claims as these are needed by the OAuthBearerToken object to be passed to the OAuthBearerValidatorCallback.

8. Optional claim validation that ensures that issuer, audience, or other claims match a given value

...

1. Providing a JWKS URL. In this mode, the JWKS data will be retrieved from the OAuth provider via the configured URL on broker startup. All then-current keys will be cached on the broker (per the ‘max age’; the jose4j library has a means to keep these-up-to-date when they age out) for incoming requests. If an authentication request is received for a JWT that includes a kid that isn’t yet in the cache, the JWKS endpoint will be queried again on demand. However, we prefer polling via a background thread to hopefully pre-populate the cache with any forthcoming keys before any JWT requests that include them are received.

2. Providing a JWKS file. On startup, the broker will load the JWKS file from a configured location and will watch the file for updates, allowing for dynamic configuration updates. The means by which the JWKS file is updated is left to the cluster administrator. In the event that an unknown JWT key is encountered, this implementation will simply issue an error and validation will fail.

Broker-to-broker Support

1. configuration updates. The means by which the JWKS file is updated is left to the cluster administrator. In the event that an unknown JWT key is encountered, this implementation will simply issue an error and validation will fail.

Broker-to-broker Support

The use of OAuth credentials for broker-to-broker communication will continue to be supported. As with the existing implementation, users can specify the protocols and implementations to use for broker-based communication.

Testing

In addition to unit and integration tests, there will be a standalone tool in the tools directory/module: org.apache.kafka.tools.OAuthCompatibilityTest. This test can be run via the bin/kafka-run-class.sh script thusly:

KAFKA_OPTS=""
export KAFKA_OPTS="$KAFKA_OPTS -DclientId=$client_id"
export KAFKA_OPTS="$KAFKA_OPTS -DclientSecret=$client_secret"
export KAFKA_OPTS="$KAFKA_OPTS -Dscope=api://$client_id/.default"
export KAFKA_OPTS="$KAFKA_OPTS -DtokenEndpointUri=$token_endpoint_uri"
export KAFKA_OPTS="$KAFKA_OPTS -DjwksEndpointUri=$jwks_uri"
export KAFKA_OPTS="$KAFKA_OPTS -DclockSkew=30"
export KAFKA_OPTS="$KAFKA_OPTS -DexpectedAudience=$expected_audience"

./bin/kafka-run-class.sh org.apache.kafka.tools.OAuthCompatibilityTest

As seen in the example invocation above, the various Kafka configuration options for the client and server properties are passed in via the existing KAFKA_OPTS environment variable. The test will connect to remote systems as needed to authenticate, retrieve tokens, retrieve JWKS, and perform validation (nothing is "mocked"). 

This tool serves three basic purposes:

1. Allows validation that a given OAuth/OIDC provider is compliant with our implementation of the specification
2. Provides users a means to reproduce and diagnose issues for support needs
3. Provides a means to perform system tests against live providers for tracking regressions

The output of the test is easily consumable:

PASSED 1/5: client configuration
PASSED 2/5: client JWT retrieval
PASSED 3/5: client JWT validation
PASSED 4/5: broker configuration
PASSED 5/5: broker JWT validation

Additionally debugging can be selectively enabled using the standard tools-log4j.properties file if errors are detected by the tool. All configuration options supported by the client and server are also supported by the tool; it attempts to be as faithful as possible to the runtime logic to minimize discrepancies between the tool and the client and serverThe use of OAuth credentials for broker-to-broker communication will continue to be supported. As with the existing implementation, users can specify the protocols and implementations to use for broker-based communication.

Compatibility, Deprecation, and Migration Plan

...