// Module included in the following assemblies:
//
// assembly-securing-kafka-brokers.adoc

[id='con-securing-kafka-authentication-{context}']
= Configuring client authentication on listeners

[role="_abstract"]
Configure client authentication for Kafka brokers when creating listeners.
Specify the listener authentication type using the `Kafka.spec.kafka.listeners.authentication` property in the `Kafka` resource. 

For clients inside the Kubernetes cluster, you can create `plain` (without encryption) or `tls` _internal_ listeners.
The `internal` listener type use a headless service and the DNS names given to the broker pods. 
As an alternative to the headless service, you can also create a `cluster-ip` type of internal listener to expose Kafka using per-broker `ClusterIP` services.
For clients outside the Kubernetes cluster, you create _external_ listeners and specify a connection mechanism,
which can be `nodeport`, `loadbalancer`, `ingress` (Kubernetes only), or `route` (OpenShift only).

For more information on the configuration options for connecting an external client, see xref:deploy-client-access-{context}[].

Supported authentication options:

. mTLS authentication (only on the listeners with TLS enabled encryption)
. SCRAM-SHA-512 authentication
. xref:assembly-oauth-authentication_str[OAuth 2.0 token-based authentication]
. link:{BookURLConfiguring}#type-KafkaListenerAuthenticationCustom-reference[Custom authentication^]
. link:{BookURLConfiguring}#con-common-configuration-ssl-reference[TLS versions and cipher suites^]

If you're using OAuth 2.0 for client access management, user authentication and authorization credentials are handled through the authorization server.

The authentication option you choose depends on how you wish to authenticate client access to Kafka brokers.

NOTE: Try exploring the standard authentication options before using custom authentication. Custom authentication allows for any type of Kafka-supported authentication. It can provide more flexibility, but also adds complexity.

.Kafka listener authentication options
image::listener-config-options.png[options for listener authentication configuration]

The listener `authentication` property is used to specify an authentication mechanism specific to that listener.

If no `authentication` property is specified then the listener does not authenticate clients which connect through that listener.
The listener will accept all connections without authentication.

Authentication must be configured when using the User Operator to manage `KafkaUsers`.

The following example shows:

* A `plain` listener configured for SCRAM-SHA-512 authentication
* A `tls` listener with mTLS authentication
* An `external` listener with mTLS authentication

Each listener is configured with a unique name and port within a Kafka cluster.

IMPORTANT: When configuring listeners for client access to brokers, you can use port 9092 or higher (9093, 9094, and so on), but with a few exceptions.
The listeners cannot be configured to use the ports reserved for interbroker communication (9090 and 9091), Prometheus metrics (9404), and JMX (Java Management Extensions) monitoring (9999).

.Example listener authentication configuration
[source,yaml,subs="attributes+"]
----
apiVersion: {KafkaApiVersion}
kind: Kafka
metadata:
  name: my-cluster
  namespace: myproject
spec:
  kafka:
    # ...
    listeners:
      - name: plain
        port: 9092
        type: internal
        tls: true
        authentication:
          type: scram-sha-512
      - name: tls
        port: 9093
        type: internal
        tls: true
        authentication:
          type: tls
      - name: external3
        port: 9094
        type: loadbalancer
        tls: true
        authentication:
          type: tls
# ...
----

[id='con-mutual-tls-authentication-{context}']
== mTLS authentication

mTLS authentication is always used for the communication between Kafka nodes.

Strimzi can configure Kafka to use TLS (Transport Layer Security) to provide encrypted communication between Kafka brokers and clients either with or without mutual authentication.
For mutual, or two-way, authentication, both the server and the client present certificates.
When you configure mTLS authentication, the broker authenticates the client (client authentication) and the client authenticates the broker (server authentication).

mTLS listener configuration in the `Kafka` resource requires the following:

* `tls: true` to specify TLS encryption and server authentication
* `authentication.type: tls` to specify the client authentication

When a Kafka cluster is created by the Cluster Operator, it creates a new secret with the name `<cluster_name>-cluster-ca-cert`.
The secret contains a CA certificate.
The CA certificate is in xref:certificates-and-secrets-formats-{context}[PEM and PKCS #12 format].
To verify a Kafka cluster, add the CA certificate to the truststore in your client configuration.
To verify a client, add a user certificate and key to the keystore in your client configuration.
For more information on configuring a client for mTLS, see xref:con-securing-client-authentication-str[].

NOTE: TLS authentication is more commonly one-way, with one party authenticating the identity of another.
For example, when HTTPS is used between a web browser and a web server, the browser obtains proof of the identity of the web server.

[id='con-scram-sha-authentication-{context}']
== SCRAM-SHA-512 authentication

SCRAM (Salted Challenge Response Authentication Mechanism) is an authentication protocol that can establish mutual authentication using passwords.
Strimzi can configure Kafka to use SASL (Simple Authentication and Security Layer) SCRAM-SHA-512 to provide authentication on both unencrypted and encrypted client connections.

When SCRAM-SHA-512 authentication is used with a TLS connection, the TLS protocol provides the encryption, but is not used for authentication.

The following properties of SCRAM make it safe to use SCRAM-SHA-512 even on unencrypted connections:

* The passwords are not sent in the clear over the communication channel.
Instead the client and the server are each challenged by the other to offer proof that they know the password of the authenticating user.

* The server and client each generate a new challenge for each authentication exchange.
This means that the exchange is resilient against replay attacks.

When `KafkaUser.spec.authentication.type` is configured with `scram-sha-512` the User Operator will generate a random 32-character password consisting of upper and lowercase ASCII letters and numbers.