Chapter 6. Configuring an AMQ Streams on RHEL deployment

1. The Kafka client requests an access token from the authorization server using a client ID and secret, and optionally a refresh token. Alternatively, the client may authenticate using a username and a password.
2. The authorization server generates a new access token.
3. The Kafka client authenticates with the Kafka broker using the SASL OAUTHBEARER mechanism to pass the access token.
4. The Kafka broker validates the access token by calling a token introspection endpoint on the authorization server using its own client ID and secret.
5. A Kafka client session is established if the token is valid.

1. The Kafka client authenticates with the authorization server from the token endpoint, using a client ID and secret, and optionally a refresh token. Alternatively, the client may authenticate using a username and a password.
2. The authorization server generates a new access token.
3. The Kafka client authenticates with the Kafka broker using the SASL OAUTHBEARER mechanism to pass the access token.
4. The Kafka broker validates the access token locally using a JWT token signature check, and local token introspection.

1. The Kafka client authenticates with the Kafka broker using the SASL OAUTHBEARER mechanism to pass the long-lived access token.
2. The Kafka broker validates the access token by calling a token introspection endpoint on the authorization server, using its own client ID and secret.
3. A Kafka client session is established if the token is valid.

1. The Kafka client authenticates with the Kafka broker using the SASL OAUTHBEARER mechanism to pass the long-lived access token.
2. The Kafka broker validates the access token locally using a JWT token signature check and local token introspection.

1. The Kafka client passes a clientId as a username and a secret as a password.
2. The Kafka broker uses a token endpoint to pass the clientId and secret to the authorization server.
3. The authorization server returns a fresh access token or an error if the client credentials are not valid.

4. The Kafka broker validates the token in one of the following ways:

   1. If a token introspection endpoint is specified, the Kafka broker validates the access token by calling the endpoint on the authorization server. A session is established if the token validation is successful.
   2. If local token introspection is used, a request is not made to the authorization server. The Kafka broker validates the access token locally using a JWT token signature check.

1. The Kafka client passes a username and password. The password provides the value of an access token that was obtained manually and configured before running the client.

2. The password is passed with or without an $accessToken: string prefix depending on whether or not the Kafka broker listener is configured with a token endpoint for authentication.

   1. If the token endpoint is configured, the password should be prefixed by $accessToken: to let the broker know that the password parameter contains an access token rather than a client secret. The Kafka broker interprets the username as the account username.
   2. If the token endpoint is not configured on the Kafka broker listener (enforcing a no-client-credentials mode), the password should provide the access token without the prefix. The Kafka broker interprets the username as the account username. In this mode, the client doesn’t use a client ID and secret, and the password parameter is always interpreted as a raw access token.

3. The Kafka broker validates the token in one of the following ways:

   1. If a token introspection endpoint is specified, the Kafka broker validates the access token by calling the endpoint on the authorization server. A session is established if token validation is successful.
   2. If local token introspection is used, there is no request made to the authorization server. Kafka broker validates the access token locally using a JWT token signature check.

Procedure

1. Install Red Hat Single Sign-On.

   You can install from a ZIP file or by using an RPM.

2. Log in to the Red Hat Single Sign-On Admin Console to create the OAuth 2.0 policies for AMQ Streams.

   Login details are provided when you deploy Red Hat Single Sign-On.

3. Create and enable a realm.

   You can use an existing master realm.

4. Adjust the session and token timeouts for the realm, if required.
5. Create a client called kafka-broker.

6. From the Settings tab, set:

   • Access Type to Confidential
   • Standard Flow Enabled to OFF to disable web login for this client
   • Service Accounts Enabled to ON to allow this client to authenticate in its own name
7. Click Save before continuing.
8. From the Credentials tab, take a note of the secret for using in your AMQ Streams Kafka cluster configuration.

9. Repeat the client creation steps for any application client that will connect to your Kafka brokers.

   Create a definition for each new client.

   You will use the names as client IDs in your configuration.

Procedure

1. Configure the Kafka broker listener configuration in the server.properties file.

   For example, using the OAUTHBEARER mechanism:

   sasl.enabled.mechanisms=OAUTHBEARER
   listeners=CLIENT://0.0.0.0:9092
   listener.security.protocol.map=CLIENT:SASL_PLAINTEXT
   listener.name.client.sasl.enabled.mechanisms=OAUTHBEARER
   sasl.mechanism.inter.broker.protocol=OAUTHBEARER
   inter.broker.listener.name=CLIENT
   listener.name.client.oauthbearer.sasl.server.callback.handler.class=io.strimzi.kafka.oauth.server.JaasServerOauthValidatorCallbackHandler
   listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required ;
   listener.name.client.oauthbearer.sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler

   Copy to Clipboard Toggle word wrap

2. Configure broker connection settings as part of the listener.name.client.oauthbearer.sasl.jaas.config.

   The examples here show connection configuration options.

   Example 1: Local token validation using a JWKS endpoint configuration

   listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
     oauth.valid.issuer.uri="https://AUTH-SERVER-ADDRESS/auth/realms/REALM-NAME" \
     oauth.jwks.endpoint.uri="https://AUTH-SERVER-ADDRESS/auth/realms/REALM-NAME/protocol/openid-connect/certs" \
     oauth.jwks.refresh.seconds="300" \
     oauth.jwks.refresh.min.pause.seconds="1" \
     oauth.jwks.expiry.seconds="360" \
     oauth.username.claim="preferred_username" \
     oauth.ssl.truststore.location="PATH-TO-TRUSTSTORE-P12-FILE" \
     oauth.ssl.truststore.password="TRUSTSTORE-PASSWORD" \
     oauth.ssl.truststore.type="PKCS12" ;
   listener.name.client.oauthbearer.connections.max.reauth.ms=3600000

   Copy to Clipboard Toggle word wrap

   Example 2: Delegating token validation to the authorization server through the OAuth 2.0 introspection endpoint

   listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
     oauth.introspection.endpoint.uri="https://AUTH-SERVER-ADDRESS/auth/realms/REALM-NAME/protocol/openid-connect/introspection" \
     # ...

   Copy to Clipboard Toggle word wrap

3. If required, configure access to the authorization server.

   This step is normally required for a production environment, unless a technology like service mesh is used to configure secure channels outside containers.

   1. Provide a custom truststore for connecting to a secured authorization server. SSL is always required for access to the authorization server.

      Set properties to configure the truststore.

      For example:

      listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
        # ...
        oauth.client.id="kafka-broker" \
        oauth.client.secret="kafka-broker-secret" \
        oauth.ssl.truststore.location="PATH-TO-TRUSTSTORE-P12-FILE" \
        oauth.ssl.truststore.password="TRUSTSTORE-PASSWORD" \
        oauth.ssl.truststore.type="PKCS12" ;

      Copy to Clipboard Toggle word wrap

   2. If the certificate hostname does not match the access URL hostname, you can turn off certificate hostname validation:

      oauth.ssl.endpoint.identification.algorithm=""

      Copy to Clipboard Toggle word wrap

      The check ensures that client connection to the authorization server is authentic. You may wish to turn off the validation in a non-production environment.

4. Configure additional properties according to your chosen authentication flow:

   listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
     # ...
     oauth.token.endpoint.uri="https://AUTH-SERVER-ADDRESS/auth/realms/REALM-NAME/protocol/openid-connect/token" \ 

   1

   
     oauth.custom.claim.check="@.custom == 'custom-value'" \ 

   2

   
     oauth.scope="SCOPE" \ 

   3

   
     oauth.check.audience="true" \ 

   4

   
     oauth.audience="AUDIENCE" \ 

   5

   
     oauth.valid.issuer.uri="https://https://AUTH-SERVER-ADDRESS/auth/REALM-NAME" \ 

   6

   
     oauth.client.id="kafka-broker" \ 

   7

   
     oauth.client.secret="kafka-broker-secret" \ 

   8

   
     oauth.connect.timeout.seconds=60 \ 

   9

   
     oauth.read.timeout.seconds=60 \ 

   10

   
     oauth.http.retries=2 \ 

   11

   
     oauth.http.retry.pause.millis=300 \ 

   12

   
     oauth.groups.claim="$.groups" \ 

   13

   
     oauth.groups.claim.delimiter="," ; 

   14

   Copy to Clipboard Toggle word wrap

   1

   The OAuth 2.0 token endpoint URL to your authorization server. For production, always use https:// urls. Required when KeycloakRBACAuthorizer is used, or an OAuth 2.0 enabled listener is used for inter-broker communication.

   2

   (Optional) Custom claim checking. A JsonPath filter query that applies additional custom rules to the JWT access token during validation. If the access token does not contain the necessary data, it is rejected. When using the introspection endpoint method, the custom check is applied to the introspection endpoint response JSON.

   3

   (Optional) A scope parameter passed to the token endpoint. A scope is used when obtaining an access token for inter-broker authentication. It is also used in the name of a client for OAuth 2.0 over PLAIN client authentication using a clientId and secret. This only affects the ability to obtain the token, and the content of the token, depending on the authorization server. It does not affect token validation rules by the listener.

   4

   (Optional) Audience checking. If your authorization server provides an aud (audience) claim, and you want to enforce an audience check, set ouath.check.audience to true. Audience checks identify the intended recipients of tokens. As a result, the Kafka broker will reject tokens that do not have its clientId in their aud claims. Default is false.

   5

   (Optional) An audience parameter passed to the token endpoint. An audience is used when obtaining an access token for inter-broker authentication. It is also used in the name of a client for OAuth 2.0 over PLAIN client authentication using a clientId and secret. This only affects the ability to obtain the token, and the content of the token, depending on the authorization server. It does not affect token validation rules by the listener.

   6

   A valid issuer URI. Only access tokens issued by this issuer will be accepted. (Always required.)

   7

   The configured client ID of the Kafka broker, which is the same for all brokers. This is the client registered with the authorization server as kafka-broker. Required when an introspection endpoint is used for token validation, or when KeycloakRBACAuthorizer is used.

   8

   The configured secret for the Kafka broker, which is the same for all brokers. When the broker must authenticate to the authorization server, either a client secret, access token or a refresh token has to be specified.

   9

   (Optional) The connect timeout in seconds when connecting to the authorization server. The default value is 60.

   10

   (Optional) The read timeout in seconds when connecting to the authorization server. The default value is 60.

   11

   The maximum number of times to retry a failed HTTP request to the authorization server. The default value is 0, meaning that no retries are performed. To use this option effectively, consider reducing the timeout times for the oauth.connect.timeout.seconds and oauth.read.timeout.seconds options. However, note that retries may prevent the current worker thread from being available to other requests, and if too many requests stall, it could make the Kafka broker unresponsive.

   12

   The time to wait before attempting another retry of a failed HTTP request to the authorization server. By default, this time is set to zero, meaning that no pause is applied. This is because many issues that cause failed requests are per-request network glitches or proxy issues that can be resolved quickly. However, if your authorization server is under stress or experiencing high traffic, you may want to set this option to a value of 100 ms or more to reduce the load on the server and increase the likelihood of successful retries.

   13

   A JsonPath query used to extract groups information from JWT token or introspection endpoint response. Not set by default. This can be used by a custom authorizer to make authorization decisions based on user groups.

   14

   A delimiter used to parse groups information when returned as a single delimited string. The default value is ',' (comma).

5. Depending on how you apply OAuth 2.0 authentication, and the type of authorization server being used, add additional configuration settings:

   listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
     # ...
     oauth.check.issuer=false \ 

   1

   
     oauth.fallback.username.claim="CLIENT-ID" \ 

   2

   
     oauth.fallback.username.prefix="CLIENT-ACCOUNT" \ 

   3

   
     oauth.valid.token.type="bearer" \ 

   4

   
     oauth.userinfo.endpoint.uri="https://AUTH-SERVER-ADDRESS/auth/realms/REALM-NAME/protocol/openid-connect/userinfo" ; 

   5

   Copy to Clipboard Toggle word wrap

   1

   If your authorization server does not provide an iss claim, it is not possible to perform an issuer check. In this situation, set oauth.check.issuer to false and do not specify a oauth.valid.issuer.uri. Default is true.

   2

   An authorization server may not provide a single attribute to identify both regular users and clients. When a client authenticates in its own name, the server might provide a client ID. When a user authenticates using a username and password, to obtain a refresh token or an access token, the server might provide a username attribute in addition to a client ID. Use this fallback option to specify the username claim (attribute) to use if a primary user ID attribute is not available.

   3

   In situations where oauth.fallback.username.claim is applicable, it may also be necessary to prevent name collisions between the values of the username claim, and those of the fallback username claim. Consider a situation where a client called producer exists, but also a regular user called producer exists. In order to differentiate between the two, you can use this property to add a prefix to the user ID of the client.

   4

   (Only applicable when using oauth.introspection.endpoint.uri) Depending on the authorization server you are using, the introspection endpoint may or may not return the token type attribute, or it may contain different values. You can specify a valid token type value that the response from the introspection endpoint has to contain.

   5

   (Only applicable when using oauth.introspection.endpoint.uri) The authorization server may be configured or implemented in such a way to not provide any identifiable information in an introspection endpoint response. In order to obtain the user ID, you can configure the URI of the userinfo endpoint as a fallback. The oauth.fallback.username.claim, oauth.fallback.username.claim, and oauth.fallback.username.prefix settings are applied to the response of the userinfo endpoint.

Procedure

1. Add the client library with OAuth 2.0 support to the pom.xml file for the Kafka client:

   <dependency>
    <groupId>io.strimzi</groupId>
    <artifactId>kafka-oauth-client</artifactId>
    <version>0.12.0.redhat-00006</version>
   </dependency>

   Copy to Clipboard Toggle word wrap

2. Configure the client properties by specifying the following configuration in a properties file:

   • The security protocol
   • The SASL mechanism

   • The JAAS module and authentication properties according to the method being used

     For example, we can add the following to a client.properties file:

     Client credentials mechanism properties

     security.protocol=SASL_SSL 

     1

     
     sasl.mechanism=OAUTHBEARER 

     2

     
     ssl.truststore.location=/tmp/truststore.p12 

     3

     
     ssl.truststore.password=$STOREPASS
     ssl.truststore.type=PKCS12
     sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
       oauth.token.endpoint.uri="<token_endpoint_url>" \ 

     4

     
       oauth.client.id="<client_id>" \ 

     5

     
       oauth.client.secret="<client_secret>" \ 

     6

     
       oauth.ssl.truststore.location="/tmp/oauth-truststore.p12" \ 

     7

     
       oauth.ssl.truststore.password="$STOREPASS" \ 

     8

     
       oauth.ssl.truststore.type="PKCS12" \ 

     9

     
       oauth.scope="<scope>" \ 

     10

     
       oauth.audience="<audience>" ; 

     11

     Copy to Clipboard Toggle word wrap

     1

     SASL_SSL security protocol for TLS-encrypted connections. Use SASL_PLAINTEXT over unencrypted connections for local development only.

     2

     The SASL mechanism specified as OAUTHBEARER or PLAIN.

     3

     The truststore configuration for secure access to the Kafka cluster.

     4

     URI of the authorization server token endpoint.

     5

     Client ID, which is the name used when creating the client in the authorization server.

     6

     Client secret created when creating the client in the authorization server.

     7

     The location contains the public key certificate (truststore.p12) for the authorization server.

     8

     The password for accessing the truststore.

     9

     The truststore type.

     10

     (Optional) The scope for requesting the token from the token endpoint. An authorization server may require a client to specify the scope.

     11

     (Optional) The audience for requesting the token from the token endpoint. An authorization server may require a client to specify the audience.

     Password grants mechanism properties

     security.protocol=SASL_SSL
     sasl.mechanism=OAUTHBEARER
     ssl.truststore.location=/tmp/truststore.p12
     ssl.truststore.password=$STOREPASS
     ssl.truststore.type=PKCS12
     sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
       oauth.token.endpoint.uri="<token_endpoint_url>" \
       oauth.client.id="<client_id>" \ 

     1

     
       oauth.client.secret="<client_secret>" \ 

     2

     
       oauth.password.grant.username="<username>" \ 

     3

     
       oauth.password.grant.password="<password>" \ 

     4

     
       oauth.ssl.truststore.location="/tmp/oauth-truststore.p12" \
       oauth.ssl.truststore.password="$STOREPASS" \
       oauth.ssl.truststore.type="PKCS12" \
       oauth.scope="<scope>" \
       oauth.audience="<audience>" ;

     Copy to Clipboard Toggle word wrap

     1

     Client ID, which is the name used when creating the client in the authorization server.

     2

     (Optional) Client secret created when creating the client in the authorization server.

     3

     Username for password grant authentication. OAuth password grant configuration (username and password) uses the OAuth 2.0 password grant method. To use password grants, create a user account for a client on your authorization server with limited permissions. The account should act like a service account. Use in environments where user accounts are required for authentication, but consider using a refresh token first.

     4

     Password for password grant authentication.

     Note

     SASL PLAIN does not support passing a username and password (password grants) using the OAuth 2.0 password grant method.

     Access token properties

     security.protocol=SASL_SSL
     sasl.mechanism=OAUTHBEARER
     ssl.truststore.location=/tmp/truststore.p12
     ssl.truststore.password=$STOREPASS
     ssl.truststore.type=PKCS12
     sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
       oauth.token.endpoint.uri="<token_endpoint_url>" \
       oauth.access.token="<access_token>" ; 

     1

     
       oauth.ssl.truststore.location="/tmp/oauth-truststore.p12" \
       oauth.ssl.truststore.password="$STOREPASS" \
       oauth.ssl.truststore.type="PKCS12" \

     Copy to Clipboard Toggle word wrap

     1

     Long-lived access token for Kafka clients.

     Refresh token properties

     security.protocol=SASL_SSL
     sasl.mechanism=OAUTHBEARER
     ssl.truststore.location=/tmp/truststore.p12
     ssl.truststore.password=$STOREPASS
     ssl.truststore.type=PKCS12
     sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
       oauth.token.endpoint.uri="<token_endpoint_url>" \
       oauth.client.id="<client_id>" \ 

     1

     
       oauth.client.secret="<client_secret>" \ 

     2

     
       oauth.refresh.token="<refresh_token>" ; 

     3

     
       oauth.ssl.truststore.location="/tmp/oauth-truststore.p12" \
       oauth.ssl.truststore.password="$STOREPASS" \
       oauth.ssl.truststore.type="PKCS12" \

     Copy to Clipboard Toggle word wrap

     1

     Client ID, which is the name used when creating the client in the authorization server.

     2

     (Optional) Client secret created when creating the client in the authorization server.

     3

     Long-lived refresh token for Kafka clients.

3. Input the client properties for OAUTH 2.0 authentication into the Java client code.

   Example showing input of client properties

   Properties props = new Properties();
   try (FileReader reader = new FileReader("client.properties", StandardCharsets.UTF_8)) {
     props.load(reader);
   }

   Copy to Clipboard Toggle word wrap

4. Verify that the Kafka client can access the Kafka brokers.