// 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 also provide the configuration for full authorization, including the ACLs that allow user operations on the source and target Kafka clusters.

When configuring user access to source and target Kafka clusters, ACLs must grant access rights to internal MirrorMaker 2 connectors and read/write access to the cluster group and internal topics used by the underlying Kafka Connect framework in the target cluster. 
If you've renamed the cluster group or internal topics, such as when xref:con-config-mm2-multiple-instances-{context}[configuring MirrorMaker 2 for multiple instances], use those names in the ACLs configuration.

Simple authorization uses ACL rules managed by the Kafka `AclAuthorizer` and `StandardAuthorizer` plugins to ensure appropriate access levels.
For more information on configuring a `KafkaUser` resource to use simple authorization, see the link:{BookURLConfiguring}#type-AclRule-reference[`AclRule` schema reference^].

.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.
--
+
.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:
      # cluster group
      - resource:
          type: group
          name: mirrormaker2-cluster
        operations:
          - Read
      # access to config.storage.topic    
      - resource:
          type: topic
          name: mirrormaker2-cluster-configs
        operations:
          - Create
          - Describe
          - DescribeConfigs
          - Read
          - Write
      # access to status.storage.topic    
      - resource:
          type: topic
          name: mirrormaker2-cluster-status
        operations:
          - Create
          - Describe
          - DescribeConfigs
          - Read
          - Write
      # access to offset.storage.topic    
      - 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>_
----
