<a id="cluster-link-security"></a>

# Manage Security for Cluster Linking on Confluent Platform

All security configurations used to connect to the source cluster can be configured on
the cluster link when the link is created. Each link is associated with exactly
one link credential that will be used for authentication of connections to the
source cluster using that link. Different cluster links on the same cluster may
use different security credentials. The link credential must be granted
appropriate permissions on the source cluster.

## Authentication

The following example shows how to configure SASL_SSL with GSSAPI as the SASL
mechanism for the cluster link to talk to the source cluster. You can set
these configurations using a `config-file`, as described in the section on
[how to set properties on a cluster link](configs.md#cluster-link-specific-configs).

```bash
security.protocol=SASL_SSL
ssl.truststore.location=/path/to/truststore.p12
ssl.truststore.password=truststore-password
ssl.truststore.type=PKCS12
sasl.mechanism=GSSAPI
sasl.kerberos.service.name=kafka
sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \
    useKeyTab=true \
    storeKey=true \
    keyTab="/path/to/link.keytab" \
    principal="clusterlink1@EXAMPLE.COM";
```

Cluster Linking configurations should include client-side TLS/SSL and SASL/GSSAPI configuration options for
connections to the source cluster in this scenario.

If you reference a `keystore`/`truststore` directly (for example, `keystore.jks`),
the same files must be available in the same location on each of the brokers

For details on creating TLS/SSL key and trust stores, see [Use TLS Authentication in Confluent Platform](../../security/authentication/mutual-tls/overview.md#kafka-ssl-authentication). For details on
SASL/GSSAPI, see [Configure GSSAPI in Confluent Platform clusters](../../security/authentication/sasl/gssapi/overview.md#kafka-sasl-auth-gssapi).

To configure cluster links to use other SASL mechanisms, include client-side
security configurations for that mechanism. See [SASL](../../security/authentication/overview.md#kafka-sasl-auth) for other
supported mechanisms. To use mutual TLS authentication as the security protocol,
a key store should also be configured for the link. See
[Use TLS Authentication in Confluent Platform](../../security/authentication/mutual-tls/overview.md#kafka-ssl-authentication) for details.

#### NOTE
The cluster links use source credentials configured on the link to communicate
with the source cluster. These credentials must be valid in order for the link to function.

<a id="cluster-link-allowed-urls"></a>

## Restrict OAuth bearer token endpoint URLs

The `org.apache.kafka.sasl.oauthbearer.allowed.urls` system property restricts
which URLs are allowed as SASL/OAUTHBEARER token or JSON Web Key Set (JWKS)
endpoints for cluster links. This security control prevents cluster links from
connecting to unauthorized OAuth endpoints.

This property defaults to an empty list, which denies all URLs. To allow
specific endpoints, set this system property to a comma-separated list of
URLs when starting the Kafka broker:

```bash
-Dorg.apache.kafka.sasl.oauthbearer.allowed.urls=https://auth.example.com,https://trusted-oauth.example.com
```

Allowed URL types:

- HTTPS endpoints (`https://`)
- File-based token locations (`file:///path/to/token`)

Use `*` as the value to allow any endpoint. Otherwise, each URL in the list
is matched exactly against the endpoint URL. This property does not support
prefix or pattern matching.

#### IMPORTANT
Set the allowed URL list before you rely on SASL or OAUTHBEARER
authentication for cluster links. An empty list denies all
token and JWKS endpoint URLs.

## Restrict cluster link token handler URLs

When a cluster link authenticates to a Metadata Service (MDS) or token endpoint using an
OAuth token login callback handler, set the
`confluent.cluster.link.token.handler.url.validation.enable` broker
configuration to `true` to validate the configured metadata server URLs
against the allowed URL list before the broker creates an outbound client
for those URLs. This control stops a cluster link from connecting to
arbitrary or internal URLs.

This configuration defaults to `false`, and the allowed URL list defaults
to an empty list. An empty allowed URL list denies all URLs, so set the
allowed URL list first, then enable this configuration to enforce the
check.

#### IMPORTANT
Set the allowed URL list before you enable
`confluent.cluster.link.token.handler.url.validation.enable`.
If you enable this configuration without first adding the
metadata server URLs to the allowed URL list, the broker
denies the cluster link’s connections to those URLs.

The cluster link token handler validator accepts only HTTP and HTTPS
endpoints. It rejects file-based token locations and any other
non-HTTP(S) scheme, even though the allowed URL list supports file-based
locations for other SASL/OAUTHBEARER use cases.

Set the allowed URL list as a system property when you start the broker.
This is the same allow list described in
[Restrict OAuth bearer token endpoint URLs](#cluster-link-allowed-urls).
If you already configured URLs there, append the metadata server URLs to
that list instead of replacing it:

```bash
-Dorg.apache.kafka.sasl.oauthbearer.allowed.urls=https://auth.example.com,https://mds.example.com
```

`confluent.cluster.link.token.handler.url.validation.enable` is a dynamic
broker configuration. Set it with `kafka-configs` without restarting the
broker:

```bash
$CONFLUENT_HOME/bin/kafka-configs --bootstrap-server localhost:9092 --entity-type brokers --entity-default --alter --add-config confluent.cluster.link.token.handler.url.validation.enable=true
```

## Mutual TLS (mTLS)

Cluster Linking can use mutual TLS (mTLS) for some, but not all, data exchanges:

- Confluent Cloud to Confluent Cloud does not use mTLS; it uses TLS and SASL as described in [Cluster Linking on Confluent Cloud](/cloud/current/multi-cloud/cluster-linking/security-cloud.html).
- Data coming into Confluent Cloud from OSS Kafka can be configured to use mTLS
- Data coming into Confluent Cloud from Confluent Platform can be configured to use either mTLS or source-initiated
  cluster links with TLS+SASL. mTLS can be used as the Confluent Platform side of a source-initiated cluster link.

Follow these guidelines when configuring mTLS for Confluent Platform to Confluent Platform cluster links:

- To learn more about PEM files and possible configurations, see the
  [Apache Kafka KIP that introduced PEM file support](https://cwiki.apache.org/confluence/display/KAFKA/KIP-651+-+Support+PEM+format+for+SSL+certificates+and+private+key).
- If you reference a `keystore`/`truststore` directly (for example, `keystore.jks`),
  the same files must be available in the same location on each of the brokers
- Only PKCS #8 keys are supported for the private key specified in `ssl.keystore.key`. If the key is encrypted, the key password must be specified using `ssl.key.password`.
- If the files cannot be copied to the brokers directly, be sure to list them in PEM format.
  - If using `kafka-cluster-links` , add `\n` to indicate a new lines in the encrypted certificates.
    The example below is modified; it doesn’t include full keys, but should be enough to show the use of `\n`:
    ```none
    security.protocol=SSL
    ssl.keystore.type=PEM
    ssl.truststore.type=PEM
    ssl.key.password=clientpass
    ssl.endpoint.identification.algorithm=
    ssl.keystore.key=-----BEGIN ENCRYPTED PRIVATE KEY-----\nMIIE6TAbBgkqhkiG9w0BBQMwDgQI4d+aGvbJsIcCAggABIIEyB+qVnCcV9BbBz97\njmG1Xm6p2HIbO4tgMTld/Kmfx/4RHECdnhjiP3plH4yxZnaHBbiRHlno1d2xL93n\n4pqQkfhfJhBIqzoO3A==\n-----END ENCRYPTED PRIVATE KEY-----
    ssl.keystore.certificate.chain=-----BEGIN CERTIFICATE-----\nMIIETjCCAzagAwIBAgIUX64Yeyvwae61we2rC4gniQUcrdIwDQYJKoZIhvcNAQEL\nBQAwZTEYMBYGA1UEAwwPVGVzdENvbmZsdWVudENBMQwwCgYDVQQLDANFQUExEDAO\nrPhy0+XGDZbC4PWYi3FogFTKKjKzjtO3ZP5nXt6zvF9/nCn8RpljKJH4brIIlhPM\n3Us=\n-----END CERTIFICATE-----
    ssl.truststore.certificates=-----BEGIN CERTIFICATE-----\nMIIDRjCCAi4CCQCW7jXMNbE1XzANBgkqhkiG9w0BAQsFADBlMRgwFgYDVQQDDA9U\nZXN0Q29uZmx1ZW50Q0ExDDAKBgNVBAsMA0VBQTEQMA4GA1UECgwHRnJlZW1hbjEP\n4Q/DVCHHUvJwHxd/5Bc08s56FYHFetoB1d4=\n-----END CERTIFICATE----
    ```
  - If using `confluent kafka link`, the command doesn’t interpret `\n` in the same way. Instead, replace literal `\n` with a single space.
    ```none
    ssl.endpoint.identification.algorithm=
    security.protocol=SSL
    ssl.keystore.type=PEM
    ssl.keystore.certificate.chain=-----BEGIN CERTIFICATE-----\
    MIIDeTCCAmGgAwIBAgIJAKU/BVTP/yyUMA0GCSqGSIb3DQEBCwUAMHMxCzAJBgNV\
    BAYTAlVTMQswCQYDVQQIDAJNVDEQMA4GA1UEBwwHQm96ZW1hbjEPMA0GA1UECgwG\
    RmlndXJlMQ8wDQYDVQQLDAZGaWd1cmUxIzAhBgNVBAMMGmthZmthLXJvb3QudGVz\
    ...
    k1PdbjufUewo7KQf8nde1IefAbSARG6Fu4oY2g4=\
    -----END CERTIFICATE-----
    ssl.keystore.key=-----BEGIN ENCRYPTED PRIVATE KEY-----\
    MIIE5DAcBgoqhkiG9w0BDAEBMA4ECIYZ0EPRimP2AgIIAASCBMJLvS+Mtm9HzU4O\
    fu3EuTu7LyjH4KEyzSYsuKnhYMDbFxHD1V/dSIr3N8ZqDjz5Xr3TvsN3pVwa5BFh\
    Tv8NhhzEWB6jQtf7xo0cGPlL2VVO95D2aAvLBTQegWxqOXrhGIzqffyw/59uyJi9\
    ...
    WsrgdacPOGc4pC+bFXUOySXUBrvI47rPLn3tHHWnGfKEOKA0zMRPHjMUWmooKhKl\
    VKz1+zPeZ1s=\
    -----END ENCRYPTED PRIVATE KEY-----
    ssl.key.password=<REDACTED>
    ssl.truststore.type=PEM
    ssl.truststore.certificates=-----BEGIN CERTIFICATE-----\
    MIIDYjCCAkoCCQCw8cwWkAxRaDANBgkqhkiG9w0BAQsFADBzMQswCQYDVQQGEwJV\
    ...
    CIi0IXotll17POA+hgHeWBxhKg2ULmPxR7UgHgCcnajTNMmbbaBfOSa6xe1qOJvU\
    CzFNJ0Yz\
    -----END CERTIFICATE-----
    ```

    Alternatively, you can use an escaped newline; that is, backslash (\\), and then an actual newline.

<a id="cluster-link-acls"></a>

## Authorization (ACLs)

In deployments where ACLs are enabled, additional ACLs must be added in both the
source and destination clusters. For details on creating ACLs, see [Use Access Control Lists (ACLs) for Authorization in Confluent Platform](../../security/authorization/acls/overview.md#kafka-authorization).
For a full list of associated operations, resources, and APIs, see the subtopic, [ACL operations and allowed APIs](../../security/authorization/acls/overview.md#acl-operations).

Here are some important points to keep in mind:

- If you are using ACL migration in your pre-7.1.0 deployments, you should disable it by setting `acl.sync.enable=false` on your cluster links, or preferably, upgrade to a later version.
- ACL syncing is handled by more efficient and less error-prone methods. You may notice that the cluster link occasionally appends data to ACLs on the primary cluster.
  When Cluster Linking is active with `acl.sync` enabled, a link ID list is added to ACLs to establish provenance within Cluster Linking.
  This provenance assists the cluster link when making decisions about replicating or removing ACLs between clusters; a more reliable method
  than that used in earlier versions. This information is set when cluster link’s ACL sync creates the ACL on the cluster. You can view all ACLs
  created by a cluster link via the following command:
  ```bash
  kafka-acls --bootstrap-server <destination cluster > --list --link-id <link id>
  ```
- To successfully execute the ACL migrations defined in your sync filters, the cluster link principal requires authorization on both the source and destination clusters.
  Review the required permissions below:

  | Task                      | Operation   | Resource         | RBAC Role                 | API          |
  |---------------------------|-------------|------------------|---------------------------|--------------|
  | Read Source ACLs          | DESCRIBE    | Cluster (Source) | Operator (on Cluster)     | DescribeAcls |
  | Write / Sync ACLs to Dest | ALTER       | Cluster (Dest)   | Operator (on Cluster)     | CreateAcls   |
  |                           |             |                  | ClusterAdmin (on Cluster) | DeleteAcls   |

### ACLs for user-issued cluster link and mirror commands

The following destination cluster ACLs are available for the user executing cluster link or mirror commands.

| Task                                     | Operation     | Resource   | RBAC Role                          | API                              |
|------------------------------------------|---------------|------------|------------------------------------|----------------------------------|
| Create/Delete Cluster Link               | ALTER         | Cluster    | ClusterAdmin (on Cluster)          | CreateClusterLinks               |
|                                          | ALTER         | Cluster    | ResourceOwner (on Cluster)         | DeleteClusterLinks               |
|                                          | ALTER         | Cluster    | SystemAdmin                        |                                  |
| Alter Link Configs                       | ALTER_CONFIGS | Cluster    | ClusterAdmin (on Cluster)          | AlterConfigs                     |
|                                          | ALTER_CONFIGS | Cluster    | ResourceOwner (on Cluster)         |                                  |
| List/Describe Links                      | DESCRIBE      | Cluster    | DeveloperRead                      | ListClusterLinks                 |
|                                          | DESCRIBE      | Cluster    | Operator                           |                                  |
|                                          | DESCRIBE      | Cluster    | ClusterAdmin                       |                                  |
| Create Mirror Topic                      | ALTER         | Cluster    | DeveloperManage (on target Topics) | CreateTopics                     |
|                                          | CREATE        | Topic      | ClusterAdmin                       |                                  |
| Manage Mirror Topics <sup>[1](#f1)</sup> | ALTER         | Topic      | DeveloperManage (on target Topics) | AlterMirrors <sup>[1](#f1)</sup> |
|                                          | ALTER         | Topic      | ClusterAdmin                       |                                  |
| Sync Client ACLs via Link                | ALTER         | Cluster    | Operator (on Cluster)              | CreateAcls                       |
|                                          | ALTER         | Cluster    | ClusterAdmin                       | DeleteAcls                       |
* <a id='f1'>**[1]**</a> Regarding AlterMirrors, the commands `reverse-and-start` and `reverse-and-pause` also require Alter:Cluster on both the local and remote cluster. `truncate-and-restore` requires Alter:Cluster on the local cluster. To learn more about `reverse-and-start`, `reverse-and-pause`, and `truncate-and-restore`, see [Reverse a source and mirror topic](mirror-topics-cp.md#reverse-source-and-mirror-topic) and [Convert a mirror topic to a normal topic](mirror-topics-cp.md#convert-mirror-topic-to-normal-topic).

#### ACLs required to list and describe mirror topics

The `list mirrors` command lists the mirror topics on a cluster and/or a cluster link.
You can call this command through the API with ``REST`: `GET /kafka/v3/clusters/<cluster-id>/links/<link-name>/mirrors``
or from the Confluent Platform CLI with `kafka-mirrors --list`.

The `describe mirror` command gives information about a specific mirror topic.
You can call this command through the API with ``REST`: `GET /kafka/v3/clusters/<cluster-id>/links/<link-name>/mirrors/<mirror-topic>``
or from the Confluent Platform CLI with `kafka-mirrors --describe --topics <mirror-topic>`.

To list and describe mirror topics, you need one of the following ACLs:

- The DESCRIBE ACL on a Cluster resource, which provides access to both list and describe for all topics on the cluster.
- The DESCRIBE ACL on specific Topic(s), which provides access to both list and describe for the specified topics only.

### ACLs for brokers on destination cluster

If offset migration is not enabled, no additional permissions are required for the brokers.

If offset migration is enabled, additional ACLs are required for the brokers in the destination cluster.

| Task                          | Operation     | Resource   | RBAC Role                          | API             |
|-------------------------------|---------------|------------|------------------------------------|-----------------|
| Create/Manage Mirror Topics   | CREATE        | Topic      | DeveloperManage (on target Topics) | CreateTopics    |
|                               | ALTER         | Topic      | ClusterAdmin (on Cluster)          |                 |
| Write Replicated Data         | WRITE         | Topic      | DeveloperWrite (on target Topics)  | Produce         |
|                               | ALTER_CONFIGS | Topic      |                                    | AlterConfigs    |
| Sync Client ACLs (Provenance) | ALTER         | Cluster    | Operator (on Cluster)              | CreateAcls      |
|                               | ALTER         | Cluster    | ClusterAdmin (on Cluster)          | DeleteAcls      |
| Commit Migrated Offsets       | READ          | Group      | DeveloperWrite (on target Groups)  | OffsetCommit    |
|                               |               |            | Operator (on Cluster)              | TxnOffsetCommit |

<a id="cluster-link-acls-for-link-on-source"></a>

### ACLs for link on source cluster

Source cluster ACLs are required for the link credential as follows.

| Task                      | Operation        | Resource   | RBAC Role                          | API                                 |
|---------------------------|------------------|------------|------------------------------------|-------------------------------------|
| Mirroring Data            | READ             | Topic      | DeveloperRead (on target Topics)   | Fetch                               |
|                           | DESCRIBE_CONFIGS | Topic      | DeveloperManage (on target Topics) | DescribeConfigs <sup>[2](#f2)</sup> |
| Consumer Offset Migration | DESCRIBE         | Topic      | DeveloperRead (on target Topics)   | ListOffsets                         |
|                           | DESCRIBE         | Group      | DeveloperRead (on target Groups)   | OffsetFetch                         |
|                           | DESCRIBE_CONFIGS | Cluster    | Operator (on Cluster)              | DescribeConfigs <sup>[2](#f2)</sup> |
| ACL Synchronization       | DESCRIBE         | Cluster    | Operator (on Cluster)              | DescribeAcls                        |
| Source-Initiated Links    | ALTER            | Cluster    | ClusterAdmin (on Cluster)          | CreateClusterLinks                  |
* <a id='f2'>**[2]**</a> If `link.mode` is bi-directional.

#### IMPORTANT
**Permission validation timing**

Permissions are validated once during cluster link establishment and again upon
reconnection. However, permissions are not periodically validated for an active
cluster link. If permissions are revoked while a link is actively mirroring, the
link will continue operating until it reconnects or encounters an authorization error.

<a id="cluster-link-acls-for-source-initiated-links"></a>

### ACLs for source-initiated links

#### ACLs required by the cluster link principal on the Destination Cluster

For a source-initiated cluster link, the cluster link’s principal requires the
`Cluster: Alter` ACL on the Destination cluster.

`Cluster: Alter` is required on the Destination cluster to ensure that the user who
created the cluster link is authorized to create a cluster link to this destination cluster.

#### ACLs required by the cluster link principal on the Source Cluster

The cluster link’s source cluster principal requires the `Cluster: Alter` ACL
in addition to the normal ACLs required by a cluster link on a source cluster.

`Cluster: Alter` is required on the Source cluster because the cluster link needs to
create reverse connections on the cluster, which is considered a privileged action.

If both [auto-create mirror topics](/platform/current/multi-dc-deployments/cluster-linking/mirror-topics-cp.html#auto-create-mirror-topics)
and [prefixing](/platform/current/multi-dc-deployments/cluster-linking/mirror-topics-cp.html#prefix-mirror-topics-and-consumer-group-names)
are enabled, `Describe:Cluster` is required on the source cluster.

#### ACLs required by the user principal on the Source Cluster

To perform actions to a source-initiated cluster link on the source cluster, the
user’s principal requires the same ACLs as if they were creating a
destination-initiated cluster link on that cluster:

| Task                         | Operation     | Resource   | RBAC Role                 | API                |
|------------------------------|---------------|------------|---------------------------|--------------------|
| Create Outbound Cluster Link | ALTER         | Cluster    | ClusterAdmin (on Cluster) | CreateClusterLinks |
|                              | ALTER         | Cluster    | SystemAdmin               |                    |
| Delete Outbound Cluster Link | ALTER         | Cluster    | ClusterAdmin (on Cluster) | DeleteClusterLinks |
|                              | ALTER         | Cluster    | SystemAdmin               |                    |
| Alter Outbound Link Configs  | ALTER_CONFIGS | Cluster    | ClusterAdmin (on Cluster) | AlterConfigs       |
| List/Describe Outbound Links | DESCRIBE      | Cluster    | DeveloperRead             | ListClusterLinks   |
|                              | DESCRIBE      | Cluster    | Operator                  |                    |
|                              | DESCRIBE      | Cluster    | ClusterAdmin              |                    |

Keep in mind the following points for this use case:

- In a standard link setup, an administrator runs commands against the Destination cluster, which then reaches out to pull from the Source.
  However, if your network configuration forces you to run commands against the Source cluster to push out an outbound connection link to a Destination,
  the human operator or admin tool needs specific permissions directly on the Source cluster itself.
- if they are pushing an outbound link from a source cluster that utilizes the Metadata Service (MDS) for RBAC, the user principal running the CLI commands
  needs the ClusterAdmin role explicitly bound to that local source cluster. Otherwise, standard Kafka ACLs (ALTER, ALTER_CONFIGS, or DESCRIBE on the local Cluster resource) must be used.

<a id="cluster-link-acls-migrate"></a>

## Use ACL filters to migrate or sync ACLs from Source to Destination cluster

Cluster Linking allows you to connect clusters together and replicate data across
Kafka clusters using the brokers for topic sharing, cluster migration, and hybrid
architectures. For all scenarios, Cluster Linking provides the option to migrate
the ACLs on the source to the destination. You can migrate ACLs across the
clusters using the link, with fine-grained control as to which ACLs to migrate.

When you create the cluster link, you can specify which ACLs to migrate to the
destination (mirror) by resource, principal, and host. You can migrate any
number of specific ACLs needed for the destination cluster, or migrate all ACLs.

Migrating consumer groups: You can also [migrate consumer groups from the source to the destination cluster](commands.md#cluster-link-migrate-consumer-groups).

### Prerequisites

- **Authorizations**
  - An authorizer, such as the [Configure Confluent Server Authorizer in Confluent Platform](../../security/csa-introduction.md#confluent-server-authorizer) or [AclAuthorizer](https://kafka.apache.org/documentation/#security_authz) from Apache Kafka®, must be configured.
  - To sync centralized ACLs, set this configuration: `confluent.authorizer.access.rule.providers=CONFLUENT`. To learn more, see [Configure the Confluent Server Authorizer](../../kafka/configure-mds/index.md#csa-mds-settings-1).
  - You must have ACL Authorization configured on the clusters and cluster link, as described in [Use Access Control Lists (ACLs) for Authorization in Confluent Platform](../../security/authorization/acls/overview.md#kafka-authorization) and [Authorization (ACLs)](#cluster-link-acls).
  - You must have the appropriate authorizations: DESCRIBE Cluster ACLs (`DescribeAcls` API) on the source cluster
    and ALTER Cluster ACLs (`CreateAcls`/`DeleteAcls` APIs) on the destination cluster.
- **ACL sync does not support prefixing**
  - ACL syncing cannot be used in cluster links that are configured with a `cluster.link.prefix`.
    For more details, see [Limitations on prefixing](mirror-topics-cp.md#limitations-on-prefixing-in-mirror-topics).

### Prefixing and pattern matching behavior

The behavior of [ResourcePatternFilter](https://docs.confluent.io/platform/current/clients/javadocs/javadoc/org/apache/kafka/common/resource/ResourcePatternFilter.html) matching in ACL is as follows:

- If you use `PatternType.LITERAL` or `PatternType.PREFIXED` in the resourceFilter, it will only try to match ACLs with that same pattern type and an exact match for the resource name.
- If you use `PatternType.ANY`, it will match as long as the resource name matches, or if the resource name is null.
- If you use `PatternType.MATCH`, which performs pattern-style matching, it behaves as described in [MATCH](https://javadoc.io/doc/org.apache.kafka/kafka-clients/latest/org/apache/kafka/common/resource/PatternType.html#MATCH) in the Kafka documentation.

### Configurations for periodic ACL migration

You can specify ACL migration for a cluster link.

The recommended method is to define the ACLs in a file, `acl.filters.json`,
and pass the file name as an argument to the `--acl-filters-json-file` flag on
[CLI commands](commands.md#bootstrap-cluster-links).

You can configure ACLs at the time you [create the cluster link](commands.md#cluster-link-create-command),
or as an [update to an existing configuration](commands.md#cluster-link-alter-command).

`acl.sync.ms` is a property that specifies how often to refresh ACLs. The default is 5 seconds (5000 ms).

An additional required parameter is `acl.filters.json`. This is a JSON string
that gives the list of ACLs to migrate. The default is `””`. You can populate
this string by passing a JSON file that specifies the ACLs.

Here is an example of setting up a cluster link with ACL migration when you create the cluster link.
Note that in this example, the link configuration (including ACL migration properties) are defined in
a file, `link-configs.txt` rather than specified directly on the command line.

```bash
.bin/kafka-cluster-links --bootstrap-server destinationCluster:9092 \
 --command-config dest-credentials.txt --create --link example-link \
 --config-file link-configs.txt --acl-filters-json-file acls.filters.json --consumer-group-filters-json-file consumer-groups.json
```

You can then pass the configurations as an argument to `--config-file` in `link-configs.txt`. Here is an example of what you might have in `link-configs.txt`:

```properties
bootstrap.servers=sourceCluster:9092

ssl.endpoint.identification.algorithm=https

security.protocol=SASL_SSL

sasl.mechanism=PLAIN

sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="sourceClusterUserName" password="sourceClusterPassword";

acl.sync.enable=true

acl.sync.ms=1000

consumer.offset.sync.enable=true

consumer.offset.sync.ms=1000
```

The following sections provide examples of how you might define the actual ACLs in `acls.filters.json`.

### Examples

The following examples show how to configure the JSON for various types of ACL migration,
with granularity on a topic, a resource, or a mixed set.

#### Migrate all ACLs from source to destination cluster

To migrate all ACLs from the source to the destination cluster, provide the following for `acl.filters.json`.

```json
{
   "aclFilters": [
     {
       "resourceFilter": {
         "resourceType": "any",
         "patternType": "any"
       },
       "accessFilter": {
         "operation": "any",
         "permissionType": "any"
       }
     }
   ]
 }
```

Each field in the JSON has the following options:

* `resourceType`: Can either be `any` or one of those specified in [ACL resource types](../../security/authorization/acls/overview.md#acl-resources) in [Use Access Control Lists (ACLs) for Authorization in Confluent Platform](../../security/authorization/acls/overview.md#kafka-authorization).
* `patternType`: Can either be `any`, `literal`, `prefixed`, or `match`
* `name`: Name of resource. If not set (left empty), will default to match all the names of the specified `resourceType`. Asterisks do not act like wildcards in this context, but as explicit values. This means that if you set `name` to `*`, it will match resources named `*`.
* `principal`: Name of principal. If not set (left empty), will default to match all principals with the specified `operation` and `permissionType`. Asterisks do not act like wildcards in this context, but as explicit values. This means that if you set `principal` to `*`, it will match principals named `*`.
* `permissionType`: Can be `any`, `allow`, or `deny`
* `host`: Host for which operations can be coming from. If not set (left empty), will default to match all hosts with the specified `operation` and `permissionType`. Asterisks do not act like wildcards in this context, but as explicit values. This means that if you set `host` to `*`, it will match hosts named `*`.

<a id="cluster-link-acl-sync-limitations-prefixing"></a>

##### Limitations on filtering ACLs

In Cluster Linking, ACL sync does not support filtering resources using a prefix pattern. For example, using `"patternType": "PREFIXED"`
to sync all ACLs for topics starting with `orders-` is not supported.

Instead, you can use the following alternatives to search for prefixed resources using ACL filters:

- Configure ACL filters using the `LITERAL` pattern type, and define the exact name of each of the desired prefixed resources in each filter.
- Configure ACL filters that each define a specific `principal` with access to the desired prefixed resources.
  - This alternative is best when ACLs for the desired prefixed resources are all assigned to the same principals.

    For example, if you want to sync read ACLs for topics `orders-us`, `orders-eu`,
    `orders-apac`, and `orders-latam` that are all assigned to the same principal,
    you would define a separate filter entry for each topic:
    ```properties
    acl.filters={
      "aclFilters": [
        {
          "resourceFilter": {
            "resourceType": "topic",
            "patternType": "LITERAL",
            "name": "orders-us"
          },
          "accessFilter": {
            "principal": "User:sa-12345",
            "operation": "read",
            "permissionType": "allow"
          }
        },
        {
          "resourceFilter": {
            "resourceType": "topic",
            "patternType": "LITERAL",
            "name": "orders-eu"
          },
          "accessFilter": {
            "principal": "User:sa-12345",
            "operation": "read",
            "permissionType": "allow"
          }
        },
        {
          "resourceFilter": {
            "resourceType": "topic",
            "patternType": "LITERAL",
            "name": "orders-apac"
          },
          "accessFilter": {
            "principal": "User:sa-12345",
            "operation": "read",
            "permissionType": "allow"
          }
        },
        {
          "resourceFilter": {
            "resourceType": "topic",
            "patternType": "LITERAL",
            "name": "orders-latam"
          },
          "accessFilter": {
            "principal": "User:sa-12345",
            "operation": "read",
            "permissionType": "allow"
          }
        }
      ]
    }
    ```

#### Sync all ACLs specific to a topic

To sync all ACLs specific to a single topic, in this case topic `pineapple`,
provide the following configurations in `acl.filters`:

```properties
acl.filters={                     \
  "aclFilters": [                 \
    {                             \
      "resourceFilter": {         \
        "resourceType": "topic",  \
        "patternType": "literal", \
        "name": "pineapple”       \
      },                          \
      "accessFilter": {           \
        "operation": "any",       \
        "permissionType": "any"   \
      }                           \
    }                             \
   ]                              \
 }
```

#### Sync ACLs specific to a principal or permission

To sync all ACLs specific to a service account with ID 12345 and also all ACLs
that deny access to the cluster, provide the following configurations in `acl.filters`:

```properties
acl.filters={                         \
  "aclFilters": [                     \
    {                                 \
      "resourceFilter": {             \
        "resourceType": "any",        \
        "patternType": "any"          \
      },                              \
      "accessFilter": {               \
        “principal”: “User:sa-12345”, \
        "operation": "any",           \
        "permissionType": "any"       \
      }                               \
    },                                \
    {                                 \
      "resourceFilter": {             \
        "resourceType": "any",        \
        "patternType": "any"          \
      },                              \
      "accessFilter": {               \
        "operation": "any",           \
        "permissionType": "deny"      \
      }                               \
     }                                \
   ]                                  \
 }
```

#### Migrate all ACLs specific to a topic

To migrate all ACLs specific to a single topic (in this case topic `pineapple`), provide the following for `acl.filters.json`.

```json
{
  "aclFilters": [
    {
      "resourceFilter": {
        "resourceType": "topic",
        "patternType": "literal",
        "name": "pineapple"
      },
      "accessFilter": {
        "operation": "any",
        "permissionType": "any"
      }
    }
  ]
}
```

#### Migrate ACLs specific to a principal

To migrate all ACLs specific to a principal, Alice, provide the following for `acl.filters.json`.

```json
{
  "aclFilters": [
    {
      "resourceFilter": {
        "resourceType": "any",
        "patternType": "any"
      },
      "accessFilter": {
        "principal": "User:Alice",
        "operation": "any",
        "permissionType": "any"
      }
    }
  ]
}
```

#### Migrate a mix of ACLs

You can also do more complex ACL migrations with a specific ACL JSON. Assume you only want to migrate the following ACLs on the source cluster:

1. Alice is allowed to write to any topic from host `goodHostIPAddress`.
2. Alice is allowed to read from any topic from host `goodHostIPAddress`.
3. Alice is allowed to create any topic from host `goodHostIPAddress`.
4. Bob is allowed to read from any topic prefixed with `coffee` from host `anotherGoodHostIPAddress`.
5. Bob is allowed to create any topic prefixed with `breakfast.bar` from `anotherGoodHostIPAddress`.
6. Bob is allowed to create any topic with the name `breakfast.bar` from `anotherGoodHostIPAddress`.
7. Mallory is denied all operations from any topic from any host.
8. Admin is allowed all operations on any topic from host `adminHostIPAddress`.
9. Trent is allowed to read from any topic from host `greatHostIPAddress`.
10. Eve is denied to read from any topic from any host.

To migrate the ACLs specified above, you would specify the following ACL filters in the JSON file for `acl.filters.json`.

```json
{
  "aclFilters": [
  // filter for 1
    {
      "resourceFilter": {
        "resourceType": "topic",
        "patternType": "any"
      },
      "accessFilter": {
        "principal": "User:Alice",
        "host": "goodHostIPAddress",
        "operation": "read",
        "permissionType": "allow"
      }
    },
    // filter for 2
    {
      "resourceFilter": {
        "resourceType": "topic",
        "patternType": "any"
      },
      "accessFilter": {
        "principal": "User:Alice",
        "host": "goodHostIPAddress",
        "operation": "write",
        "permissionType": "allow"
      }
    },
    // filter for 3
    {
      "resourceFilter": {
        "resourceType": "topic",
        "patternType": "any"
      },
      "accessFilter": {
        "principal": "User:Alice",
        "host": "goodHostIPAddress",
        "operation": "create",
        "permissionType": "allow"
      }
    },
    // filter for 4
   {
      "resourceFilter": {
        "resourceType": "topic",
        "name": "coffee",
        "patternType": "prefixed"
      },
      "accessFilter": {
        "principal": "User:Bob",
        "host": "anotherGoodHostIPAddress",
        "operation": "read",
        "permissionType": "allow"
      }
    },
    // filter for 5 and 6
   {
      "resourceFilter": {
        "resourceType": "topic",
        "name": "breakfast.bar",
        // match serves as a catch-all for all the names of
        // a topic this principal is authorized to access
        "patternType": "match"
      },
      "accessFilter": {
        "principal": "User:Bob",
        "host": "anotherGoodHostIPAddress",
        "operation": "create",
        "permissionType": "allow"
      }
    },
    // filter for 7
   {
      "resourceFilter": {
        "resourceType": "any",
        "patternType": "any"
      },
      "accessFilter": {
        "principal": "User:Mallory",
        "operation": "any",
        "permissionType": "deny"
      }
    },
    // filter for 8
   {
      "resourceFilter": {
        "resourceType": "any",
        "patternType": "any"
      },
      "accessFilter": {
        "principal": "User:Admin",
        "host":"adminHost",
        "operation": "any",
        "permissionType": "allow"
      }
    },
    // filter for 9
   {
      "resourceFilter": {
        "resourceType": "topic",
        "patternType": "any"
      },
      "accessFilter": {
        "principal": "User:Trent",
        "host":"greatHost",
        "operation": "read",
        "permissionType": "allow"
      }
    },
    // filter for 10
   {
      "resourceFilter": {
        "resourceType": "topic",
        "patternType": "any"
      },
      "accessFilter": {
        "principal": "User:Eve",
        "operation": "read",
        "permissionType": "deny"
      }
    }
  ]
}
```

## Related content

- [Use TLS Authentication in Confluent Platform](../../security/authentication/mutual-tls/overview.md#kafka-ssl-authentication)
- [Use Access Control Lists (ACLs) for Authorization in Confluent Platform](../../security/authorization/acls/overview.md#kafka-authorization)
- [ACL operations and allowed APIs](../../security/authorization/acls/overview.md#acl-operations)
- [Configure GSSAPI in Confluent Platform clusters](../../security/authentication/sasl/gssapi/overview.md#kafka-sasl-auth-gssapi)