11.6 Authorization

"Kafka brokers manage access control using a customizable authorizer... When a request is processed, the broker verifies that the principal associated with the connection is authorized to perform that request."

authorizer.class.name=kafka.security.authorizer.AclAuthorizer

SimpleAclAuthorizer

"AclAuthorizer was introduced in Apache Kafka 2.3. Older versions from 0.9.0.0 onward had kafka.security.auth.SimpleAclAuthorizer, which has been DEPRECATED but is still supported."

6.1 AclAuthorizer

"ACLs are stored in ZooKeeper and CACHED IN MEMORY by EVERY broker to enable high-performance lookup. ACLs are loaded into the cache when the broker starts up, and the cache is kept up-to-date using notifications based on a ZOOKEEPER WATCHER."

(This is why Deny ACLs are the fastest incident response — the ZK watcher propagates them near-instantly to every broker.)

The six components of an ACL binding

Example: “User:Alice has Allow permission for Write to Prefixed Topic:customer from 192.168.0.1”.

The evaluation rule and the implicit grants

Authorized ⟺ (No matching Deny ACL) And ( ≥1 matching Allow ACL)

Implicit grants — easy to forget:

Wildcard ACLs: “ACLs with pattern type LITERAL and resource name * are used as wildcard ACLs that match All resource names of a resource type.”

Who needs what — the practical summary

The full ACL → request mapping (Table 11-1)

Managing ACLs

# ① Broker ACLs created DIRECTLY IN ZOOKEEPER — "useful to create broker ACLs
#    PRIOR TO STARTING BROKERS"
bin/kafka-acls.sh --add --cluster --operation ClusterAction \
  --authorizer-properties zookeeper.connect=localhost:2181 \
  --allow-principal User:kafka

# ② producer convenience flag; LITERAL by default
bin/kafka-acls.sh --bootstrap-server localhost:9092 \
  --command-config admin.props --add --topic customerOrders \
  --producer --allow-principal User:Alice

# ③ PREFIXED ACL: Bob may read all topics starting with "customer"
bin/kafka-acls.sh --bootstrap-server localhost:9092 \
  --command-config admin.props --add --resource-pattern-type PREFIXED \
  --topic customer --operation Read --allow-principal User:Bob

⚠️ The two dangerous shortcuts

super.users=User:Carol;User:Admin        # NOTE: SEMICOLON-separated!
allow.everyone.if.no.acl.found=true

SUPER USER SEPARATOR

"Unlike other list configurations in Kafka that are comma-separated, super.users are separated by a SEMICOLON since user principals such as distinguished names from SSL certificates OFTEN CONTAIN COMMAS."

super.users:

"Super users are granted access for all operations on all resources without any restrictions and CANNOT BE DENIED ACCESS USING Deny ACLs. If Carol's credentials are compromised, Carol must be REMOVED FROM super.users, AND BROKERS MUST BE RESTARTED to apply the changes. It is SAFER to grant specific access using ACLs to users in production systems to ensure access can be revoked easily."

allow.everyone.if.no.acl.found:

"all users are granted access to resources without any ACLs. May be useful when enabling authorization for the first time or during development, BUT IS NOT SUITABLE FOR PRODUCTION since access may be granted UNINTENTIONALLY to NEW resources. Access may also be UNEXPECTEDLY REMOVED when ACLs for a matching PREFIX or WILDCARD are added, if the condition for no.acl.found NO LONGER APPLIES."

That second failure mode is genuinely surprising: adding an ACL can revoke access for everyone else who was relying on the "no ACL found" fallback.

6.2 Customizing authorization

Example 1 — restrict certain requests to the internal listener:

public class CustomAuthorizer extends AclAuthorizer {
  private static final Set<Short> internalOps =
    Utils.mkSet(CREATE_ACLS.id, DELETE_ACLS.id);
  private static final String internalListener = "INTERNAL";

  @Override
  public List<AuthorizationResult> authorize(
          AuthorizableRequestContext context, List<Action> actions) {
    if (!context.listenerName().equals(internalListener) &&    // ①
        internalOps.contains((short) context.requestType()))
      return Collections.nCopies(actions.size(), DENIED);
    else
      return super.authorize(context, actions);                // ②
  }
}

① "Authorizers are given the request context with metadata that includes LISTENER NAMES, SECURITY PROTOCOL, REQUEST TYPES, etc., enabling custom authorizers to add or remove restrictions based on the context." ② "We REUSE functionality from the built-in Kafka authorizer using the public API."

Example 2 — group/role-based access control (RBAC) from LDAP:

class RbacAuthorizer extends AclAuthorizer {
  @volatile private var groups = Map.empty[KafkaPrincipal, Set[KafkaPrincipal]]
    .withDefaultValue(Set.empty)                    // ① from LDAP
  @volatile private var roles  = Map.empty[KafkaPrincipal, Set[KafkaPrincipal]]
    .withDefaultValue(Set.empty)                    // ② from LDAP

  override def authorize(context: AuthorizableRequestContext,
      actions: util.List[Action]): util.List[AuthorizationResult] = {
    val principals    = groups(context.principal) + context.principal   // ③
    val allPrincipals = principals.flatMap(roles) ++ principals
    val contexts      = allPrincipals.map(authorizeContext(context, _)) // ⑤
    actions.asScala.map { action =>
      val authorized = contexts.exists(                                 // ④
        super.authorize(_, List(action).asJava).get(0) == ALLOWED)
      if (authorized) ALLOWED else DENIED
    }.asJava
  }
  // authorizeContext(...) wraps the original context, swapping the principal
}

③ "We perform authorization for the user as well as for ALL the groups and roles of the user." ④ "If ANY of the contexts are authorized, we return ALLOWED. Note that this example DOESN'T SUPPORT Deny ACLs for groups or roles."

# ACLs for a GROUP principal
bin/kafka-acls.sh ... --add --topic customer --producer \
  --resource-pattern-type PREFIXED --allow-principal Group:Sales
# ACLs for a ROLE principal
bin/kafka-acls.sh ... --add --cluster --operation Alter \
  --allow-principal=Role:Operator

6.3 Authorization security considerations

• ⚠ “Since AclAuthorizer Stores ACLs in zookeeper, access to zookeeper should be restricted. Deployments without a secure ZooKeeper can implement Custom authorizers to store ACLs in a Secure external database.”
• ✓ Scaling ACL management: “Reserving Different resource prefixes for different departments enables the use of Prefixed ACLs that Minimize the number of ACLs required.” Combine with group/role-based ACLs.
• ✓ Principle of least privilege: “granting access Only to the resources necessary… and Removing ACLs when they are no longer required. ACLs should be removed Immediately when a user principal is no longer in use, for instance, When a person leaves the organization.”
• 💡 “LONG-RUNNING APPLICATIONS CAN BE CONFIGURED WITH SERVICE CREDENTIALS RATHER THAN CREDENTIALS ASSOCIATED WITH A SPECIFIC USER TO AVOID ANY DISRUPTION WHEN EMPLOYEES LEAVE THE ORGANIZATION.”
• ⚠ “Since Long-lived connections with a user principal May continue to process requests even after the user has been removed, DENY ACLs can be used to ensure that the principal is Not unintentionally granted access through ACLs with wildcard principals.”
• ⚠ “Reuse of principals must be avoided if possible to prevent access from being granted to connections using The older version of a principal.”