:_mod-docs-content-type: PROCEDURE

// Module included in the following assemblies:

// assembly-config.adoc

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

[role="_abstract"]
This procedure describes how to secure connections between MirrorMaker 2 and the source and target Kafka clusters.

You configure these settings independently for the source Kafka cluster and the target Kafka cluster.
You also need separate `KafkaUser` resources to provide the credentials required for MirrorMaker 2 to connect to the source and target 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 creating the cluster and user credentials, reference them in the `KafkaMirrorMaker2` resource to enable 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)].

To enforce the permissions defined in each `KafkaUser` resource, the Kafka listeners on the source and target clusters must enable an authentication mechanism that allows MirrorMaker 2 to identify as those users.
Configure authentication on the Kafka listeners in the `Kafka` resources and in the `KafkaMirrorMaker2` resource.

.Prerequisites

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

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 `StandardAuthorizer` plugin 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^].

The procedure assumes that the source and target Kafka clusters are installed to separate namespaces.
If you want to use the Topic Operator, you must install the clusters in separate namespaces.
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.
+
Configure listener authentication and enable authorization on each Kafka cluster.
+
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}
    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
  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}
    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
  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: "*"
          patternType: literal
        operations:
          - DescribeConfigs
          - Read
      # MirrorCheckpointConnector
      - resource:
          type: cluster
        operations:
          - Describe
      - resource: # Needed for every group for which offsets are synced
          type: group
          name: "*"
          patternType: literal
        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: "*"
          patternType: literal
        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: "*"
          patternType: literal
        operations:
          - Read
          - Describe
----
+
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 client 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
  target:
    alias: "my-cluster-target"
    bootstrapServers: my-cluster-target-kafka-bootstrap:9092
    groupId: my-mirror-maker2-group
    configStorageTopic: my-mirror-maker2-config
    offsetStorageTopic: my-mirror-maker2-offset
    statusStorageTopic: my-mirror-maker2-status
    tls: # <1>
      trustedCertificates:
        - secretName: my-target-cluster-cluster-ca-cert
          pattern: "*.crt"
    authentication: # <2>
      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:
    - source:
        alias: "my-source-cluster"
        bootstrapServers: my-source-cluster-kafka-bootstrap:9093
        tls: # <3>
          trustedCertificates:
            - secretName: my-source-cluster-cluster-ca-cert
              pattern: "*.crt"
        authentication: # <4>
          type: tls
          certificateAndKey:
            secretName: my-source-user
            certificate: user.crt
            key: user.key
      sourceConnector:
        config:
          replication.factor: 1
          offset-syncs.topic.replication.factor: -1
          sync.topic.acls.enabled: "false"
      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 target Kafka cluster.
<2> The user authentication for accessing the target Kafka cluster.
<3> 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.
<4> The user authentication for accessing the source Kafka cluster using the TLS mechanism. Supported authentication methods include `tls`, `scram-sha-256`, `scram-sha-512`, `plain`, and `custom`.

. Apply the changes to the `KafkaMirrorMaker2` resource to the same namespace as the target Kafka cluster.
