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

   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.

   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
       clientAudience: AUDIENCE 11
       clientScope: SCOPE 12

   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 to true to enable PLAIN authentication on the listener, which is supported by all clients on all platforms. The Kafka client must enable the PLAIN mechanism and set the username and password. PLAIN can be used to authenticate either by using the OAuth access token, or the OAuth clientId and secret (the client credentials). The behavior is additionally controlled by whether tokenEndpointUri is specified or not. Default is false. If tokenEndpointUri is specified and the client sets password to start with the string $accessToken:, the server interprets the password as the access token and the username as the account username. Otherwise, the username is interpreted as the clientId and the password as the client secret, which the broker uses to obtain the access token in the client’s name. If tokenEndpointUri is not specified, the password is always interpreted as an access token and the username is always interpreted as the account username, which must match the principal id extracted from the token. This is known as 'no-client-credentials' mode because the client must always obtain the access token by itself, and can’t use clientId and secret.

   9

   Additional configuration for PLAIN mechanism to allow clients to authenticate by passing clientId and secret as username and password as described in the previous point. If not specified the clients can authenticate over PLAIN only by passing an access token as password parameter.

   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.

   11

   (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.

   12

   (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.

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>{oauth-version}</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
       audience: kafka 5

   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.

   5

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

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.

Procedure

1. Install the Red Hat Single Sign-On server using the Red Hat Single Sign-On Operator as described in Server Installation and Configuration in the Red Hat Single Sign-On documentation.
2. Wait until the Red Hat Single Sign-On instance is running.

3. Get the external hostname to be able to access the Admin Console.

   NS=sso
   oc get ingress keycloak -n $NS

   In this example, we assume the Red Hat Single Sign-On server is running in the sso namespace.

4. Get the password for the admin user.

   oc get -n $NS pod keycloak-0 -o yaml | less

   The password is stored as a secret, so get the configuration YAML file for the Red Hat Single Sign-On instance to identify the name of the secret (secretKeyRef.name).

5. Use the name of the secret to obtain the clear text password.

   SECRET_NAME=credential-keycloak
   oc get -n $NS secret $SECRET_NAME -o yaml | grep PASSWORD | awk '{print $2}' | base64 -D

   In this example, we assume the name of the secret is credential-keycloak.

6. Log in to the Admin Console with the username admin and the password you obtained.

   Use https://HOSTNAME to access the OpenShift ingress.

   You can now upload the example realm to Red Hat Single Sign-On using the Admin Console.

7. Click Add Realm to import the example realm.

8. Add the examples/security/keycloak-authorization/kafka-authz-realm.json file, and then click Create.

   You now have kafka-authz as your current realm in the Admin Console.

   The default view displays the Master realm.

9. In the Red Hat Single Sign-On Admin Console, go to Clients > kafka > Authorization > Settings and check that Decision Strategy is set to Affirmative.

   An affirmative policy means that at least one policy must be satisfied for a client to access the Kafka cluster.

10. In the Red Hat Single Sign-On Admin Console, go to Groups, Users, Roles and Clients to view the realm configuration.

    Groups

    Groups are used to create user groups and set user permissions. Groups are sets of users with a name assigned. They are used to compartmentalize users into geographical, organizational or departmental units. Groups can be linked to an LDAP identity provider. You can make a user a member of a group through a custom LDAP server admin user interface, for example, to grant permissions on Kafka resources.

    Users

    Users are used to create users. For this example, alice and bob are defined. alice is a member of the ClusterManager group and bob is a member of ClusterManager-my-cluster group. Users can be stored in an LDAP identity provider.

    Roles

    Roles mark users or clients as having certain permissions. Roles are a concept analogous to groups. They are usually used to tag users with organizational roles and have the requisite permissions. Roles cannot be stored in an LDAP identity provider. If LDAP is a requirement, you can use groups instead, and add Red Hat Single Sign-On roles to the groups so that when users are assigned a group they also get a corresponding role.

    Clients

    Clients can have specific configurations. For this example, kafka, kafka-cli, team-a-client, and team-b-client clients are configured.

    • The kafka client is used by Kafka brokers to perform the necessary OAuth 2.0 communication for access token validation. This client also contains the authorization services resource definitions, policies, and authorization scopes used to perform authorization on the Kafka brokers. The authorization configuration is defined in the kafka client from the Authorization tab, which becomes visible when Authorization Enabled is switched on from the Settings tab.
    • The kafka-cli client is a public client that is used by the Kafka command line tools when authenticating with username and password to obtain an access token or a refresh token.
    • The team-a-client and team-b-client clients are confidential clients representing services with partial access to certain Kafka topics.

11. In the Red Hat Single Sign-On Admin Console, go to Authorization > Permissions to see the granted permissions that use the resources and policies defined for the realm.

    For example, the kafka client has the following permissions:

    Dev Team A can write to topics that start with x_ on any cluster
    Dev Team B can read from topics that start with x_ on any cluster
    Dev Team B can update consumer group offsets that start with x_ on any cluster
    ClusterManager of my-cluster Group has full access to cluster config on my-cluster
    ClusterManager of my-cluster Group has full access to consumer groups on my-cluster
    ClusterManager of my-cluster Group has full access to topics on my-cluster

    Dev Team A

    The Dev Team A realm role can write to topics that start with x_ on any cluster. This combines a resource called Topic:x_*, Describe and Write scopes, and the Dev Team A policy. The Dev Team A policy matches all users that have a realm role called Dev Team A.

    Dev Team B

    The Dev Team B realm role can read from topics that start with x_ on any cluster. This combines Topic:x_*, Group:x_* resources, Describe and Read scopes, and the Dev Team B policy. The Dev Team B policy matches all users that have a realm role called Dev Team B. Matching users and clients have the ability to read from topics, and update the consumed offsets for topics and consumer groups that have names starting with x_.

Procedure

1. Use the hostname of the Red Hat Single Sign-On instance you deployed to prepare a truststore certificate for Kafka brokers to communicate with the Red Hat Single Sign-On server.

   SSO_HOST=SSO-HOSTNAME
   SSO_HOST_PORT=$SSO_HOST:443
   STOREPASS=storepass
   
   echo "Q" | openssl s_client -showcerts -connect $SSO_HOST_PORT 2>/dev/null | awk ' /BEGIN CERTIFICATE/,/END CERTIFICATE/ { print $0 } ' > /tmp/sso.crt

   The certificate is required as OpenShift ingress is used to make a secure (HTTPS) connection.

2. Deploy the certificate to OpenShift as a secret.

   oc create secret generic oauth-server-cert --from-file=/tmp/sso.crt -n $NS

3. Set the hostname as an environment variable

   SSO_HOST=SSO-HOSTNAME

4. Create and deploy the example Kafka cluster.

   cat examples/security/keycloak-authorization/kafka-ephemeral-oauth-single-keycloak-authz.yaml | sed -E 's#\${SSO_HOST}'"#$SSO_HOST#" | oc create -n $NS -f -

Procedure

1. Run a new interactive pod container using the AMQ Streams Kafka image to connect to a running Kafka broker.

   NS=sso
   oc run -ti --restart=Never --image=registry.redhat.io/amq7/amq-streams-kafka-28-rhel8:1.8.4 kafka-cli -n $NS -- /bin/sh

   Note

   If oc times out waiting on the image download, subsequent attempts may result in an AlreadyExists error.

2. Attach to the pod container.

   oc attach -ti kafka-cli -n $NS

3. Use the hostname of the Red Hat Single Sign-On instance to prepare a certificate for client connection using TLS.

   SSO_HOST=SSO-HOSTNAME
   SSO_HOST_PORT=$SSO_HOST:443
   STOREPASS=storepass
   
   echo "Q" | openssl s_client -showcerts -connect $SSO_HOST_PORT 2>/dev/null | awk ' /BEGIN CERTIFICATE/,/END CERTIFICATE/ { print $0 } ' > /tmp/sso.crt

4. Create a truststore for TLS connection to the Kafka brokers.

   keytool -keystore /tmp/truststore.p12 -storetype pkcs12 -alias sso -storepass $STOREPASS -import -file /tmp/sso.crt -noprompt

5. Use the Kafka bootstrap address as the hostname of the Kafka broker and the tls listener port (9093) to prepare a certificate for the Kafka broker.

   KAFKA_HOST_PORT=my-cluster-kafka-bootstrap:9093
   STOREPASS=storepass
   
   echo "Q" | openssl s_client -showcerts -connect $KAFKA_HOST_PORT 2>/dev/null | awk ' /BEGIN CERTIFICATE/,/END CERTIFICATE/ { print $0 } ' > /tmp/my-cluster-kafka.crt

6. Add the certificate for the Kafka broker to the truststore.

   keytool -keystore /tmp/truststore.p12 -storetype pkcs12 -alias my-cluster-kafka -storepass $STOREPASS -import -file /tmp/my-cluster-kafka.crt -noprompt

   Keep the session open to check authorized access.

Setting up client and admin user configuration

1. Prepare a Kafka configuration file with authentication properties for the team-a-client client.

   SSO_HOST=SSO-HOSTNAME
   
   cat > /tmp/team-a-client.properties << EOF
   security.protocol=SASL_SSL
   ssl.truststore.location=/tmp/truststore.p12
   ssl.truststore.password=$STOREPASS
   ssl.truststore.type=PKCS12
   sasl.mechanism=OAUTHBEARER
   sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
     oauth.client.id="team-a-client" \
     oauth.client.secret="team-a-client-secret" \
     oauth.ssl.truststore.location="/tmp/truststore.p12" \
     oauth.ssl.truststore.password="$STOREPASS" \
     oauth.ssl.truststore.type="PKCS12" \
     oauth.token.endpoint.uri="https://$SSO_HOST/auth/realms/kafka-authz/protocol/openid-connect/token" ;
   sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler
   EOF

   The SASL OAUTHBEARER mechanism is used. This mechanism requires a client ID and client secret, which means the client first connects to the Red Hat Single Sign-On server to obtain an access token. The client then connects to the Kafka broker and uses the access token to authenticate.

2. Prepare a Kafka configuration file with authentication properties for the team-b-client client.

   cat > /tmp/team-b-client.properties << EOF
   security.protocol=SASL_SSL
   ssl.truststore.location=/tmp/truststore.p12
   ssl.truststore.password=$STOREPASS
   ssl.truststore.type=PKCS12
   sasl.mechanism=OAUTHBEARER
   sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
     oauth.client.id="team-b-client" \
     oauth.client.secret="team-b-client-secret" \
     oauth.ssl.truststore.location="/tmp/truststore.p12" \
     oauth.ssl.truststore.password="$STOREPASS" \
     oauth.ssl.truststore.type="PKCS12" \
     oauth.token.endpoint.uri="https://$SSO_HOST/auth/realms/kafka-authz/protocol/openid-connect/token" ;
   sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler
   EOF

3. Authenticate admin user alice by using curl and performing a password grant authentication to obtain a refresh token.

   USERNAME=alice
   PASSWORD=alice-password
   
   GRANT_RESPONSE=$(curl -X POST "https://$SSO_HOST/auth/realms/kafka-authz/protocol/openid-connect/token" -H 'Content-Type: application/x-www-form-urlencoded' -d "grant_type=password&username=$USERNAME&password=$PASSWORD&client_id=kafka-cli&scope=offline_access" -s -k)
   
   REFRESH_TOKEN=$(echo $GRANT_RESPONSE | awk -F "refresh_token\":\"" '{printf $2}' | awk -F "\"" '{printf $1}')

   The refresh token is an offline token that is long-lived and does not expire.

4. Prepare a Kafka configuration file with authentication properties for the admin user alice.

   cat > /tmp/alice.properties << EOF
   security.protocol=SASL_SSL
   ssl.truststore.location=/tmp/truststore.p12
   ssl.truststore.password=$STOREPASS
   ssl.truststore.type=PKCS12
   sasl.mechanism=OAUTHBEARER
   sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
     oauth.refresh.token="$REFRESH_TOKEN" \
     oauth.client.id="kafka-cli" \
     oauth.ssl.truststore.location="/tmp/truststore.p12" \
     oauth.ssl.truststore.password="$STOREPASS" \
     oauth.ssl.truststore.type="PKCS12" \
     oauth.token.endpoint.uri="https://$SSO_HOST/auth/realms/kafka-authz/protocol/openid-connect/token" ;
   sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler
   EOF

   The kafka-cli public client is used for the oauth.client.id in the sasl.jaas.config. Since it’s a public client it does not require a secret. The client authenticates with the refresh token that was authenticated in the previous step. The refresh token requests an access token behind the scenes, which is then sent to the Kafka broker for authentication.

1. Write to topic my-topic.

   bin/kafka-console-producer.sh --broker-list my-cluster-kafka-bootstrap:9093 --topic my-topic \
     --producer.config=/tmp/team-a-client.properties
   First message

   This request returns a Not authorized to access topics: [my-topic] error.

   team-a-client has a Dev Team A role that gives it permission to perform any supported actions on topics that start with a_, but can only write to topics that start with x_. The topic named my-topic matches neither of those rules.

2. Write to topic a_messages.

   bin/kafka-console-producer.sh --broker-list my-cluster-kafka-bootstrap:9093 --topic a_messages \
     --producer.config /tmp/team-a-client.properties
   First message
   Second message

   Messages are produced to Kafka successfully.

3. Press CTRL+C to exit the CLI application.
4. Check the Kafka container log for a debug log of Authorization GRANTED for the request.

1. Fetch messages from topic a_messages.

   bin/kafka-console-consumer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic a_messages \
     --from-beginning --consumer.config /tmp/team-a-client.properties

   The request returns an error because the Dev Team A role for team-a-client only has access to consumer groups that have names starting with a_.

2. Update the team-a-client properties to specify the custom consumer group it is permitted to use.

   bin/kafka-console-consumer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic a_messages \
     --from-beginning --consumer.config /tmp/team-a-client.properties --group a_consumer_group_1

   The consumer receives all the messages from the a_messages topic.

1. List topics.

   bin/kafka-topics.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --command-config /tmp/team-a-client.properties --list

   The a_messages topic is returned.

2. List consumer groups.

   bin/kafka-consumer-groups.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --command-config /tmp/team-a-client.properties --list

   The a_consumer_group_1 consumer group is returned.

   Fetch details on the cluster configuration.

   bin/kafka-configs.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --command-config /tmp/team-a-client.properties \
     --entity-type brokers --describe --entity-default

   The request returns an error because the operation requires cluster level permissions that team-a-client does not have.

1. Write to topic a_messages.

   bin/kafka-console-producer.sh --broker-list my-cluster-kafka-bootstrap:9093 --topic a_messages \
     --producer.config /tmp/team-b-client.properties
   Message 1

   This request returns a Not authorized to access topics: [a_messages] error.

2. Write to topic b_messages.

   bin/kafka-console-producer.sh --broker-list my-cluster-kafka-bootstrap:9093 --topic b_messages \
     --producer.config /tmp/team-b-client.properties
   Message 1
   Message 2
   Message 3

   Messages are produced to Kafka successfully.

3. Write to topic x_messages.

   bin/kafka-console-producer.sh --broker-list my-cluster-kafka-bootstrap:9093 --topic x_messages \
     --producer.config /tmp/team-b-client.properties
   Message 1

   A Not authorized to access topics: [x_messages] error is returned, The team-b-client can only read from topic x_messages.

4. Write to topic x_messages using team-a-client.

   bin/kafka-console-producer.sh --broker-list my-cluster-kafka-bootstrap:9093 --topic x_messages \
     --producer.config /tmp/team-a-client.properties
   Message 1

   This request returns a Not authorized to access topics: [x_messages] error. The team-a-client can write to the x_messages topic, but it does not have a permission to create a topic if it does not yet exist. Before team-a-client can write to the x_messages topic, an admin power user must create it with the correct configuration, such as the number of partitions and replicas.

1. Create the x_messages topic as alice.

   bin/kafka-topics.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --command-config /tmp/alice.properties \
     --topic x_messages --create --replication-factor 1 --partitions 1

   The topic is created successfully.

2. List all topics as alice.

   bin/kafka-topics.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --command-config /tmp/alice.properties --list
   bin/kafka-topics.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --command-config /tmp/team-a-client.properties --list
   bin/kafka-topics.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --command-config /tmp/team-b-client.properties --list

   Admin user alice can list all the topics, whereas team-a-client and team-b-client can only list the topics they have access to.

   The Dev Team A and Dev Team B roles both have Describe permission on topics that start with x_, but they cannot see the other team’s topics because they do not have Describe permissions on them.

3. Use the team-a-client to produce messages to the x_messages topic:

   bin/kafka-console-producer.sh --broker-list my-cluster-kafka-bootstrap:9093 --topic x_messages \
     --producer.config /tmp/team-a-client.properties
   Message 1
   Message 2
   Message 3

   As alice created the x_messages topic, messages are produced to Kafka successfully.

4. Use the team-b-client to produce messages to the x_messages topic.

   bin/kafka-console-producer.sh --broker-list my-cluster-kafka-bootstrap:9093 --topic x_messages \
     --producer.config /tmp/team-b-client.properties
   Message 4
   Message 5

   This request returns a Not authorized to access topics: [x_messages] error.

5. Use the team-b-client to consume messages from the x_messages topic:

   bin/kafka-console-consumer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic x_messages \
     --from-beginning --consumer.config /tmp/team-b-client.properties --group x_consumer_group_b

   The consumer receives all the messages from the x_messages topic.

6. Use the team-a-client to consume messages from the x_messages topic.

   bin/kafka-console-consumer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic x_messages \
     --from-beginning --consumer.config /tmp/team-a-client.properties --group x_consumer_group_a

   This request returns a Not authorized to access topics: [x_messages] error.

7. Use the team-a-client to consume messages from a consumer group that begins with a_.

   bin/kafka-console-consumer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic x_messages \
     --from-beginning --consumer.config /tmp/team-a-client.properties --group a_consumer_group_a

   This request returns a Not authorized to access topics: [x_messages] error.

   Dev Team A has no Read access on topics that start with a x_.

8. Use alice to produce messages to the x_messages topic.

   bin/kafka-console-consumer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic x_messages \
     --from-beginning --consumer.config /tmp/alice.properties

   Messages are produced to Kafka successfully.

   alice can read from or write to any topic.

9. Use alice to read the cluster configuration.

   bin/kafka-configs.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --command-config /tmp/alice.properties \
     --entity-type brokers --describe --entity-default

   The cluster configuration for this example is empty.