Chapter 4. Managing secure access to Kafka

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

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

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

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

Procedure

1. Deploy Red Hat Single Sign-On to your OpenShift cluster.

   Check the progress of the deployment in your OpenShift web console.

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. Update the Kafka broker configuration (Kafka.spec.kafka) of your Kafka resource in an editor.

   oc edit kafka my-cluster

2. Configure the Kafka broker listeners configuration.

   The configuration for each type of listener does not have to be the same, as they are independent.

   The examples here show the configuration options as configured for external listeners.

   Example 1: Configuring fast local JWT token validation

   #...
   - name: external
     port: 9094
     type: loadbalancer
     tls: true
     authentication:
       type: oauth 1
       validIssuerUri: <https://<auth-server-address>/auth/realms/external> 2
       jwksEndpointUri: <https://<auth-server-address>/auth/realms/external/protocol/openid-connect/certs> 3
       userNameClaim: preferred_username 4
       maxSecondsWithoutReauthentication: 3600 5
       tlsTrustedCertificates: 6
       - secretName: oauth-server-cert
         certificate: ca.crt
       disableTlsHostnameVerification: true 7
       jwksExpirySeconds: 360 8
       jwksRefreshSeconds: 300 9
       jwksMinRefreshPauseSeconds: 1 10
       enableECDSA: "true" 11

   1

   Listener type set to oauth.

   2

   URI of the token issuer used for authentication.

   3

   URI of the JWKS certificate endpoint used for local JWT validation.

   4

   The token claim (or key) that contains the actual user name in the token. The user name is the principal used to identify the user. The userNameClaim value will depend on the authentication flow and the authorization server used.

   5

   (Optional) Activates the Kafka re-authentication mechanism that enforces session expiry to the same length of time as the access token. If the specified value is less than the time left for the access token to expire, then the client will have to re-authenticate before the actual token expiry. By default, the session does not expire when the access token expires, and the client does not attempt re-authentication.

   6

   (Optional) Trusted certificates for TLS connection to the authorization server.

   7

   (Optional) Disable TLS hostname verification. Default is false.

   8

   The duration the JWKS certificates are considered valid before they expire. Default is 360 seconds. If you specify a longer time, consider the risk of allowing access to revoked certificates.

   9

   The period between refreshes of JWKS certificates. The interval must be at least 60 seconds shorter than the expiry interval. Default is 300 seconds.

   10

   The minimum pause in seconds between consecutive attempts to refresh JWKS public keys. When an unknown signing key is encountered, the JWKS keys refresh is scheduled outside the regular periodic schedule with at least the specified pause since the last refresh attempt. The refreshing of keys follows the rule of exponential backoff, retrying on unsuccessful refreshes with ever increasing pause, until it reaches jwksRefreshSeconds. The default value is 1.

   11

   (Optional) If ECDSA is used for signing JWT tokens on authorization server, then this needs to be enabled. It installs additional crypto providers using BouncyCastle crypto library. Default is false.

   Example 2: Configuring token validation using an introspection endpoint

   - name: external
     port: 9094
     type: loadbalancer
     tls: true
     authentication:
       type: oauth
       validIssuerUri: <https://<auth-server-address>/auth/realms/external>
       introspectionEndpointUri: <https://<auth-server-address>/auth/realms/external/protocol/openid-connect/token/introspect> 1
       clientId: kafka-broker 2
       clientSecret: 3
         secretName: my-cluster-oauth
         key: clientSecret
       userNameClaim: preferred_username 4
       maxSecondsWithoutReauthentication: 3600 5

   1

   URI of the token introspection endpoint.

   2

   Client ID to identify the client.

   3

   Client Secret and client ID is used for authentication.

   4

   The token claim (or key) that contains the actual user name in the token. The user name is the principal used to identify the user. The userNameClaim value will depend on the authorization server used.

   5

   (Optional) Activates the Kafka re-authentication mechanism that enforces session expiry to the same length of time as the access token. If the specified value is less than the time left for the access token to expire, then the client will have to re-authenticate before the actual token expiry. By default, the session does not expire when the access token expires, and the client does not attempt re-authentication.

   Depending on how you apply OAuth 2.0 authentication, and the type of authorization server, there are additional (optional) configuration settings you can use:

     # ...
     authentication:
       type: oauth
       # ...
       checkIssuer: false 1
       checkAudience: true 2
       fallbackUserNameClaim: client_id 3
       fallbackUserNamePrefix: client-account- 4
       validTokenType: bearer 5
       userInfoEndpointUri: https://OAUTH-SERVER-ADDRESS/auth/realms/external/protocol/openid-connect/userinfo 6
       enableOauthBearer: false 7
       enablePlain: true 8
       tokenEndpointUri: https://OAUTH-SERVER-ADDRESS/auth/realms/external/protocol/openid-connect/token 9
       customClaimCheck: "@.custom == 'custom-value'" 10

   1

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

   2

   If your authorization server provides an aud (audience) claim, and you want to enforce an audience check, set checkAudience 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 claim. Default is false.

   3

   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.

   4

   In situations where fallbackUserNameClaim 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.

   5

   (Only applicable when using introspectionEndpointUri) 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.

   6

   (Only applicable when using introspectionEndpointUri) 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 userNameClaim, fallbackUserNameClaim, and fallbackUserNamePrefix settings are applied to the response of userinfo endpoint.

   7

   Set this to false`to disable the OAUTHBEARER mechanism on the listener. At least one of PLAIN or OAUTHBEARER has to be enabled. Default is `true.

   8

   Set this to true to enable the PLAIN mechanism on the listener, which is supported by all clients on all platforms. The Kafka client has to enable the PLAIN mechanism and set the username and the password. This mechanism can be used to authenticate either by using the OAuth access token, or by using the OAuth client id and secret (client credentials). If the client sets password to start with the string $accessToken:, the password is interpreted as the access token on the server, and username as the account username, otherwise the user is interpreted as the client id, and password as the client secret. Default is false.

   9

   This has to be set to support the client credentials authentication when enablePlain is set to true, as described in previous point.

   10

   Additional custom rules can be imposed on the JWT access token during validation by setting this to a JsonPath filter query. If the access token does not contain the necessary data, it is rejected. When using the introspectionEndpointUri, the custom check is applied to the introspection endpoint response JSON.

3. Save and exit the editor, then wait for rolling updates to complete.

4. Check the update in the logs or by watching the pod state transitions:

   oc logs -f ${POD_NAME} -c ${CONTAINER_NAME}
   oc get pod -w

   The rolling update configures the brokers to use OAuth 2.0 authentication.

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.7.1.redhat-00003</version>
   </dependency>

2. Configure the system properties for the callback:

   For example:

   System.setProperty(ClientConfig.OAUTH_TOKEN_ENDPOINT_URI, “https://<auth-server-address>/auth/realms/master/protocol/openid-connect/token”); 1
   System.setProperty(ClientConfig.OAUTH_CLIENT_ID, "<client-name>"); 2
   System.setProperty(ClientConfig.OAUTH_CLIENT_SECRET, "<client-secret>"); 3

   1

   URI of the authorization server token endpoint.

   2

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

   3

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

3. Enable the SASL OAUTHBEARER mechanism on a TLS encrypted connection in the Kafka client configuration:

   For example:

   props.put("sasl.jaas.config", "org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required;");
   props.put("security.protocol", "SASL_SSL"); 1
   props.put("sasl.mechanism", "OAUTHBEARER");
   props.put("sasl.login.callback.handler.class", "io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler");

   1

   Here we use SASL_SSL for use over TLS connections. Use SASL_PLAINTEXT over unencrypted connections.

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

Procedure

1. Create a client secret and mount it to the component as an environment variable.

   For example, here we are creating a client Secret for the Kafka Bridge:

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Secret
   metadata:
    name: my-bridge-oauth
   type: Opaque
   data:
    clientSecret: MGQ1OTRmMzYtZTllZS00MDY2LWI5OGEtMTM5MzM2NjdlZjQw 1

   1

   The clientSecret key must be in base64 format.

2. Create or edit the resource for the Kafka component so that OAuth 2.0 authentication is configured for the authentication property.

   For OAuth 2.0 authentication, you can use:

   • Client ID and secret
   • Client ID and refresh token
   • Access token
   • TLS

   KafkaClientAuthenticationOAuth schema reference provides examples of each.

   For example, here OAuth 2.0 is assigned to the Kafka Bridge client using a client ID and secret, and TLS:

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaBridge
   metadata:
     name: my-bridge
   spec:
     # ...
     authentication:
       type: oauth 1
       tokenEndpointUri: https://<auth-server-address>/auth/realms/master/protocol/openid-connect/token 2
       clientId: kafka-bridge
       clientSecret:
         secretName: my-bridge-oauth
         key: clientSecret
       tlsTrustedCertificates: 3
       - secretName: oauth-server-cert
         certificate: tls.crt

   1

   Authentication type set to oauth.

   2

   URI of the token endpoint for authentication.

   3

   Trusted certificates for TLS connection to the authorization server.

   Depending on how you apply OAuth 2.0 authentication, and the type of authorization server, there are additional configuration options you can use:

   # ...
   spec:
     # ...
     authentication:
       # ...
       disableTlsHostnameVerification: true 1
       checkAccessTokenType: false 2
       accessTokenIsJwt: false 3
       scope: any 4

   1

   (Optional) Disable TLS hostname verification. Default is false.

   2

   If the authorization server does not return a typ (type) claim inside the JWT token, you can apply checkAccessTokenType: false to skip the token type check. Default is true.

   3

   If you are using opaque tokens, you can apply accessTokenIsJwt: false so that access tokens are not treated as JWT tokens.

   4

   (Optional) The scope for requesting the token from the token endpoint. An authorization server may require a client to specify the scope. In this case it is any.

3. Apply the changes to the deployment of your Kafka resource.

   oc apply -f your-file

4. Check the update in the logs or by watching the pod state transitions:

   oc logs -f ${POD_NAME} -c ${CONTAINER_NAME}
   oc get pod -w

   The rolling updates configure the component for interaction with Kafka brokers using OAuth 2.0 authentication.