// Module included in the following assemblies:
//
// deploying/assembly_deploy-verify.adoc
// getting-started.adoc

[id='setup-external-clients-{context}']
= Setting up client access to a Kafka cluster using listeners

[role="_abstract"]
Using the address of the Kafka cluster, you can provide access to a client in the same Kubernetes cluster; or provide external access to a client on a different Kubernetes namespace or outside Kubernetes entirely.
This procedure shows how to configure client access to a Kafka cluster from outside Kubernetes or from another Kubernetes cluster.

A Kafka listener provides access to the Kafka cluster.
Client access is secured using the following configuration:

. An external listener is configured for the Kafka cluster, with TLS encryption and mTLS authentication, and Kafka `simple` authorization enabled.
. A `KafkaUser` is created for the client, with mTLS authentication, and Access Control Lists (ACLs) defined for `simple` authorization.

You can configure your listener to use mutual `tls`, `scram-sha-512`, or `oauth` authentication.
mTLS always uses encryption, but encryption is also recommended when using SCRAM-SHA-512 and OAuth 2.0 authentication.

You can configure `simple`, `oauth`, `opa`, or `custom` authorization for Kafka brokers.
When enabled, authorization is applied to all enabled listeners.

When you configure the `KafkaUser` authentication and authorization mechanisms, ensure they match the equivalent Kafka configuration:

* `KafkaUser.spec.authentication` matches `Kafka.spec.kafka.listeners[*].authentication`
* `KafkaUser.spec.authorization` matches `Kafka.spec.kafka.authorization`

You should have at least one listener supporting the authentication you want to use for the `KafkaUser`.

NOTE: Authentication between Kafka users and Kafka brokers depends on the authentication settings for each.
For example, it is not possible to authenticate a user with mTLS if it is not also enabled in the Kafka configuration.

Strimzi operators automate the configuration process and create the certificates required for authentication:

* The Cluster Operator creates the listeners and sets up the cluster and client certificate authority (CA) certificates to enable authentication with the Kafka cluster.
* The User Operator creates the user representing the client and the security credentials used for client authentication, based on the chosen authentication type.

You add the certificates to your client configuration.

In this procedure, the CA certificates generated by the Cluster Operator are used, but you can replace them by xref:installing-your-own-ca-certificates-str[installing your own certificates].
You can also configure your listener to xref:proc-installing-certs-per-listener-{context}[use a Kafka listener certificate managed by an external CA (certificate authority)].

Certificates are available in PEM (.crt) and PKCS #12 (.p12) formats.
This procedure uses PEM certificates.
Use PEM certificates with clients that use certificates in X.509 format.

NOTE: For internal clients in the same Kubernetes cluster and namespace, you can mount the cluster CA certificate in the pod specification.
For more information, see xref:configuring-internal-clients-to-trust-cluster-ca-{context}[Configuring internal clients to trust the cluster CA].

.Prerequisites

* The Kafka cluster is available for connection by a client running outside the Kubernetes cluster
* The Cluster Operator and User Operator are running in the cluster

.Procedure

. Configure the Kafka cluster with a Kafka listener.
+
* Define the authentication required to access the Kafka broker through the listener.
* Enable authorization on the Kafka broker.
+
.Example listener configuration
[source,yaml,subs="+quotes,attributes"]
----
apiVersion: {KafkaApiVersion}
kind: Kafka
metadata:
  name: my-cluster
  namespace: myproject
spec:
  kafka:
    # ...
    listeners: # <1>
    - name: external # <2>
      port: 9094 <3>
      type: _<listener_type>_ # <4>
      tls: true # <5>
      authentication:
        type: tls # <6>
      configuration: # <7>
        #...
    authorization: # <8>
      type: simple
      superUsers:
        - super-user-name # <9>
  # ...
----
<1> Configuration options for enabling external listeners are described in the link:{BookURLConfiguring}#type-GenericKafkaListener-reference[Generic Kafka listener schema reference^].
<2> Name to identify the listener. Must be unique within the Kafka cluster.
<3> Port number used by the listener inside Kafka. The port number has to be unique within a given Kafka cluster. Allowed port numbers are 9092 and higher with the exception of ports 9404 and 9999, which are already used for Prometheus and JMX. Depending on the listener type, the port number might not be the same as the port number that connects Kafka clients.
<4> External listener type specified as `route` (OpenShift only), `loadbalancer`, `nodeport` or `ingress` (Kubernetes only). An internal listener is specified as `internal` or `cluster-ip`.
<5> Required. TLS encryption on the listener. For `route` and `ingress` type listeners it must be set to `true`. For mTLS authentication, also use the `authentication` property. 
<6> Client authentication mechanism on the listener. For server and client authentication using mTLS, you specify `tls: true` and `authentication.type: tls`. 
<7> (Optional) Depending on the requirements of the listener type, you can specify additional link:{BookURLConfiguring}#type-GenericKafkaListenerConfiguration-reference[listener configuration^].
<8> Authorization specified as `simple`, which uses the `AclAuthorizer` Kafka plugin.
<9> (Optional) Super users can access all brokers regardless of any access restrictions defined in ACLs.
+
WARNING: An OpenShift Route address comprises the name of the Kafka cluster, the name of the listener, and the name of the namespace it is created in.
For example, `my-cluster-kafka-listener1-bootstrap-myproject` (_CLUSTER-NAME_-kafka-_LISTENER-NAME_-bootstrap-_NAMESPACE_).
If you are using a `route` listener type, be careful that the whole length of the address does not exceed a maximum limit of 63 characters.

. Create or update the `Kafka` resource.
+
[source,shell,subs=+quotes]
----
kubectl apply -f _<kafka_configuration_file>_
----
+
The Kafka cluster is configured with a Kafka broker listener using mTLS authentication.
+
A service is created for each Kafka broker pod.
+
A service is created to serve as the _bootstrap address_ for connection to the Kafka cluster.
+
A service is also created as the _external bootstrap address_ for external connection to the Kafka cluster using `nodeport` listeners.
+
The cluster CA certificate to verify the identity of the kafka brokers is also created in the secret `_<cluster_name>_-cluster-ca-cert`.
+
NOTE: If you scale your Kafka cluster while using external listeners, it might trigger a rolling update of all Kafka brokers. This depends on the configuration.

. Retrieve the bootstrap address you can use to access the Kafka cluster from the status of the `Kafka` resource.
+
[source,shell,subs=+quotes]
kubectl get kafka _<kafka_cluster_name>_ -o=jsonpath='{.status.listeners[?(@.name=="_<listener_name>_")].bootstrapServers}{"\n"}'
+
For example:
+
[source,shell,subs=+quotes]
kubectl get kafka my-cluster -o=jsonpath='{.status.listeners[?(@.name=="external")].bootstrapServers}{"\n"}'
+
Use the bootstrap address in your Kafka client to connect to the Kafka cluster.

. Create or modify a user representing the client that requires access to the Kafka cluster.
+
* Specify the same authentication type as the `Kafka` listener.
* Specify the authorization ACLs for `simple` authorization.
+
.Example user configuration
[source,yaml,subs="+quotes,attributes"]
----
apiVersion: {KafkaUserApiVersion}
kind: KafkaUser
metadata:
  name: my-user
  labels:
    strimzi.io/cluster: my-cluster # <1>
spec:
  authentication:
    type: tls # <2>
  authorization:
    type: simple
    acls: # <3>
      - resource:
          type: topic
          name: my-topic
          patternType: literal
        operations:
          - Describe
          - Read
      - resource:
          type: group
          name: my-group
          patternType: literal
        operations:
          - Read
----
<1> The label must match the label of the Kafka cluster.
<2> Authentication specified as mutual `tls`.
<3> Simple authorization requires an accompanying list of ACL rules to apply to the user.
The rules define the operations allowed on Kafka resources based on the _username_ (`my-user`).

. Create or modify the `KafkaUser` resource.
+
[source,shell,subs="+quotes,attributes"]
kubectl apply -f _USER-CONFIG-FILE_
+
The user is created, as well as a secret with the same name as the `KafkaUser` resource.
The secret contains a public and private key for mTLS authentication.
+
.Example secret
[source,yaml,subs="+quotes,attributes"]
----
apiVersion: v1
kind: Secret
metadata:
  name: my-user
  labels:
    strimzi.io/kind: KafkaUser
    strimzi.io/cluster: my-cluster
type: Opaque
data:
  ca.crt: _<public_key>_ # Public key of the clients CA
  user.crt: _<user_certificate>_ # Public key of the user
  user.key: _<user_private_key>_ # Private key of the user
  user.p12: _<store>_ # PKCS #12 store for user certificates and keys
  user.password: _<password_for_store>_ # Protects the PKCS #12 store
----

. Extract the cluster CA certificate from the `_<cluster_name>_-cluster-ca-cert` secret of the Kafka cluster.
+
[source,shell,subs="+quotes"]
kubectl get secret _<cluster_name>_-cluster-ca-cert -o jsonpath='{.data.ca\.crt}' | base64 -d > ca.crt

. Extract the user CA certificate from the `_<user_name>_` secret.
+
[source,shell,subs="+quotes"]
kubectl get secret _<user_name>_ -o jsonpath='{.data.user\.crt}' | base64 -d > user.crt

. Extract the private key of the user from the `_<user_name>_` secret.
+
[source,shell,subs="+quotes"]
kubectl get secret _<user_name>_ -o jsonpath='{.data.user\.key}' | base64 -d > user.key

. Configure your client with the bootstrap address hostname and port for connecting to the Kafka cluster:
+
[source,env,subs="+quotes,attributes"]
----
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "_<hostname>:<port>_");
----

. Configure your client with the truststore credentials to verify the identity of the Kafka cluster.
+
Specify the public cluster CA certificate.  
+
.Example truststore configuration
[source,env,subs="+quotes,attributes"]
----
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");
props.put(SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG, "PEM");
props.put(SslConfigs.SSL_TRUSTSTORE_CERTIFICATES_CONFIG, "_<ca.crt_file_content>_");
----
+
SSL is the specified security protocol for mTLS authentication.
Specify `SASL_SSL` for SCRAM-SHA-512 authentication over TLS.
PEM is the file format of the truststore. 

. Configure your client with the keystore credentials to verify the user when connecting to the Kafka cluster.
+
Specify the public certificate and private key. 
+
.Example keystore configuration
[source,env,subs="+quotes,attributes"]
----
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");
props.put(SslConfigs.SSL_KEYSTORE_TYPE_CONFIG, "PEM");
props.put(SslConfigs.SSL_KEYSTORE_CERTIFICATE_CHAIN_CONFIG, "_<user.crt_file_content>_");
props.put(SslConfigs.SSL_KEYSTORE_KEY_CONFIG, "_<user.key_file_content>_");
----
+
Add the keystore certificate and the private key directly to the configuration.
Add as a single-line format.
Between the `BEGIN CERTIFICATE` and `END CERTIFICATE` delimiters, start with a newline character (`\n`).
End each line from the original certificate with `\n` too.
+
.Example keystore configuration
[source,env,subs="+quotes,attributes"]
----
props.put(SslConfigs.SSL_KEYSTORE_CERTIFICATE_CHAIN_CONFIG, "-----BEGIN CERTIFICATE----- \n__<user_certificate_content_line_1>__\n__<user_certificate_content_line_n>__\n-----END CERTIFICATE---");
props.put(SslConfigs.SSL_KEYSTORE_KEY_CONFIG, "----BEGIN PRIVATE KEY-----\n__<user_key_content_line_1>__\n__<user_key_content_line_n>__\n-----END PRIVATE KEY-----");
----
+
[role="_additional-resources"]
.Additional resources
* xref:con-securing-kafka-authentication-str[]
* xref:con-securing-kafka-authorization-str[]
* If you are using an authorization server, you can use token-based authentication and authorization:
** xref:assembly-oauth-authentication_str[] 
** xref:assembly-oauth-authorization_str[]