// Module included in the following module:
//
// con-oauth-config.adoc

[id='proc-oauth-authentication-broker-config-{context}']
= Configuring OAuth 2.0 support for Kafka brokers

This procedure describes how to configure Kafka brokers so that the broker listeners are enabled to use OAuth 2.0 authentication using an authorization server.

We advise use of OAuth 2.0 over an encrypted interface through through a listener with `tls: true`.
Plain listeners are not recommended.

If the authorization server is using certificates signed by the trusted CA and matching the OAuth 2.0 server hostname, TLS connection works using the default settings.
Otherwise, you may need to configure the truststore with proper certificates or disable the certificate hostname validation.

When configuring the Kafka broker you have two options for the mechanism used to validate the access token during OAuth 2.0 authentication of the newly connected Kafka client:

* xref:example-1[Configuring fast local JWT token validation]
* xref:example-2[Configuring token validation using an introspection endpoint]

.Before you start

For more information on the configuration of OAuth 2.0 authentication for Kafka broker listeners, see:

* link:{BookURLConfiguring}#type-KafkaListenerAuthenticationOAuth-reference[`KafkaListenerAuthenticationOAuth` schema reference^]
* xref:con-oauth-authentication-flow-str[OAuth 2.0 authentication mechanisms]

.Prerequisites

* Strimzi and Kafka are running
* An OAuth 2.0 authorization server is deployed

.Procedure

. Update the Kafka broker configuration (`Kafka.spec.kafka`) of your `Kafka` resource in an editor.
+
[source,shell]
----
kubectl edit kafka my-cluster
----

. 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]]
.Example 1: Configuring fast local JWT token validation
[source,yaml,subs="+quotes,attributes"]
----
#...
- 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]]
.Example 2: Configuring token validation using an introspection endpoint
[source,yaml,subs="+quotes,attributes"]
----
- 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:
+
[source,yaml,subs="+quotes,attributes"]
----
  # ...
  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>
    connectTimeoutSeconds: 60 <13>
    readTimeoutSeconds: 60 <14>
    httpRetries: 2 <15>
    httpRetryPauseMs: 300 <16>
    groupsClaim: "$.groups" <17>
    groupsClaimDelimiter: "," <18>
----
<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 for clients on all platforms.
<9> Additional configuration for the PLAIN mechanism. If specified, clients can authenticate over PLAIN by passing an access token as the `password` using an `$accessToken:` prefix.
For production, always use `https://` urls.
<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> 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> 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.
<13> The connect timeout in seconds when connecting to the authorization server. The default value is 60.
<14> The read timeout in seconds when connecting to the authorization server. The default value is 60.
<15> 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 `connectTimeoutSeconds` and `readTimeoutSeconds` 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.
<16> 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.
<17> A JsonPath query that is used to extract groups information from either the JWT token or the introspection endpoint response. This option is not set by default. By configuring this option, a custom authorizer can make authorization decisions based on user groups.
<18> A delimiter used to parse groups information when it is returned as a single delimited string. The default value is ',' (comma).

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

. Check the update in the logs or by watching the pod state transitions:
+
[source,shell,subs="+quotes,attributes"]
----
kubectl logs -f ${POD_NAME} -c ${CONTAINER_NAME}
kubectl get pod -w
----
+
The rolling update configures the brokers to use OAuth 2.0 authentication.

.What to do next

* xref:proc-oauth-client-config-{context}[Configure your Kafka clients to use OAuth 2.0]
