Kafka 接入 Keycloak OAUTHBEARER：IAM 服务账号认证与 ACL 授权 | IDaaS Book

场景#

• Kafka 集群已经有业务在用，安全评审要求把客户端认证从 PLAINTEXT/SCRAM 换成统一 IAM（Keycloak），不再单独维护一套 SASL 账号。
• 照着一篇几年前的文章配好 sasl.jaas.config，broker 却起不来，或者客户端能拿到 token 但一直 Authentication failed。
• 想知道 token 里的 realm 角色能不能直接当 Kafka 权限用，能省掉多少 ACL 维护工作。

这篇只覆盖 Apache Kafka 内置的 SASL/OAUTHBEARER 与 Keycloak 的组合：两边的配置形态、六个最常见失败的定位方式、验证命令和回滚顺序。基线是当前最新发布版 Kafka 4.3.1（Apache 下载镜像与归档目录里最新的 4.x 就是它）；文中标为 4.3.2 / 4.4.0 的行为来自 trunk 的 docs/getting-started/upgrade.md，这两个版本尚未发布，只当前瞻提示，上线前请以你实际版本的 release notes 复核。托管发行版（Strimzi 的 strimzi-kafka-oauth、Confluent 的 OAuthKafkaPrincipalBuilder + confluent.oauth.groups.claim.name）在本文能力之上额外补了「groups claim → principal」映射，用法不同，不要混着抄。

适用与不适用#

1. 认证链路：谁在什么时候验什么#

  sequenceDiagram
    participant C as Kafka Client（producer/consumer）
    participant K as Keycloak（realm: iam）
    participant B as Kafka Broker（listener CLIENT, SASL_SSL）

    C->>K: POST /protocol/openid-connect/token<br/>grant_type=client_credentials
    K-->>C: access_token（JWT），aud/iss/sub 由 client scope 里的 mapper 决定
    C->>B: SASL/OAUTHBEARER 握手，携带 access_token
    B->>K: GET /protocol/openid-connect/certs（启动时、未知 kid、到刷新周期）
    K-->>B: JWKS
    B->>B: 用 kid 选公钥验签
    B->>B: 校验 exp/nbf（clock.skew.seconds 容忍偏差）
    B->>B: 精确比对 iss，再精确比对 aud
    B-->>C: 认证通过，principal = 配置的 claim（默认 sub）
    C->>B: 读写请求，按 ACL 判定

这张图想说明的只有一件事：这条链路上有三个独立的决策点，分别属于 Keycloak（token 里放什么）、broker（拿什么校验、拿哪个 claim 当身份）、客户端（怎么把 URL 与凭据交给 Kafka）。三个都配对了才通；配错了，症状却长得差不多，都是 SASL 握手失败。

1.1 Keycloak 的 aud 不会自动包含你要的值#

Kafka 侧 sasl.oauthbearer.expected.audience 是精确匹配，所以 aud 必须正好有你写的那个值。Keycloak 的 access token 里 aud 由协议 mapper 决定，三条要记住的源码事实：

• Audience mapper（oidc-audience-mapper）默认只加进 access token，不加进 ID token。 源码里对 id.token.claim 显式覆盖成 false，注释就是 “Don’t include audience in ID Token by default”。所以给 Kafka 用的应该是 access token（OAuth2 里本来就该如此），别被「ID token 里没有 aud」误导。
• Audience Resolve mapper 只把「该用户拥有其客户端角色的那些 client_id」加进 aud，并且跳过 client 自己。 源码里的 setAudience 就是遍历 RoleResolveUtil.getAllResolvedClientRoles 的结果，遇到自己的 client_id 直接 continue（注释原文 “Don’t add client itself to the audience”），只有角色非空的 client 才 addAudience。服务账号在 Keycloak 侧通常没有任何客户端角色，这条 mapper 对它等于不产生 aud。指望它「自动带上所有相关 audience」就会得到空 aud。
• 如果这个 client 开了轻量级 access token，Audience mapper 的「Add to access token」会被另一个开关取代。 AbstractOIDCProtocolMapper.transformAccessToken 的判定是 shouldUseLightweightToken ? includeInLightweightAccessToken(mappingModel) : includeInAccessToken(mappingModel)：一旦处于轻量 token 模式（客户端属性 client.use.lightweight.access.token.enabled，控制台里由 client policy use-lightweight-access-token 打开），mapper 必须额外勾上 Add to lightweight access token（mapper 配置项 lightweight.claim=true），否则 aud 根本不会进 access token。症状和「压根没加 mapper」完全一样：Kafka 报 audience 不匹配，而 Keycloak 控制台里 mapper 明明在。

结论：给 Kafka 显式加一个 Audience mapper，用 included.custom.audience 写一个语义化的值（例如 kafka），而不是 included.client.audience（那是「另一个 Keycloak client 的 client_id」语义，Kafka 并不是 Keycloak 的 client）。

1.2 Kafka 有两个「必须显式给」的配置#

• allowed.urls 系统属性（4.0.0 起）：默认值是空列表，等于默认拒绝一切 URL。 Kafka 源码里 ConfigurationUtils#throwIfResourceIsNotAllowed 的注释与测试断言写得很直白：// By default, no URL is allowed。没设这个 JVM 系统属性时，客户端/ broker 在 配置解析阶段 就会抛：

  Invalid value https://kc.example.com/realms/iam/protocol/openid-connect/token for configuration
  sasl.oauthbearer.token.endpoint.url: The URL cannot be accessed due to restrictions.
  Update the system property 'org.apache.kafka.sasl.oauthbearer.allowed.urls' to allow the URL to be accessed.

  匹配是字符串精确相等（allowed.contains(configValue)），没有通配：把 token endpoint 和 JWKS 两个 URL 原样、逗号分隔列出来。客户端和 broker 进程都要设（broker 的验证器同样走这段校验）。

• expected.audience / expected.issuer：只要配了 JWKS URL 就应该给。 4.3.1 的配置文档把这两项写成 strongly recommended：不设 expected.audience 时 broker 会接受不带 aud 的 token，不设 expected.issuer 时接受任意（或缺失）iss——属于静默降级，不是安全默认值。4.4.0（trunk 的 upgrade.md 已记录，尚未发布）把这条收紧成硬约束：配了 sasl.oauthbearer.jwks.endpoint.url 却没给其中之一，broker 启动即失败。逃生开关是 4.4.0 新增的 sasl.oauthbearer.allow.unverified.audience / allow.unverified.issuer（默认 false），官方措辞是 “strongly discouraged”——它们是关闭校验，不是通配符。

1.3 principal 默认是 sub，而 Keycloak 的 sub 是 UUID#

Kafka 官方文档写的是：默认 principal.builder.class 下，OAuthBearerToken 的 principalName 成为认证后的 Principal，ACL 就按它授权。而这个 principalName 默认取自 sub claim（sasl.oauthbearer.sub.claim.name 默认值 sub）。

Keycloak 的 access token sub 是用户对象的内部 id（UUID）（TokenManager.initToken 里 token.subject(user.getId())）。走 client_credentials 时，那个「用户」是客户端对应的服务账号用户，它的：

• sub = 服务账号用户的 UUID，形如 f47ac10b-58cc-...；
• preferred_username = service-account-<client-id>（Keycloak 的服务账号用户名前缀常量就是 service-account-，官方文档也这么写）；
• azp = 你配置的 client_id。

于是「ACL 到底授权给谁」变成一道选择题：

生产上建议后者，并且把「换 principal 取值方式」当成一次需要重建 ACL 的变更，而不是随手改一行。

2. Keycloak 端最小配置#

Realm:  iam
Client: kafka-client（confidential，不配 redirect URI，不开浏览器流程）

1. 建 client：Client authentication = On、Standard flow / Implicit flow / Direct access grants 全部关闭（服务间认证不需要它们），Service accounts roles = On。后者打开后，Keycloak 才会为它创建用户名 service-account-kafka-client 的服务账号用户，client_credentials 才有主体可签。

2. 把 client secret 记下来：Clients → kafka-client → Credentials → Client secret。它要进 Kafka 客户端的 sasl.oauthbearer.client.credentials.client.secret，属于「不能进配置文件明文」的那一类，放 Secret/KMS，别进 Git。

3. 加 Audience mapper（放在 client scope 里更规范，便于多客户端复用）：

   • Client scopes → 新建 kafka（类型选 Default，或按需 Assign 给客户端）→ Mappers → Add mapper → Audience；
   • Included Custom Audience = kafka。不要用 Included Client Audience——那是「另一个 Keycloak client 的 client_id」语义，而 Kafka 不是 Keycloak 的 client；两个字段都填时源码里前者优先，后者不生效，很容易填错还以为 mapper 没发挥作用；
   • 确认 Add to access token = On（源码默认就是 On）、Add to ID token 保持 Off；
   • 如果这个 client 开了轻量级 access token，再勾上 Add to lightweight access token，否则 aud 不会出现在 access token 里（原因见 1.1 第三条）。

4. 确认 issuer / token / JWKS 三个地址：直接读 discovery 文档，不要凭经验拼：

   curl -s https://kc.example.com/realms/iam/.well-known/openid-configuration \
     | jq '{issuer, token_endpoint, jwks_uri}'

   注意 issuer 由 Keycloak 的 hostname/frontend URL 配置决定：内网地址、外网地址、X-Forwarded-* 处理方式不同都会改变它。这类问题在 Keycloak hostname v2 配置 里有完整边界，Kafka 侧的表现就是 expected.issuer 永远比不上。

5. 服务账号要不要分配 realm 角色？只要 Kafka 不用它做授权，就不需要。 分配角色只影响 Keycloak 自己的审计和其它消费该 token 的系统；Kafka 的权限不读角色，见第 5 节。

3. Broker 端最小配置#

把客户端流量放到一个独立的 SASL_SSL 监听器上，inter-broker 继续走原有的 SSL/mTLS 监听器——这样认证改造的影响面只在前门，回滚也最简单。

# server.properties
listeners=BROKER://0.0.0.0:9092,CLIENT://0.0.0.0:9093
advertised.listeners=BROKER://kafka.example.com:9092,CLIENT://kafka.example.com:9093
listener.security.protocol.map=BROKER:SSL,CLIENT:SASL_SSL
inter.broker.listener.name=BROKER

# 只对 CLIENT 监听器开 OAUTHBEARER
sasl.enabled.mechanisms=OAUTHBEARER
listener.name.client.sasl.enabled.mechanisms=OAUTHBEARER

# 校验器：注意这一段是 oauthbearer.sasl.oauthbearer.*，机制名与配置前缀都会出现
listener.name.client.oauthbearer.sasl.server.callback.handler.class=org.apache.kafka.common.security.oauthbearer.OAuthBearerValidatorCallbackHandler
listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required;
listener.name.client.oauthbearer.sasl.oauthbearer.jwks.endpoint.url=https://kc.example.com/realms/iam/protocol/openid-connect/certs

# 信任条件：issuer 与 audience 都是精确匹配
sasl.oauthbearer.expected.issuer=https://kc.example.com/realms/iam
sasl.oauthbearer.expected.audience=kafka

# principal 取哪个 claim
sasl.oauthbearer.sub.claim.name=preferred_username

# 可选：JWKS 刷新与时钟偏差（默认 3600s / 30s，通常够用）
sasl.oauthbearer.jwks.endpoint.refresh.ms=300000
sasl.oauthbearer.clock.skew.seconds=30

同时给 broker 进程加 JVM 系统属性（systemd unit 的 Environment=KAFKA_OPTS=... 或 kafka-server-start.sh 的启动环境）：

export KAFKA_OPTS="-Dorg.apache.kafka.sasl.oauthbearer.allowed.urls=https://kc.example.com/realms/iam/protocol/openid-connect/certs"

三个容易配错的地方：

1. 两段前缀是官方写法，不是笔误。 broker 侧既要用 listener.name.<listener>.oauthbearer.sasl.<机制配置> 的形态声明 JAAS 与回调类，也会用到 listener.name.<listener>.oauthbearer.sasl.oauthbearer.jwks.endpoint.url 这种「机制名出现两次」的路径——后者官方文档与 OAuthBearerValidatorCallbackHandler 的 javadoc 都这么写。
2. JWKS 只允许 HTTPS 或 file://。 协议白名单是 http/https/file 三者，生产用 https；离线环境可以把 JWKS 落成文件（此时未知 kid 会直接拒绝，而不是再去拉一次）。
3. inter-broker 也要 OAUTHBEARER 的话，别忘了 broker 自己还要作为 client 取 token：加上 listener.name.<listener>.oauthbearer.sasl.login.callback.handler.class 与 token endpoint、client 凭据。这一步会让「Keycloak 不可用 = 集群内部通信也受影响」，属于可用性权衡，不是纯配置工作。

4. 客户端最小配置#

# kafka-client.properties
security.protocol=SASL_SSL
sasl.mechanism=OAUTHBEARER
sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required;

# 4.x 推荐写法（JAAS 里的 clientId/clientSecret/scope 已标记弃用）
sasl.oauthbearer.client.credentials.client.id=kafka-client
sasl.oauthbearer.client.credentials.client.secret=<从 Secret 注入>
sasl.oauthbearer.scope=kafka
sasl.oauthbearer.token.endpoint.url=https://kc.example.com/realms/iam/protocol/openid-connect/token

ssl.truststore.location=/etc/kafka/ssl/client.truststore.p12
ssl.truststore.password=<从 Secret 注入>
ssl.truststore.type=PKCS12

要点：

• sasl.jaas.config 这一行要保留（值里可以只有 required;），它决定该连接走哪个登录模块；真正的 OAuth 参数走上面那些 sasl.oauthbearer.* 配置项。
• clientId / clientSecret / scope 写在 JAAS 里是历史写法，已弃用。 新配置用 sasl.oauthbearer.client.credentials.client.id / .client.secret / sasl.oauthbearer.scope，两者都在时以配置项为准。
• sasl.oauthbearer.scope 的值必须与 Keycloak 里真实存在的 client scope 同名（这里就是第 2 节那个 kafka）。它只是「请求哪些 scope」，不决定 mapper 是否生效：该 scope 若已是 realm 的 Default 或已 assign 给这个 client，省略这一行结果一样。名字对不上不会让 aud 消失，只会白绕一圈排查。
• 命令行工具（kafka-console-consumer.sh、kafka-acls.sh）也是 Java 客户端，同样需要 KAFKA_OPTS 里的 allowed.urls（见 1.2），别只给应用进程加。
• 别抄 3.x 时期的包路径。 org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerLoginCallbackHandler 在 3.9 存在、4.3 已不存在——公开类的 oauthbearer.secured.* 在 4.x 被拍平到 oauthbearer 下，4.x 要用 org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler，否则得到一个 ClassNotFoundException。别把这个结论推广到所有叫 secured 的包：内部实现包 oauthbearer.internals.secured.*（ConfigurationUtils 就在里面）在 4.3 依旧存在。
• 非 Java 客户端（librdkafka 系）的 OAUTHBEARER 配置项与 Java 客户端不通用，需要按各自客户端文档单独配，不要拿上面的属性名硬套。

5. 授权：principal → ACL#

Kafka 的授权模型是 ACL，token 只是身份。一条最小可用授权：

kafka-acls.sh --bootstrap-server kafka.example.com:9093 \
  --command-config /etc/kafka/kafka-client.properties \
  --add --allow-principal User:service-account-kafka-client \
  --operation Read --topic orders

完整验证与问题定位：

# 列出现有 ACL，确认 principal 拼写与预期一致
kafka-acls.sh --bootstrap-server kafka.example.com:9093 \
  --command-config /etc/kafka/kafka-client.properties --list

• 这里 User: 后面的字符串就是 broker 认证出的 principal。如果你没改 sasl.oauthbearer.sub.claim.name，它会等于 token 的 sub（UUID），ACL 写 service-account-... 会一直不匹配——现象是认证通过、读写报 TopicAuthorizationException。
• 不要把 realm 角色写进 ACL：Keycloak 的角色不会自动变成 Kafka 权限。想做「按组授权」需要定制 principal builder 或使用发行版扩展能力，与本文的组合属于两条路线。
• 授权模型的分层（什么该在 IdP 做、什么该在资源侧做）见 Keycloak 细粒度权限与授权策略；Kafka 属于「资源侧自带授权引擎」这一类，IdP 只负责把主体认证出来。

6. 验证#

四步，前一步不通过就不要往下走：

# 1) 拿 token 并解码关键 claim（不要只看 HTTP 200）
TOKEN=$(curl -s -X POST https://kc.example.com/realms/iam/protocol/openid-connect/token \
  -d grant_type=client_credentials \
  -d client_id=kafka-client \
  -d client_secret="$KC_CLIENT_SECRET" | jq -r .access_token)

echo "$TOKEN" | cut -d. -f2 | tr '_-' '/+' | base64 -d 2>/dev/null \
  | jq '{iss, aud, azp, sub, preferred_username, exp}'

预期：iss 与 sasl.oauthbearer.expected.issuer 逐字符相同；aud 里含 kafka；preferred_username 为 service-account-kafka-client。

# 2) 确认 broker 真能拉到 JWKS 且 kid 对得上（比对 discovery 的 jwks_uri）
curl -s https://kc.example.com/realms/iam/protocol/openid-connect/certs | jq '.keys[].kid'

# 3) 用同一份 client.properties 做一个只读操作，把认证与授权分开看
kafka-topics.sh --bootstrap-server kafka.example.com:9093 \
  --command-config /etc/kafka/kafka-client.properties --list

# 4) 端到端消费一条，验证数据面
kafka-console-consumer.sh --bootstrap-server kafka.example.com:9093 \
  --topic orders --from-beginning --max-messages 1 \
  --consumer.config /etc/kafka/kafka-client.properties

第 3 步就失败时看 broker 日志里的 SASL 阶段：正常通过是建立连接，失败会在握手处留下 invalid_token 之类的状态，并伴随 Authentication failed。因为 iss 与 aud 都是精确匹配，invalid_token 基本只有三种成因——签名/JWKS（含 kid 未刷新）、时间窗口（exp 或时钟偏差）、iss/aud 不匹配；按这三类逐个排除比逐条改配置快。

7. 常见错误对照表#

8. 回滚#

按「影响面从小到大」设计，回滚就是反向走一遍：

1. 保留旧监听器。 客户端流量切到新 CLIENT 监听器时，旧的 SASL_SSL+mTLS/SCRAM 监听器不要拆。回滚 = 把客户端的 client.properties 换回旧配置，broker 一行都不用改。这是最省事的回滚路径，代价是多占一个端口。
2. 只有确有必要才动 inter-broker。 如果确实把 inter-broker 也切成了 OAUTHBEARER，回滚要按官方的机制变更顺序来：先在 sasl.enabled.mechanisms 里让新旧机制并存并滚动重启 → 切换客户端 → 再移除旧机制并再次滚动重启。跳步会直接把集群切到不可通信。
3. Keycloak 侧是增量改动，回滚成本低但要防误删。 Audience mapper、client scope 都是新增对象；移除 mapper 只会让 token 失去那个 aud（Kafka 随之拒绝），不会影响其它已接入系统。删 kafka-client 之前先确认没有别的消费方在用同一个 client。
4. 凭据轮换而非覆盖。 换 secret 时先在 Keycloak 侧生成新值、验证新配置可用，再停用旧值；不要在同一时刻既改 broker 又改客户端。
5. 不要在故障期间打开 allow.unverified.audience / allow.unverified.issuer「先恢复再说」。 那是把校验永久关掉，事后极容易忘记关；这类改动必须走变更单，并写清楚失效时间。

常见问题（FAQ）#

Kafka 用 Keycloak 认证后，realm 角色能直接当 Kafka 权限用吗？ 不能。Apache Kafka 内置的 OAuth 链路只把 token 换成一个 principal，授权完全在 ACL。想按身份组授权需要定制 principal.builder.class 或使用发行版的 OAuth 扩展（Strimzi、Confluent 都有各自实现）。把「角色 → 组 → ACL」这层映射交给 IdP 是常见的期望，但在 Apache Kafka 里它不存在，别按期望配置。

为什么我按老文章配了 listener.name.<x>.oauthbearer.sasl.jaas.config 里的 clientId/clientSecret 还是能跑？ JAAS 里的 clientId / clientSecret / scope 目前仍可用但已弃用，官方保留的是向后兼容。能跑不代表应该继续用：迁移到 sasl.oauthbearer.client.credentials.client.id / .client.secret / sasl.oauthbearer.scope 是一次纯配置改动，趁现在做比等它被移除再做便宜。

sasl.oauthbearer.expected.audience 能写成 * 或省略吗？ 不能。Apache Kafka 的实现是精确字符串比对（allowed.contains(configValue) 那种字段级比对同理，没有通配）；省略则等于不校验 audience（4.4.0 起在配了 JWKS 时直接启动失败）。真正的处理方式是让 Keycloak 发出正确的 aud，用 mapper 解决，而不是放宽 broker 的校验。

mapper 配好了，解码 access token 就是没有 aud，接下来查什么？ 两处最容易被忽略。一是这个 client 是否处于轻量级 access token 模式（客户端属性 client.use.lightweight.access.token.enabled，控制台里由 client policy use-lightweight-access-token 打开）：这种模式下 transformAccessToken 只认 mapper 的 lightweight.claim，「Add to access token」被忽略，勾上 Add to lightweight access token 才会有 aud。二是 mapper 里填的是哪个字段：included.client.audience 优先于 included.custom.audience，只填了前者而 Kafka 要的 kafka 写在后者上，就什么都看不到。

主要来源#

• Apache Kafka 官方文档（4.3）： Authentication using SASL（生产 broker/客户端的 OAUTHBEARER 配置形态、listener.name.<listener>.oauthbearer.sasl.oauthbearer.jwks.endpoint.url 的双段前缀、unsecured 实现的生产边界）、 Broker Configs
• Apache Kafka 源码 4.3 分支：clients/.../config/SaslConfigs.java（配置名与默认值——jwks.endpoint.refresh.ms 3600 秒、clock.skew.seconds 30、header.urlencode false、sub.claim.name 默认 sub、client.credentials.client.id / .client.secret、expected.audience 默认空列表）、clients/.../config/internals/BrokerSecurityConfigs.java（org.apache.kafka.sasl.oauthbearer.allowed.urls，默认值 ""）、.../oauthbearer/internals/secured/ConfigurationUtils.java（allowed.contains(configValue) 精确匹配、URL 只允许 http/https/file、报错文案）与 ConfigurationUtilsTest.java（By default, no URL is allowed）、.../oauthbearer/OAuthBearerValidatorCallbackHandler.java、.../oauthbearer/OAuthBearerLoginCallbackHandler.java（4.x 的公开包路径；对照 3.9 的 oauthbearer/secured/ 路径）
• Apache Kafka 升级说明 docs/getting-started/upgrade.md：4.0.0 引入 org.apache.kafka.sasl.oauthbearer.allowed.urls 且「By default, the value is an empty list」；trunk 中 4.4.0 一节的 OAUTHBEARER 校验器 fail-fast 与新增的 allow.unverified.*
• Keycloak 源码 26.7.4：AudienceProtocolMapper.java（// Don't include audience in ID Token by default、included.client.audience 优先于 included.custom.audience）、AudienceResolveProtocolMapper.java（// Don't add client itself to the audience）、AbstractOIDCProtocolMapper.java（轻量 token 下以 lightweight.claim 取代 access token 开关）与 OIDCAttributeMapperHelper.java（INCLUDE_IN_LIGHTWEIGHT_ACCESS_TOKEN = "lightweight.claim"）、models/Constants.java（client.use.lightweight.access.token.enabled）、clientpolicy/executor/UseLightweightAccessTokenExecutorFactory.java（client policy id use-lightweight-access-token）、common/.../constants/ServiceAccountConstants.java（SERVICE_ACCOUNT_USER_PREFIX = "service-account-"）、protocol/oidc/TokenManager.java（token.subject(user.getId())）
• Keycloak 文档： Server Administration Guide（服务账号与客户端凭据、协议 mapper）