Status

Current state: "Under Discussion"

Discussion thread: here

JIRA: here

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

Kafka supports authorize access to resources like topics, consumer groups etc. by way of ACLs. The current supported semantic of resource name in ACL definition is either full resource name or special wildcard '*', which matches everything.

Kafka should support a way of defining bulk ACLs instead of specifying individual ACLs.
Example use cases:

• Principal “com.company.product1.client” has access to all topics that start with “com.company.product1.”.
• Principal “com.company.client1” has access to all consumer groups that start with “com.company.client1.”.

This support would greatly simplify ACL operational story in a multi-tenant environment.

Public Interfaces

• Add new field 'ResourceNameType' to Resource and ResourceFilter classes, which would define if the resource name is a literal or wildcard-suffix (ResourceNameType is an enum to support more types in the future).

• Enhance getAcls(resource) to return all matching wildcard-suffixed ACLs if 'ResourceNameType' is wildcard-suffix
  • The behavior of getAcls(resource) method today is to return ACLs matching only the resource literal.
  • authorize(...) method calls getAcls(resource) and getAcls('*') to get all the matching ACLs today.
  • Backward compatibility would be maintained because 'ResourceNameType' defaults to literal.

• Changes to command line tool class https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/admin/AclCommand.scala
  • To expose the above new API (if we go that route).
  • Expose a '--wildcard-suffix-resource' flag which is "false" by default to maintain backwards compatibility (though "true" is more user friendly going forward)
    • bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181
      --add --allow-principal User:Bob --allow-principal User:Alice --allow-host
      198.51.100.0 --allow-host 198.51.100.1 --operation Read --group my-app-* –wildcard-suffix-resource true
• Changes to AdminClient - None.
• New API keys for CreateAclsRequest / DeleteAclsRequest / DescribeAclsRequest which would have a new field in schema to distinguish literals vs wildcard-suffix resource names.
• Update the public documentation with the details of the new feature.

Proposed Changes

Solution

The proposal is to extend the concept of wildcard ACL (‘*’) to support wildcard-suffixed ACLs (‘name*’).
This means that it would be possible to create ACLs of type: User:clientA has READ access on topic orgA* from hostA,
i.e clientA has READ access to all topics that start with `orgA` from hostA.
The concept of wildcard-suffixed ACLs would be applicable only to resource names.

Storage model

Currently, ACLs are stored on ZK under path /kafka-acl/<resource-type>/<resource-name>.

For example:
ACLs for topic topicName would be stored under /kafka-acl/Topic/topicName.
ACLs for consumer group groupId would be stored under /kafka-acl/Group/groupId.

An example ACL definition looks like:

$ get /kafka-acl/Topic/topicName
{"version":1,"acls":[{"principal":"User:clientA","permissionType":"Allow","operation":"Read","host":"*"},{"principal":"User:clientA","permissionType":"Allow","operation":"Write","host":"*"},{"principal":"clientB","permissionType":"Allow","operation":"Write","host":"host1"}]}

Current supported resource names are either full resource names like topicName or a special wildcard '*'.

$ get /kafka-acl/Topic/*
{"version":1,"acls":[{"principal":"User:clientA","permissionType":"Allow","operation":"Read","host":"*"}]}
which means that clientA has read access to all topics from all hosts.

The challenge here is that some resources like consumer groups don't have any defined naming convention and can have '*' in their names.

We extend the same storage model to store wildcard suffix ACLs in a different location 'kafka-wildcard-acl'. Changes will first be stored at 'kafka-wildcard-acl-changes'.

$ get /kafka-wildcard-acl/Topic/teamA*
{"version":1,"acls":[{"principal":"User:clientA","permissionType":"Allow","operation":"Read","host":"*"}]}

ACLs write path

Write to a new location 'kafka-wildcard-acl'.

$ get /kafka-wildcard-acl/Topic/orgName*
{"version":1,"acls":[{"principal":"User:clientA","permissionType":"Allow","operation":"Read","host":"*"}]}

ACLs read path

On read path, we look for all matching ACLs when:

a) getMatchingAcls(resourceWithWildcardSuffix) is called.
b) authorize(…) is called.

Access would be allowed if there is at least one ALLOW matching acl and no DENY matching acl (current behavior is maintained). Note that the length of the prefix doesn't play any role here.

Matching algorithm

  /**
    * Returns true if two strings match, both of which might be ending in a special wildcard which matches everything.
    * Examples:
    *   matchWildcardSuffixedString("rob", "rob") => true
    *   matchWildcardSuffixedString("*", "rob") => true
    *   matchWildcardSuffixedString("ro*", "rob") => true
    *   matchWildcardSuffixedString("rob", "bob") => false
    *   matchWildcardSuffixedString("ro*", "bob") => false
    *
    *   matchWildcardSuffixedString("rob", "*") => true
    *   matchWildcardSuffixedString("rob", "ro*") => true
    *   matchWildcardSuffixedString("bob", "ro*") => false
    *
    *   matchWildcardSuffixedString("ro*", "ro*") => true
    *   matchWildcardSuffixedString("rob*", "ro*") => false
    *   matchWildcardSuffixedString("ro*", "rob*") => true
    *
    * @param valueInZk Value stored in ZK in either resource name or Acl.
    * @param input Value present in the request.
    * @return true if there is a match (including wildcard-suffix matching).
    */
  def matchWildcardSuffixedString(valueInZk: String, input: String): Boolean = {
    if (valueInZk.equals(input) || valueInZk.equals(Acl.WildCardString) || input.equals(Acl.WildCardString)) {
      // if strings are equal or either of acl or input is a wildcard
      true
    } else if (valueInZk.endsWith(Acl.WildCardString)) {
      val aclPrefix = valueInZk.substring(0, valueInZk.length - Acl.WildCardString.length)
      if (input.endsWith(Acl.WildCardString)) {
        // when both acl and input ends with wildcard, non-wildcard prefix of input should start with non-wildcard prefix of acl
        val inputPrefix = input.substring(0, input.length - Acl.WildCardString.length)
        inputPrefix.startsWith(aclPrefix)
      } else {
        // when acl ends with wildcard but input doesn't, then input should start with non-wildcard prefix of acl
        input.startsWith(aclPrefix)
      }
    } else {
      if (input.endsWith(Acl.WildCardString)) {
        // when input ends with wildcard but acl doesn't, then acl should start with non-wildcard prefix of input
        val inputPrefix = input.substring(0, input.length - Acl.WildCardString.length)
        valueInZk.startsWith(inputPrefix)
      } else {
        // when neither acl nor input ends with wildcard, they have to match exactly.
        valueInZk.equals(input)
      }
    }
  }

Compatibility, Deprecation, and Migration Plan

On downgrade, the wildcard ACLs will be treated as literals and hence never match anything. This means that any wildcard ACLs would be treated as if they were never added. This is fine for ALLOW ACLs, but might have security implications if DENY ACLs are ignored.

Rejected Alternatives

• Use escaping to identify wildcard-suffix vs literals (won't work on ZK). We decided to use a separate path for wildcard-suffix ACLs