Cluster Linking Security

Important

This feature is available as a preview feature. A preview feature is a component of Confluent Platform that is being introduced to gain early feedback from developers. This feature can be used for evaluation and non-production testing purposes or to provide feedback to Confluent.

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.

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 SSL and SASL/GSSAPI configuration options for connections to the source cluster in this scenario.

For details on creating SSL key and trust stores, see Encryption and Authentication with SSL. For details on SASL/GSSAPI, see Configuring GSSAPI. Brokers must be configured with password.encoder.secret for encrypting sensitive link configurations when security is enabled.

To configure cluster links to use other SASL mechanisms, include client-side security configurations for that mechanism. See Authentication with SASL using JAAS for other supported mechanisms. To use two-way SSL authentication with SSL as the security protocol, a key store should also be configured for the link. See Encryption and Authentication with SSL for details.

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 Authorization using ACLs. For a full list of associated operations, resources, and APIs, see the subtopic, Operations.

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.

Operation Resource API
READ Topic APIs used for consumer offset migration
READ Group APIs used for consumer offset migration
ALTER Topic (Mirror) AlterTopicMirrors

Migrating 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.

Tip

You can also migrate consumer groups from the source to the destination cluster.

Prerequisites

  • An authorizer, such as the Overview or AclAuthorizer from Apache Kafka®, must be configured.
  • You must have ACL Authorization configured on the clusters and cluster link, as described in Authorization using ACLs and Authorization (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.

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 kafka-cluster-links command.

You can configure ACLs at the time you create the cluster link, or as an update to an existing configuration.

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.

.bin/kafka-cluster-links --bootstrap-server destinationCluster:9092 \
 --command-config dest-credentials.txt --create --link-name 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:

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.

Tip

  • You must use IP addresses, not host names.
  • If you do not specify a host, the ACLs will default to wildcard * to provide access from all hosts.

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.

{
  "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 Resources in Authorization using ACLs.
  • patternType: Can either be any, literal, prefixed, or match
  • name: Name of resource, if left empty will default to the * wildcard which will match all names of the specified resourceType
  • principal: Name of principal, if left empty will default to the * wildcard which will match all principals with the specified operation and permissionType
  • operation: Can either be any or one of those specified in Operations under the “Operation” column
  • permissionType: Can be any, allow, or deny
  • host: Host for which operations can be coming from, if left empty this defaults to the * wildcard which will match all hosts with the specified operation and permissionType

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.

{
  "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.

{
  "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.

{
  "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"
      }
    }
  ]
}