// Module included in the following assemblies:

// assembly-config.adoc

[id='proc-config-mirrormaker2-securing-connection-{context}']
= Securing a Kafka MirrorMaker 2 deployment

[role="_abstract"]
This procedure describes in outline the configuration required to secure a MirrorMaker 2 deployment.

You need separate configuration for the source Kafka cluster and the target Kafka cluster.
You also need separate user configuration to provide the credentials required for MirrorMaker to connect to the source and target Kafka clusters.

For the Kafka clusters, you specify internal listeners for secure connections within a Kubernetes cluster and external listeners for connections outside the Kubernetes cluster.

You can configure authentication and authorization mechanisms.
The security options implemented for the source and target Kafka clusters must be compatible with the security options implemented for MirrorMaker 2.

After you have created the cluster and user authentication credentials, you specify them in your MirrorMaker configuration for secure connections.

NOTE: In this procedure, the 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-str[use a Kafka listener certificate managed by an external CA (certificate authority)].

.Before you start
Before starting this procedure, take a look at the xref:config-examples-{context}[example configuration files] provided by Strimzi.
They include examples for securing a deployment of MirrorMaker 2 using mTLS or SCRAM-SHA-512 authentication.
The examples specify internal listeners for connecting within a Kubernetes cluster.

The examples provide the configuration for full authorization, including all the ACLs needed by MirrorMaker 2 to allow operations on the source and target Kafka clusters.

.Prerequisites

* Strimzi is running
* Separate namespaces for source and target clusters

The procedure assumes that the source and target Kafka clusters are installed to separate namespaces
If you want to use the Topic Operator, you'll need to do this.
The Topic Operator only watches a single cluster in a specified namespace.

By separating the clusters into namespaces, you will need to copy the cluster secrets so they can be accessed outside the namespace.
You need to reference the secrets in the MirrorMaker configuration.

.Procedure

. Configure two `Kafka` resources, one to secure the source Kafka cluster and one to secure the target Kafka cluster.
+
You can add listener configuration for authentication and enable authorization.
+
In this example, an internal listener is configured for a Kafka cluster with TLS encryption and mTLS authentication.
Kafka `simple` authorization is enabled.
+
.Example source Kafka cluster configuration with TLS encryption and mTLS authentication
[source,yaml,subs="attributes+"]
----
apiVersion: {KafkaApiVersion}
kind: Kafka
metadata:
  name: my-source-cluster
spec:
  kafka:
    version: {DefaultKafkaVersion}
    replicas: 1
    listeners:
      - name: tls
        port: 9093
        type: internal
        tls: true
        authentication:
          type: tls
    authorization:
      type: simple
    config:
      offsets.topic.replication.factor: 1
      transaction.state.log.replication.factor: 1
      transaction.state.log.min.isr: 1
      default.replication.factor: 1
      min.insync.replicas: 1
      inter.broker.protocol.version: "{DefaultInterBrokerVersion}"
    storage:
      type: jbod
      volumes:
      - id: 0
        type: persistent-claim
        size: 100Gi
        deleteClaim: false
  zookeeper:
    replicas: 1
    storage:
      type: persistent-claim
      size: 100Gi
      deleteClaim: false
  entityOperator:
    topicOperator: {}
    userOperator: {}
----
+
.Example target Kafka cluster configuration with TLS encryption and mTLS authentication
[source,yaml,subs="attributes+"]
----
apiVersion: {KafkaApiVersion}
kind: Kafka
metadata:
  name: my-target-cluster
spec:
  kafka:
    version: {DefaultKafkaVersion}
    replicas: 1
    listeners:
      - name: tls
        port: 9093
        type: internal
        tls: true
        authentication:
          type: tls
    authorization:
      type: simple
    config:
      offsets.topic.replication.factor: 1
      transaction.state.log.replication.factor: 1
      transaction.state.log.min.isr: 1
      default.replication.factor: 1
      min.insync.replicas: 1
      inter.broker.protocol.version: "{DefaultInterBrokerVersion}"
    storage:
      type: jbod
      volumes:
        - id: 0
          type: persistent-claim
          size: 100Gi
          deleteClaim: false
  zookeeper:
    replicas: 1
    storage:
      type: persistent-claim
      size: 100Gi
      deleteClaim: false
  entityOperator:
    topicOperator: {}
    userOperator: {}
----

. Create or update the `Kafka` resources in separate namespaces.
+
[source,shell,subs=+quotes]
----
kubectl apply -f _<kafka_configuration_file>_ -n _<namespace>_
----
+
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 certificates are created in the secret `_<cluster_name>_-cluster-ca-cert`.

. Configure two `KafkaUser` resources, one for a user of the source Kafka cluster and one for a user of the target Kafka cluster.
+
--
.. Configure the same authentication and authorization types as the corresponding source and target Kafka cluster.
For example, if you used `tls` authentication and the `simple` authorization type in the `Kafka` configuration for the source Kafka cluster, use the same in the `KafkaUser` configuration.

.. Configure the ACLs needed by MirrorMaker 2 to allow operations on the source and target Kafka clusters.
+
The ACLs are used by the internal MirrorMaker connectors, and by the underlying Kafka Connect framework.
--
+
.Example source user configuration for mTLS authentication
[source,yaml,subs="attributes+"]
----
apiVersion: {KafkaUserApiVersion}
kind: KafkaUser
metadata:
  name: my-source-user
  labels:
    strimzi.io/cluster: my-source-cluster
spec:
  authentication:
    type: tls
  authorization:
    type: simple
    acls:
      # MirrorSourceConnector
      - resource: # Not needed if offset-syncs.topic.location=target
          type: topic
          name: mm2-offset-syncs.my-target-cluster.internal
        operations:
          - Create
          - DescribeConfigs
          - Read
          - Write
      - resource: # Needed for every topic which is mirrored
          type: topic
          name: "*"
        operations:
          - DescribeConfigs
          - Read
      # MirrorCheckpointConnector
      - resource:
          type: cluster
        operations:
          - Describe
      - resource: # Needed for every group for which offsets are synced
          type: group
          name: "*"
        operations:
          - Describe
      - resource: # Not needed if offset-syncs.topic.location=target
          type: topic
          name: mm2-offset-syncs.my-target-cluster.internal
        operations:
          - Read
----
+
.Example target user configuration for mTLS authentication
[source,yaml,subs="attributes+"]
----
apiVersion: {KafkaUserApiVersion}
kind: KafkaUser
metadata:
  name: my-target-user
  labels:
    strimzi.io/cluster: my-target-cluster
spec:
  authentication:
    type: tls
  authorization:
    type: simple
    acls:
      # Underlying Kafka Connect internal topics to store configuration, offsets, or status
      - resource:
          type: group
          name: mirrormaker2-cluster
        operations:
          - Read
      - resource:
          type: topic
          name: mirrormaker2-cluster-configs
        operations:
          - Create
          - Describe
          - DescribeConfigs
          - Read
          - Write
      - resource:
          type: topic
          name: mirrormaker2-cluster-status
        operations:
          - Create
          - Describe
          - DescribeConfigs
          - Read
          - Write
      - resource:
          type: topic
          name: mirrormaker2-cluster-offsets
        operations:
          - Create
          - Describe
          - DescribeConfigs
          - Read
          - Write
      # MirrorSourceConnector
      - resource: # Needed for every topic which is mirrored
          type: topic
          name: "*"
        operations:
          - Create
          - Alter
          - AlterConfigs
          - Write
      # MirrorCheckpointConnector
      - resource:
          type: cluster
        operations:
          - Describe
      - resource:
          type: topic
          name: my-source-cluster.checkpoints.internal
        operations:
          - Create
          - Describe
          - Read
          - Write
      - resource: # Needed for every group for which the offset is synced
          type: group
          name: "*"
        operations:
          - Read
          - Describe
      # MirrorHeartbeatConnector
      - resource:
          type: topic
          name: heartbeats
        operations:
          - Create
          - Describe
          - Write
----
+
NOTE: You can use a certificate issued outside the User Operator by setting `type` to `tls-external`.
For more information, see the link:{BookURLConfiguring}#type-KafkaUserSpec-reference[`KafkaUserSpec` schema reference^].

. Create or update a `KafkaUser` resource in each of the namespaces you created for the source and target Kafka clusters.
+
[source,shell,subs=+quotes]
----
kubectl apply -f _<kafka_user_configuration_file>_ -n _<namespace>_
----
+
The User Operator creates the users representing the client (MirrorMaker), and the security credentials used for client authentication, based on the chosen authentication type.
+
The User Operator creates a new secret with the same name as the `KafkaUser` resource.
The secret contains a private and public key for mTLS authentication.
The public key is contained in a user certificate, which is signed by the clients CA.

. Configure a `KafkaMirrorMaker2` resource with the authentication details to connect to the source and target Kafka clusters.
+
.Example MirrorMaker 2 configuration with TLS encryption and mTLS authentication
[source,yaml,subs="attributes+"]
----
apiVersion: {KafkaMirrorMaker2ApiVersion}
kind: KafkaMirrorMaker2
metadata:
  name: my-mirror-maker-2
spec:
  version: {DefaultKafkaVersion}
  replicas: 1
  connectCluster: "my-target-cluster"
  clusters:
    - alias: "my-source-cluster"
      bootstrapServers: my-source-cluster-kafka-bootstrap:9093
      tls: # <1>
        trustedCertificates:
          - secretName: my-source-cluster-cluster-ca-cert
            certificate: ca.crt
      authentication: # <2>
        type: tls
        certificateAndKey:
          secretName: my-source-user
          certificate: user.crt
          key: user.key
    - alias: "my-target-cluster"
      bootstrapServers: my-target-cluster-kafka-bootstrap:9093
      tls: # <3>
        trustedCertificates:
          - secretName: my-target-cluster-cluster-ca-cert
            certificate: ca.crt
      authentication: # <4>
        type: tls
        certificateAndKey:
          secretName: my-target-user
          certificate: user.crt
          key: user.key
      config:
        # -1 means it will use the default replication factor configured in the broker
        config.storage.replication.factor: -1
        offset.storage.replication.factor: -1
        status.storage.replication.factor: -1
  mirrors:
    - sourceCluster: "my-source-cluster"
      targetCluster: "my-target-cluster"
      sourceConnector:
        config:
          replication.factor: 1
          offset-syncs.topic.replication.factor: 1
          sync.topic.acls.enabled: "false"
      heartbeatConnector:
        config:
          heartbeats.topic.replication.factor: 1
      checkpointConnector:
        config:
          checkpoints.topic.replication.factor: 1
          sync.group.offsets.enabled: "true"
      topicsPattern: "topic1|topic2|topic3"
      groupsPattern: "group1|group2|group3"
----
<1> The TLS certificates for the source Kafka cluster. If they are in a separate namespace, copy the cluster secrets from the namespace of the Kafka cluster.
<2> The user authentication for accessing the source Kafka cluster using the TLS mechanism.
<3> The TLS certificates for the target Kafka cluster.
<4> The user authentication for accessing the target Kafka cluster.

. Create or update the `KafkaMirrorMaker2` resource in the same namespace as the target Kafka cluster.
+
[source,shell,subs=+quotes]
----
kubectl apply -f _<mirrormaker2_configuration_file>_ -n _<namespace_of_target_cluster>_
----
