:_mod-docs-content-type: PROCEDURE

// Module included in the following assemblies:
//
// deploying/assembly-securing-access.adoc

[id='setup-external-clients-{context}']
= Example: Setting up secure client access

[role="_abstract"]
This procedure shows how to configure client access to a Kafka cluster from outside Kubernetes or from another Kubernetes cluster.
It's split into two parts:

* Securing Kafka brokers
* Securing user access to Kafka

.Resource configuration

Client access to the Kafka cluster is secured with the following configuration:

. An external listener is configured with TLS encryption and mutual TLS (mTLS) authentication in the `Kafka` resource, as well as `simple` authorization.
. A `KafkaUser` is created for the client, utilizing mTLS authentication, and Access Control Lists (ACLs) are defined for `simple` authorization.

At least one listener supporting the desired authentication must be configured for the `KafkaUser`.

Listeners can be configured for mutual `TLS`, `SCRAM-SHA-512`, or `custom` authentication. 
While mTLS always uses encryption, it's also recommended when using SCRAM-SHA-512 or a custom configuration for OAuth 2.0 authentication.

Authorization options for Kafka include `simple` or `custom`. 
When enabled, authorization is applied to all enabled listeners.

To ensure compatibility between Kafka and clients, configuration of the following authentication and authorization mechanisms must align:

* For `type: tls` and `type: scram-sha-512` authentication types, `Kafka.spec.kafka.listeners[*].authentication` must match `KafkaUser.spec.authentication`
* For `type: simple` authorization,`Kafka.spec.kafka.authorization` must match `KafkaUser.spec.authorization`

For example, mTLS authentication for a user is only possible if it's also enabled in the Kafka configuration.

.Automation and certificate management

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 within 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.
Alternatively, you can replace them by xref:installing-your-own-ca-certificates-str[installing your own custom CA certificates].
You can also configure listeners to xref:proc-installing-certs-per-listener-{context}[use Kafka listener certificates managed by an external CA].

Certificates are available in PEM (.crt) and PKCS #12 (.p12) formats.
This procedure uses PEM certificates.
Use PEM certificates with clients that support the X.509 certificate 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

[id='proc-securing-kafka-{context}']
== Securing Kafka brokers

. 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="+attributes"]
----
apiVersion: {KafkaApiVersion}
kind: Kafka
metadata:
  name: my-cluster
  namespace: myproject
spec:
  kafka:
    # ...
    listeners: # <1>
    - name: external1 # <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 `StandardAuthorizer` 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 Kafka cluster name, the listener name, the project name, and the domain of the router.
For example, `my-cluster-kafka-external1-bootstrap-my-project.domain.com` (<cluster_name>-kafka-<listener_name>-bootstrap-<namespace>.<domain>). 
Each DNS label (between periods "`.`") must not exceed 63 characters, and the total length of the address must not exceed 255 characters.

. Apply the changes to the `Kafka` resource configuration.
+
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]
kubectl get kafka <kafka_cluster_name> -o=jsonpath='{.status.listeners[?(@.name=="<listener_name>")].bootstrapServers}{"\n"}'
+
For example:
+
[source,shell]
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.

[id='proc-configuring-secure-kafka-user-{context}']
== Securing user access to Kafka

. 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="+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`).

. Apply the changes to the `KafkaUser` resource configuration.
+
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 with user credentials
[source,yaml,subs="+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 used to sign this user certificate
  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]
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]
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]
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="+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="+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="+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="+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-----");
----