// Module included in the following assemblies:
//
// assembly-oauth-authentication.adoc

[id='con-oauth-authentication-broker-{context}']
= OAuth 2.0 Kafka broker configuration

Kafka broker configuration for OAuth 2.0 involves:

* Creating the OAuth 2.0 client in the authorization server
* Configuring OAuth 2.0 authentication in the Kafka custom resource

NOTE: In relation to the authorization server, Kafka brokers and Kafka clients are both regarded as OAuth 2.0 clients.

== OAuth 2.0 client configuration on an authorization server

To configure a Kafka broker to validate the token received during session initiation,
the recommended approach is to create an OAuth 2.0 _client_ definition in an authorization server, configured as _confidential_, with the following client credentials enabled:

* Client ID of `kafka` (for example)
* Client ID and Secret as the authentication mechanism

NOTE: You only need to use a client ID and secret when using a non-public introspection endpoint of the authorization server.
The credentials are not typically required when using public authorization server endpoints, as with fast local JWT token validation.

== OAuth 2.0 authentication configuration in the Kafka cluster

To use OAuth 2.0 authentication in the Kafka cluster, you specify, for example, a `tls` listener configuration for your Kafka cluster custom resource with the authentication method `oauth`:

.Assigining the authentication method type for OAuth 2.0
[source,yaml,subs="+quotes, attributes"]
----
apiVersion: {KafkaApiVersion}
kind: Kafka
spec:
  kafka:
    # ...
    listeners:
      - name: tls
        port: 9093
        type: internal
        tls: true
        authentication:
          type: *oauth*
      #...
----

You can configure OAuth 2.0 authentication in your listeners.
We recommend using OAuth 2.0 authentication together with TLS encryption (`tls: true`).
Without encryption, the connection is vulnerable to network eavesdropping and unauthorized access through token theft.

You configure an `external` listener with `type: oauth` for a secure transport layer to communicate with the client.

.Using OAuth 2.0 with an external listener
[source,yaml,subs="+quotes"]
----
# ...
listeners:
  - name: external3
    port: 9094
    type: loadbalancer
    tls: true
    authentication:
      type: *oauth*
    #...
----

The `tls` property is _false_ by default, so it must be enabled.

When you have defined the type of authentication as OAuth 2.0, you add configuration based on the type of validation, either as xref:con-oauth-authentication-broker-fast-local[fast local JWT validation] or xref:con-oauth-authentication-broker-intro-local[token validation using an introspection endpoint].

The procedure to configure OAuth 2.0 for listeners, with descriptions and examples, is described in xref:proc-oauth-authentication-broker-config-{context}[Configuring OAuth 2.0 support for Kafka brokers].

[[con-oauth-authentication-broker-fast-local]]
== Fast local JWT token validation configuration

Fast local JWT token validation checks a JWT token signature locally.

The local check ensures that a token:

* Conforms to type by containing a (_typ_) claim value of `Bearer` for an access token
* Is valid (not expired)
* Has an issuer that matches a `validIssuerURI`

You specify a `validIssuerURI` attribute when you configure the listener, so that any tokens not issued by the authorization server are rejected.

The authorization server does not need to be contacted during fast local JWT token validation.
You activate fast local JWT token validation by specifying a `jwksEndpointUri` attribute, the endpoint exposed by the OAuth 2.0 authorization server.
The endpoint contains the public keys used to validate signed JWT tokens, which are sent as credentials by Kafka clients.

NOTE: All communication with the authorization server should be performed using TLS encryption.

You can configure a certificate truststore as a Kubernetes Secret in your Strimzi project namespace, and use a `tlsTrustedCertificates` attribute to point to the Kubernetes Secret containing the truststore file.

You might want to configure a `userNameClaim` to properly extract a username from the JWT token.
If required, you can use a JsonPath expression like `"['user.info'].['user.id']"` to retrieve the username from nested JSON attributes within a token.   

If you want to use Kafka ACL authorization, you need to identify the user by their username during authentication.
(The `sub` claim in JWT tokens is typically a unique ID, not a username.)

.Example configuration for fast local JWT token validation
[source,yaml,subs="+quotes, attributes"]
----
apiVersion: {KafkaApiVersion}
kind: Kafka
spec:
  kafka:
    #...
    listeners:
      - name: tls
        port: 9093
        type: internal
        tls: true
        authentication:
          type: *oauth*
          validIssuerUri: <https://<auth_server_address>/auth/realms/tls>
          jwksEndpointUri: <https://<auth_server_address>/auth/realms/tls/protocol/openid-connect/certs>
          userNameClaim: preferred_username
          maxSecondsWithoutReauthentication: 3600
          tlsTrustedCertificates:
          - secretName: oauth-server-cert
            certificate: ca.crt
    #...
----

[[con-oauth-authentication-broker-intro-local]]
== OAuth 2.0 introspection endpoint configuration

Token validation using an OAuth 2.0 introspection endpoint treats a received access token as opaque.
The Kafka broker sends an access token to the introspection endpoint, which responds with the token information necessary for validation.
Importantly, it returns up-to-date information if the specific access token is valid, and also information about when the token expires.

To configure OAuth 2.0 introspection-based validation, you specify an `introspectionEndpointUri` attribute rather than the `jwksEndpointUri` attribute specified for fast local JWT token validation.
Depending on the authorization server, you typically have to specify a `clientId` and `clientSecret`, because the introspection endpoint is usually protected.

.Example configuration for an introspection endpoint
[source,yaml,subs="+quotes, attributes"]
----
apiVersion: {KafkaApiVersion}
kind: Kafka
spec:
  kafka:
    listeners:
      - name: tls
        port: 9093
        type: internal
        tls: true
        authentication:
          type: *oauth*
          clientId: kafka-broker
          clientSecret:
            secretName: my-cluster-oauth
            key: clientSecret
          validIssuerUri: <https://<auth_server_-_address>/auth/realms/tls>
          introspectionEndpointUri: <https://<auth_server_address>/auth/realms/tls/protocol/openid-connect/token/introspect>
          userNameClaim: preferred_username
          maxSecondsWithoutReauthentication: 3600
          tlsTrustedCertificates:
          - secretName: oauth-server-cert
            certificate: ca.crt
----
