ACLs and security

Active contributors: Manikumar Reddy, Mickael Maison, José Armando García Sancio, Colin Patrick McCabe, Andrew Schofield

Kafka supports authentication, authorization, and on-the-wire encryption on every listener. Authentication is via TLS (mTLS), SASL (PLAIN, SCRAM-256/512, OAUTHBEARER, GSSAPI/Kerberos), or delegation tokens. Authorization is performed by a pluggable Authorizer SPI; the default implementation reads ACL records straight out of the metadata log.

Listeners and protocols

Each broker accepts connections on one or more listeners, each configured with a name, host:port, and a security protocol:

Security protocol	Authentication	Encryption
PLAINTEXT	none	none
SSL	none or mutual TLS	TLS
SASL_PLAINTEXT	SASL (PLAIN, SCRAM, GSSAPI, OAUTH)	none
SASL_SSL	SASL on top of TLS	TLS

Configured via listeners, listener.security.protocol.map, inter.broker.listener.name, controller.listener.names, advertised.listeners. See core/src/main/scala/kafka/network/SocketServer.scala and clients/src/main/java/org/apache/kafka/common/security/.

Authentication mechanisms

Mechanism	Source root	Notes
TLS / mTLS	clients/.../common/security/ssl/	SslEngineFactory, SslChannelBuilder. Mutual TLS extracts the client's principal from the cert.
SASL/PLAIN	clients/.../common/security/plain/	Username/password; configure callbacks for credential lookup.
SASL/SCRAM	clients/.../common/security/scram/	SCRAM-SHA-256 and SCRAM-SHA-512. Credentials stored in the metadata log.
SASL/OAUTHBEARER	clients/.../common/security/oauthbearer/	Pluggable token validation; OIDC integrations supplied by users.
Kerberos (GSSAPI)	clients/.../common/security/kerberos/	JAAS-based; uses standard Java GSS provider.
Delegation tokens	clients/.../common/security/token/	Short-lived tokens issued by the broker; useful for distributed jobs.

The SaslChannelBuilder chooses a mechanism per listener at handshake time; the per-mechanism SaslServer / SaslClient are loaded via the standard Java SASL provider mechanism.

The principal reported to the authorizer is built by a KafkaPrincipalBuilder (configurable via principal.builder.class). The default extracts the SASL identity, the TLS DN, or "ANONYMOUS" depending on the listener.

Authorization (Authorizer)

Authorizer is the SPI plugins implement (server-common/src/main/java/org/apache/kafka/server/authorizer/Authorizer.java):

List<AuthorizationResult> authorize(AuthorizableRequestContext, List<Action>);

Brokers consult it for every authorized RPC via AuthHelper in core/src/main/scala/kafka/server/AuthHelper.scala.

The default authorizer is StandardAuthorizer (metadata/src/main/java/org/apache/kafka/metadata/authorizer/StandardAuthorizer.java). It:

1. Subscribes to ACL records on the metadata log via MetadataLoader.
2. Indexes them in memory by ResourceType + ResourceName + PrincipalType + PrincipalName + Host + Operation + PermissionType.
3. Applies the standard "any DENY wins; otherwise any ALLOW grants" decision.

Because the index is updated by MetadataDelta events, ACL changes propagate to all brokers without any explicit notification — the same path that propagates topic creates and config changes.

ACL records live in metadata/src/main/resources/common/metadata/AccessControlEntryRecord.json; admins manipulate them via the kafka-acls.sh CLI which uses the CreateAcls / DeleteAcls admin RPCs handled by AclControlManager on the controller.

Resource model

ACLs apply to one of these ResourceTypes:

• Topic — produce, consume, describe, alter, delete, ...
• Group — read (consume), describe, delete.
• Cluster — cluster admin operations.
• TransactionalId — for transactional producers.
• DelegationToken — describe / alter tokens.
• User — SCRAM credential management.

Operations are listed in clients/.../common/acl/AclOperation.java. Permission types are ALLOW and DENY. Resource patterns can be LITERAL or PREFIXED.

Quotas

Quotas are independent of authorization but configured similarly. Each broker enforces:

• Producer / consumer byte-rate quotas per user / client-id (clients/.../common/quota/).
• Request-rate quotas to limit broker CPU per principal.
• Replication quotas during reassignments.

Quota records (ClientQuotaRecord in metadata/src/main/resources/common/metadata/) live on the metadata log; the ClientQuotaControlManager on the controller updates them.

SCRAM credentials

SCRAM credentials are stored as records in the metadata log (UserScramCredentialRecord). The controller's ScramControlManager validates and persists them; brokers index them via MetadataLoader. kafka-configs.sh --alter --add-config 'SCRAM-SHA-256=[password=...]' is the user-facing path.

Delegation tokens

Delegation tokens are short-lived credentials a user can request from a broker after authenticating with another mechanism. They make it possible for distributed jobs (Streams, Connect, Spark) to ship a token to workers without sharing the user's primary credentials. Implementation: core/src/main/scala/kafka/server/DelegationTokenManager.scala, plus controller-side DelegationTokenControlManager. Token records are stored on the metadata log.

OAuth / OIDC

SASL/OAUTHBEARER supports OAuth tokens. The default OAuthBearerLoginCallbackHandler validates a JWT presented by the client. Users implementing OIDC typically supply a custom login + validator callback that talks to their identity provider; this is configured per listener via JAAS Login modules.

Broker-driven client metrics (KIP-714)

KIP-714 introduces an authenticated PushTelemetry RPC by which clients send selected metrics back to the broker. Subscriptions are managed via kafka-client-metrics.sh and gated by ACL on the Cluster:DescribeClientMetrics operation. Implementation lives in clients/.../common/telemetry/ and the broker's ClientMetricsManager.

Configuring a typical secure cluster

Quick checklist (the docs under docs/security.html are the authoritative reference):

1. Generate TLS keystores / truststores; set ssl.keystore.location etc. on every broker and listener.
2. Pick a SASL mechanism (often SCRAM-SHA-256 for ad-hoc clusters, OAUTHBEARER + OIDC for managed environments). Configure sasl.enabled.mechanisms and the JAAS file.
3. Set inter.broker.listener.name to a SASL listener — broker-to-broker traffic is authenticated.
4. Set authorizer.class.name=org.apache.kafka.metadata.authorizer.StandardAuthorizer.
5. Set super.users=User:Admin;... so that operator commands can run before any ACLs are in place.
6. Provision SCRAM credentials (kafka-configs.sh) or whatever the chosen mechanism requires.
7. Add ACLs via kafka-acls.sh.
8. Restart, verify with kafka-broker-api-versions.sh and a kafka-acls.sh --list.

Entry points for modification

• New SASL mechanism: a new package under clients/src/main/java/org/apache/kafka/common/security/ plus updates to SaslChannelBuilder. KIP required for any user-visible config or behavior.
• New authorizer: implement Authorizer; ship as a JAR; configure authorizer.class.name.
• Custom principal builder: implement KafkaPrincipalBuilder; configure principal.builder.class.
• Add an ACL operation: edit AclOperation enum + the relevant *ControlManager. KIP required.

Related pages

• Modules: clients — SASL / SSL plumbing.
• Modules: core (broker) — AuthHelper, listener wiring.
• Modules: metadata — StandardAuthorizer, ACL records.
• Reference: Configuration — security configs.