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

[id='con-securing-client-authentication-{context}']
= User authentication

[role="_abstract"]
Use the `KafkaUser` custom resource to configure authentication credentials for users (clients) that require access to a Kafka cluster. 
Configure the credentials using the `authentication` property in `KafkaUser.spec`.
By specifying a `type`, you control what credentials are generated.

Supported authentication types:

* `tls` for mTLS authentication
* `tls-external` for mTLS authentication using external certificates
* `scram-sha-512` for SCRAM-SHA-512 authentication

If `tls` or `scram-sha-512` is specified, the User Operator creates authentication credentials when it creates the user.
If `tls-external` is specified, the user still uses mTLS, but no authentication credentials are created.
Use this option when you're providing your own certificates.
When no authentication type is specified, the User Operator does not create the user or its credentials.

You can use `tls-external` to authenticate with mTLS using a certificate issued outside the User Operator.
The User Operator does not generate a TLS certificate or a secret.
You can still manage ACL rules and quotas through the User Operator in the same way as when you're using the `tls` mechanism.
This means that you use the `CN=__USER-NAME__` format when specifying ACL rules and quotas.
_USER-NAME_ is the common name given in a TLS certificate.

== mTLS authentication

To use mTLS authentication, you set the `type` field in the `KafkaUser` resource to `tls`.

.Example user with mTLS authentication enabled
[source,yaml,subs="attributes+"]
----
apiVersion: {KafkaUserApiVersion}
kind: KafkaUser
metadata:
  name: my-user
  labels:
    strimzi.io/cluster: my-cluster
spec:
  authentication:
    type: tls
  # ...
----

The authentication type must match the equivalent configuration for the `Kafka` listener used to access the Kafka cluster.

When the user is created by the User Operator, it creates a new secret with the same name as the `KafkaUser` resource.
The secret contains a private and public key for mTLS.
The public key is contained in a user certificate, which is signed by a clients CA (certificate authority) when it is created.
All keys are in X.509 format.

NOTE: If you are using the clients CA generated by the Cluster Operator, the user certificates generated by the User Operator are also renewed when the clients CA is renewed by the Cluster Operator.

The user secret xref:certificates-and-secrets-formats-{context}[provides keys and certificates in PEM and PKCS #12 formats].

.Example secret with user credentials
[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
----

When you configure a client, you specify the following:

* *Truststore* properties for the public cluster CA certificate to verify the identity of the Kafka cluster
* *Keystore* properties for the user authentication credentials to verify the client

The configuration depends on the file format (PEM or PKCS #12).
This example uses PKCS #12 stores, and the passwords required to access the credentials in the stores. 

.Example client configuration using mTLS in PKCS #12 format
[source,properties,subs="+quotes,attributes"]
----
bootstrap.servers=__<kafka_cluster_name>__-kafka-bootstrap:9093 # <1>
security.protocol=SSL # <2>
ssl.truststore.location=/tmp/ca.p12 # <3>
ssl.truststore.password=__<truststore_password>__ # <4>
ssl.keystore.location=/tmp/user.p12 # <5>
ssl.keystore.password=__<keystore_password>__ # <6>
----
<1> The bootstrap server address to connect to the Kafka cluster. 
<2> The security protocol option when using TLS for encryption.
<3> The truststore location contains the public key certificate (`ca.p12`) for the Kafka cluster. A cluster CA certificate and password is generated by the Cluster Operator in the `<cluster_name>-cluster-ca-cert` secret when the Kafka cluster is created.
<4> The password (`ca.password`) for accessing the truststore.
<5> The keystore location contains the public key certificate (`user.p12`) for the Kafka user.
<6> The password (`user.password`) for accessing the keystore.

== mTLS authentication using a certificate issued outside the User Operator

To use mTLS authentication using a certificate issued outside the User Operator, you set the `type` field in the `KafkaUser` resource to `tls-external`.
A secret and credentials are not created for the user.

.Example user with mTLS authentication that uses a certificate issued outside the User Operator
[source,yaml,subs="attributes+"]
----
apiVersion: {KafkaUserApiVersion}
kind: KafkaUser
metadata:
  name: my-user
  labels:
    strimzi.io/cluster: my-cluster
spec:
  authentication:
    type: tls-external
  # ...
----

== SCRAM-SHA-512 authentication

To use the SCRAM-SHA-512 authentication mechanism, you set the `type` field in the `KafkaUser` resource to `scram-sha-512`.

.Example user with SCRAM-SHA-512 authentication enabled
[source,yaml,subs="attributes+"]
----
apiVersion: {KafkaUserApiVersion}
kind: KafkaUser
metadata:
  name: my-user
  labels:
    strimzi.io/cluster: my-cluster
spec:
  authentication:
    type: scram-sha-512
  # ...
----

When the user is created by the User Operator, it creates a new secret with the same name as the `KafkaUser` resource.
The secret contains the generated password in the `password` key, which is encoded with base64.
In order to use the password, it must be decoded.

.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:
  password: Z2VuZXJhdGVkcGFzc3dvcmQ= <1>
  sasl.jaas.config: b3JnLmFwYWNoZS5rYWZrYS5jb21tb24uc2VjdXJpdHkuc2NyYW0uU2NyYW1Mb2dpbk1vZHVsZSByZXF1aXJlZCB1c2VybmFtZT0ibXktdXNlciIgcGFzc3dvcmQ9ImdlbmVyYXRlZHBhc3N3b3JkIjsK <2>
----
<1> The generated password, base64 encoded.
<2> The JAAS configuration string for SASL SCRAM-SHA-512 authentication, base64 encoded.

Decoding the generated password:
----
echo "Z2VuZXJhdGVkcGFzc3dvcmQ=" | base64 --decode
----

=== Custom password configuration

When a user is created, Strimzi generates a random password.
You can use your own password instead of the one generated by Strimzi. To do so, create a secret with the password and reference it in the `KafkaUser` resource.

.Example user with a password set for SCRAM-SHA-512 authentication
[source,yaml,subs="attributes+"]
----
apiVersion: {KafkaUserApiVersion}
kind: KafkaUser
metadata:
  name: my-user
  labels:
    strimzi.io/cluster: my-cluster
spec:
  authentication:
    type: scram-sha-512
    password:
      valueFrom:
        secretKeyRef:
          name: my-secret <1>
          key: my-password <2>
  # ...
----
<1> The name of the secret containing the predefined password.
<2> The key for the password stored inside the secret.