[id="registry-persistence-kafkasql-scram"]
= Configuring Kafka storage with SCRAM security

You can configure the {kafka-streams} Operator and {operator} to use Salted Challenge Response Authentication Mechanism (SCRAM-SHA-512) for the Kafka cluster.

The recommended approach for SCRAM authentication is to use the Strimzi Kafka Access Operator, which automatically provides all connection details (bootstrap servers, TLS certificates, and SASL credentials) in a single Kubernetes Secret.

NOTE: The Kafka Access Operator integration is available starting with {registry} version 3.2.0. For earlier versions, see <<scram-env-vars>>.

.Prerequisites

* You have installed the {operator} using the OperatorHub or command line.
* You have installed the {kafka-streams} Operator or have Kafka accessible from your OpenShift cluster.
* You have installed the link:https://github.com/strimzi/kafka-access-operator[Strimzi Kafka Access Operator].

NOTE: This section assumes that the {kafka-streams} Operator and the Kafka Access Operator are available.
If the Kafka Access Operator is not available, see <<scram-env-vars>> for an alternative using environment variables.

.Procedure

. In the OpenShift web console, click *Installed Operators*, select the *{kafka-streams}* Operator details, and then the *Kafka* tab.

. Click *Create Kafka* to provision a new Kafka cluster for {registry} storage.

. Configure the `authorization` and `tls` fields to use SCRAM-SHA-512 authentication for the Kafka cluster, for example:
+
[source,yaml]
----
apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
  namespace: registry-example-kafkasql-scram
  # Change or remove the explicit namespace
spec:
  kafka:
    config:
      offsets.topic.replication.factor: 3
      transaction.state.log.replication.factor: 3
      transaction.state.log.min.isr: 2
    version: 3.7.0
    storage:
      type: ephemeral
    replicas: 3
    listeners:
      - name: tls
        port: 9093
        type: internal
        tls: true
        authentication:
          type: scram-sha-512
    authorization:
      type: simple
  entityOperator:
    topicOperator: {}
    userOperator: {}
  zookeeper:
    storage:
      type: ephemeral
    replicas: 3
----
+
The default Kafka topic name automatically created by {registry} to store data is `kafkasql-journal`. You can override this behavior or the default topic name by setting environment variables. The default values are as follows:

** `REGISTRY_KAFKASQL_TOPIC_AUTO_CREATE=true`
** `REGISTRY_KAFKASQL_TOPIC=kafkasql-journal`

+
If you decide not to create the Kafka topic manually, skip the next step.

. Click the *Kafka Topic* tab, and then *Create Kafka Topic* to create the `kafkasql-journal` topic:
+
[source,yaml]
----
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaTopic
metadata:
  name: kafkasql-journal
  labels:
    strimzi.io/cluster: my-cluster
  namespace: registry-example-kafkasql-scram
spec:
  partitions: 2
  replicas: 1
  config:
    cleanup.policy: delete
    retention.ms: -1
    retention.bytes: -1
----

. Create a *Kafka User* resource to configure SCRAM authentication and authorization for the {registry} user. You can specify a user name in the `metadata` section or use the default `my-user`.
+
[source,yaml]
----
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: my-user
  labels:
    strimzi.io/cluster: my-cluster
  namespace: registry-example-kafkasql-scram
spec:
  authentication:
    type: scram-sha-512
  authorization:
    acls:
      - operation: All
        resource:
          name: '*'
          patternType: literal
          type: topic
      - operation: All
        resource:
          name: '*'
          patternType: literal
          type: cluster
      - operation: All
        resource:
          name: '*'
          patternType: literal
          type: transactionalId
      - operation: All
        resource:
          name: '*'
          patternType: literal
          type: group
    type: simple
----
+
NOTE: This simple example assumes admin permissions and creates the Kafka topic automatically. You must configure the `authorization` section specifically for the topics and resources that the {registry} requires.
+
The following example shows the minimum configuration required when the Kafka topic is created manually:
+
[source,yaml]
----
 ...
  authorization:
    acls:
    - operations:
        - Read
        - Write
      resource:
        name: kafkasql-journal
        patternType: literal
        type: topic
    - operations:
        - Read
        - Write
      resource:
        name: apicurio-registry-
        patternType: prefix
        type: group
    type: simple
----

. Create a *KafkaAccess* resource that binds the Kafka cluster and user. The Kafka Access Operator produces a Secret containing all connection details (bootstrap servers, TLS certificates, and SCRAM credentials):
+
[source,yaml]
----
apiVersion: access.strimzi.io/v1alpha1
kind: KafkaAccess
metadata:
  name: my-kafka-access
  namespace: registry-example-kafkasql-scram
spec:
  kafka:
    name: my-cluster
    namespace: registry-example-kafkasql-scram
    listener: tls
  user:
    kind: KafkaUser
    apiGroup: kafka.strimzi.io
    name: my-user
    namespace: registry-example-kafkasql-scram
----

. Configure the following example settings to deploy the {registry} using the `kafkaAccessSecretName` field:
+
[source,yaml]
----
include::example$apicurioregistry_kafkasql_scram_cr.yaml[]
----

IMPORTANT: The `kafkaAccessSecretName` replaces the need for manual `bootstrapServers`, `tls`, and `auth` configuration. The {operator} reads all connection details from the referenced Secret at pod startup.

[id="scram-env-vars"]
== Alternative: Configuring SCRAM using environment variables

If the Kafka Access Operator is not available, you can configure SCRAM authentication using environment variables in the `ApicurioRegistry3` CR:

[source,yaml]
----
apiVersion: registry.apicur.io/v1
kind: ApicurioRegistry3
metadata:
  name: example-apicurioregistry-kafkasql-scram
spec:
  app:
    storage:
      type: kafkasql
      kafkasql:
        bootstrapServers: "my-cluster-kafka-bootstrap.registry-example-kafkasql-scram.svc:9093"
        tls:
          truststoreSecretRef:
            name: my-cluster-cluster-ca-cert
          truststorePasswordSecretRef:
            name: my-cluster-cluster-ca-cert
    env:
      - name: APICURIO_KAFKA_COMMON_SECURITY_PROTOCOL
        value: "SASL_SSL"
      - name: APICURIO_KAFKA_COMMON_SASL_MECHANISM
        value: "SCRAM-SHA-512"
      - name: APICURIO_KAFKA_COMMON_SASL_JAAS_CONFIG
        valueFrom:
          secretKeyRef:
            name: my-user
            key: sasl.jaas.config
    ingress:
      host: example-kafkasql-scram-app.apps.cluster.example
  ui:
    ingress:
      host: example-kafkasql-scram-ui.apps.cluster.example
----

NOTE: When using manual SCRAM configuration, you must construct the `sasl.jaas.config` value or reference it from a Secret. The value follows the format: `org.apache.kafka.common.security.scram.ScramLoginModule required username="my-user" password="<password>";`.
