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

[id='con-securing-kafka-authentication-{context}']
= Listener authentication

[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^]

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.

NOTE: Listeners cannot be configured to use the ports reserved for inter-broker communication (9091 or 9090) and metrics (9404).

.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: external
        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 brokers and ZooKeeper pods.

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 12-character password consisting of upper and lowercase ASCII letters and numbers.

[id='assembly-kafka-broker-listener-network-policies-{context}']
== Network policies

By default, Strimzi automatically creates a `NetworkPolicy` resource for every listener that is enabled on a Kafka broker.
This `NetworkPolicy` allows applications to connect to listeners in all namespaces.
Use network policies as part of the listener configuration.

If you want to restrict access to a listener at the network level to only selected applications or namespaces, use the `networkPolicyPeers` property.
Each listener can have a different link:{BookURLConfiguring}#configuration-listener-network-policy-reference[`networkPolicyPeers` configuration].
For more information on network policy peers, refer to the {K8sNetworkPolicyPeerAPI}.

If you want to use custom network policies, you can set the `STRIMZI_NETWORK_POLICY_GENERATION` environment variable to `false` in the Cluster Operator configuration.
For more information, see xref:ref-operator-cluster-{context}[].

NOTE: Your configuration of Kubernetes must support ingress `NetworkPolicies` in order to use network policies in Strimzi.

== Providing listener certificates
You can provide your own server certificates, called _Kafka listener certificates_, for TLS listeners or external listeners which have TLS encryption enabled.
For more information, see xref:proc-installing-certs-per-listener-{context}[].

[role="_additional-resources"]
.Additional resources

* link:{BookURLConfiguring}#type-GenericKafkaListener-reference[`GenericKafkaListener` schema reference]
