6.4. 配置 Kafka

1. Kafka 客户端使用客户端 ID 和 secret 从授权服务器请求访问令牌，以及可选的刷新令牌。或者，客户端也可以使用用户名和密码进行身份验证。
2. 授权服务器生成新的访问令牌。
3. Kafka 客户端使用 SASL OAUTHBEARER 机制通过 Kafka 代理进行身份验证来传递访问令牌。
4. Kafka 代理通过使用自己的客户端 ID 和 secret，在授权服务器上调用令牌内省端点来验证访问令牌。
5. 如果令牌有效，则会建立 Kafka 客户端会话。

1. Kafka 客户端使用令牌端点、使用客户端 ID 和 secret 以及刷新令牌（可选）从令牌端点验证。或者，客户端也可以使用用户名和密码进行身份验证。
2. 授权服务器生成新的访问令牌。
3. Kafka 客户端使用 SASL OAUTHBEARER 机制通过 Kafka 代理进行身份验证来传递访问令牌。
4. Kafka 代理使用 JWT 令牌签名检查和本地令牌内省在本地验证访问令牌。

1. Kafka 客户端使用 SASL OAUTHBEARER 机制通过 Kafka 代理进行身份验证，以传递长期访问令牌。
2. Kafka 代理通过使用自己的客户端 ID 和 secret，在授权服务器上调用令牌内省端点来验证访问令牌。
3. 如果令牌有效，则会建立 Kafka 客户端会话。

1. Kafka 客户端使用 SASL OAUTHBEARER 机制通过 Kafka 代理进行身份验证，以传递长期访问令牌。
2. Kafka 代理使用 JWT 令牌签名检查和本地令牌内省在本地验证访问令牌。

1. Kafka 客户端以用户名和密码形式 传递 clientId。
2. Kafka 代理使用令牌端点将 clientId 和 secret 传递给授权服务器。
3. 如果客户端凭据无效，授权服务器会返回新的访问令牌或错误。

4. Kafka 代理以以下方法之一验证令牌：

   1. 如果指定了令牌内省端点，Kafka 代理通过在授权服务器上调用端点来验证访问令牌。如果令牌验证成功，则会建立会话。
   2. 如果使用本地令牌内省，则不会向授权服务器发出请求。Kafka 代理使用 JWT 令牌签名检查在本地验证访问令牌。

1. Kafka 客户端会传递用户名和密码。密码提供访问令牌的值，该令牌已在运行客户端之前手动获取和配置。

2. 密码通过或不使用 $accessToken: 字符串前缀来传递，具体取决于 Kafka 代理侦听程序是否配置了令牌端点来进行身份验证。

   1. 如果配置了令牌端点，密码应以 $accessToken: 前缀，以便代理知道 password 参数包含访问令牌而不是客户端 secret。Kafka 代理将用户名解释为帐户用户名。
   2. 如果没有在 Kafka 代理监听程序上配置令牌端点（强制使用 no-client-credentials 模式），密码应提供没有前缀的访问令牌。Kafka 代理将用户名解释为帐户用户名。在这个模式中，客户端不使用客户端 ID 和 secret，password 参数始终被解释为原始访问令牌。

3. Kafka 代理以以下方法之一验证令牌：

   1. 如果指定了令牌内省端点，Kafka 代理通过在授权服务器上调用端点来验证访问令牌。如果令牌验证成功，则会建立会话。
   2. 如果使用本地令牌内省，则不会向授权服务器发出请求。Kafka 代理使用 JWT 令牌签名检查在本地验证访问令牌。

流程

1. Install Red Hat Single Sign-On.

   您可以从 ZIP 文件或使用 RPM 安装。

2. 登录到 Red Hat Single Sign-On Admin 控制台，为 AMQ Streams 创建 OAuth 2.0 策略。

   部署 Red Hat Single Sign-On 时提供登录详情。

3. 创建并启用域。

   您可以使用现有的 master 域。

4. 如果需要，调整域的会话和令牌超时。
5. 创建名为 kafka-broker 的客户端。

6. 在 Settings 选项卡中设置：

   • 访问 Type 为 Confidential
   • Standard Flow Enabled 为 OFF 为这个客户端禁用 Web 登录
   • Service Accounts Enabled 为 ON，以允许此客户端在其自己的名称中进行身份验证
7. 在继续操作前，点 Save。
8. 在 Credentials 选项卡中，记录使用 AMQ Streams Kafka 集群配置的 secret。

9. 对要连接到 Kafka 代理的任何应用程序客户端重复客户端创建步骤。

   为每个新客户端创建一个定义。

   您将在您的配置中将名称用作客户端 ID。

流程

1. 在 server.properties 文件中配置 Kafka 代理监听程序配置。

   例如，使用 OAUTHBEARER 机制：

   sasl.enabled.mechanisms=OAUTHBEARER
   listeners=CLIENT://0.0.0.0:9092
   listener.security.protocol.map=CLIENT:SASL_PLAINTEXT
   listener.name.client.sasl.enabled.mechanisms=OAUTHBEARER
   sasl.mechanism.inter.broker.protocol=OAUTHBEARER
   inter.broker.listener.name=CLIENT
   listener.name.client.oauthbearer.sasl.server.callback.handler.class=io.strimzi.kafka.oauth.server.JaasServerOauthValidatorCallbackHandler
   listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required ;
   listener.name.client.oauthbearer.sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler

   Copy to Clipboard Toggle word wrap

2. 将代理连接设置配置为 listener.name.client.oauthbearer.sasl.jaas.config 的一部分。

   此处的示例显示了连接配置选项。

   示例 1：使用 JWKS 端点配置的本地令牌验证

   listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
     oauth.valid.issuer.uri="https://AUTH-SERVER-ADDRESS/auth/realms/REALM-NAME" \
     oauth.jwks.endpoint.uri="https://AUTH-SERVER-ADDRESS/auth/realms/REALM-NAME/protocol/openid-connect/certs" \
     oauth.jwks.refresh.seconds="300" \
     oauth.jwks.refresh.min.pause.seconds="1" \
     oauth.jwks.expiry.seconds="360" \
     oauth.username.claim="preferred_username" \
     oauth.ssl.truststore.location="PATH-TO-TRUSTSTORE-P12-FILE" \
     oauth.ssl.truststore.password="TRUSTSTORE-PASSWORD" \
     oauth.ssl.truststore.type="PKCS12" ;
   listener.name.client.oauthbearer.connections.max.reauth.ms=3600000

   Copy to Clipboard Toggle word wrap

   示例 2：通过 OAuth 2.0 内省端点将令牌验证委派给授权服务器

   listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
     oauth.introspection.endpoint.uri="https://AUTH-SERVER-ADDRESS/auth/realms/REALM-NAME/protocol/openid-connect/introspection" \
     # ...

   Copy to Clipboard Toggle word wrap

3. 如果需要，配置对授权服务器的访问。

   生产环境通常需要这一步，除非使用 service mesh 等技术来配置容器外的安全频道。

   1. 提供用于连接安全授权服务器的自定义信任存储。访问授权服务器时始终需要 SSL。

      设置属性来配置信任存储。

      例如：

      listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
        # ...
        oauth.client.id="kafka-broker" \
        oauth.client.secret="kafka-broker-secret" \
        oauth.ssl.truststore.location="PATH-TO-TRUSTSTORE-P12-FILE" \
        oauth.ssl.truststore.password="TRUSTSTORE-PASSWORD" \
        oauth.ssl.truststore.type="PKCS12" ;

      Copy to Clipboard Toggle word wrap

   2. 如果证书主机名与访问 URL 主机名不匹配，您可以关闭证书主机名验证：

      oauth.ssl.endpoint.identification.algorithm=""

      Copy to Clipboard Toggle word wrap

      检查可确保与授权服务器的连接是真实的。您可能想要在非生产环境中关闭验证。

4. 根据您选择的身份验证流程配置附加属性：

   listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
     # ...
     oauth.token.endpoint.uri="https://AUTH-SERVER-ADDRESS/auth/realms/REALM-NAME/protocol/openid-connect/token" \ 

   1

   
     oauth.custom.claim.check="@.custom == 'custom-value'" \ 

   2

   
     oauth.scope="SCOPE" \ 

   3

   
     oauth.check.audience="true" \ 

   4

   
     oauth.audience="AUDIENCE" \ 

   5

   
     oauth.valid.issuer.uri="https://https://AUTH-SERVER-ADDRESS/auth/REALM-NAME" \ 

   6

   
     oauth.client.id="kafka-broker" \ 

   7

   
     oauth.client.secret="kafka-broker-secret" \ 

   8

   
     oauth.connect.timeout.seconds=60 \ 

   9

   
     oauth.read.timeout.seconds=60 \ 

   10

   
     oauth.http.retries=2 \ 

   11

   
     oauth.http.retry.pause.millis=300 \ 

   12

   
     oauth.groups.claim="$.groups" \ 

   13

   
     oauth.groups.claim.delimiter="," ; 

   14

   Copy to Clipboard Toggle word wrap

   1

   您的授权服务器的 OAuth 2.0 令牌端点 URL。对于生产环境，请始终使用 https:// urls。当使用 KeycloakRBACAuthorizer 或启用了 OAuth 2.0 的监听程序时，需要用于 Inter-broker 间通信。

   2

   （可选 ）自定义声明检查。JsonPath 过滤器查询，该查询在验证期间将其他自定义规则应用到 JWT 访问令牌。如果访问令牌不包含必要的数据，则会被拒绝。使用 introspection 端点方法时，自定义检查将应用到内省端点响应 JSON。

   3

   （可选）传递给令牌端点的 scope 参数。获取访问令牌进行代理身份验证时使用的 scope。它还在使用 clientId 和 secret 的 PLAIN 客户端验证中用于 OAuth 2.0 的客户端名称。这只会影响获取令牌的能力，以及令牌的内容，具体取决于授权服务器。它不会影响监听程序的令牌验证规则。

   4

   （可选） 对象检查。如果您的授权服务器提供 aud (audience)声明，并且您希望强制读者检查，请将 ouath.check.audience 设置为 true。受众检查标识令牌的预期接收者。因此，Kafka 代理将拒绝在其 aud 声明中没有 clientId 的令牌。默认为 false。

   5

   （可选）传递给令牌端点的 audience 参数。在获取用于代理身份验证的访问令牌时，需要使用 audience。它还在使用 clientId 和 secret 的 PLAIN 客户端验证中用于 OAuth 2.0 的客户端名称。这只会影响获取令牌的能力，以及令牌的内容，具体取决于授权服务器。它不会影响监听程序的令牌验证规则。

   6

   有效的签发者 URI。仅接受此签发者发布的访问令牌。（始终必需。）

   7

   Kafka 代理配置的客户端 ID，适用于所有代理。这是在 授权服务器中注册的客户端为 kafka-broker。当内省端点用于令牌验证时，或者在使用 KeycloakRBACAuthorizer 时是必需的。

   8

   Kafka 代理配置的 secret，适用于所有代理。当代理必须向授权服务器（客户端 secret、访问令牌或刷新令牌）进行身份验证时，必须指定刷新令牌。

   9

   （可选）连接到授权服务器时连接超时（以秒为单位）。默认值为 60。

   10

   （可选）连接到授权服务器时的读取超时（以秒为单位）。默认值为 60。

   11

   将失败的 HTTP 请求重试到授权服务器的次数上限。默认值为 0，表示没有执行重试。要有效地使用这个选项，请考虑减少 oauth.connect.timeout.seconds 和 oauth.read.timeout.seconds 选项的超时时间。但请注意，重试可能会阻止当前 worker 线程对其他请求使用，如果太多请求停滞，可能会导致 Kafka 代理无响应。

   12

   在尝试另一个失败 HTTP 请求重试到授权服务器前等待的时间。默认情况下，这段时间设为零，这意味着不会应用暂停。这是因为，导致失败请求的很多问题都是可快速解决的网络 glitches 或代理问题。但是，如果您的授权服务器处于压力或遇到高流量，您可能希望将此选项设置为 100 ms 或更高值，以减少服务器上的负载并增加成功重试的可能性。

   13

   JsonPath 查询，用于从 JWT 令牌或内省端点响应中提取组信息。默认情况下不设置。这可以供自定义授权者用于根据用户组做出授权决策。

   14

   以单个分隔字符串返回时用于解析组信息的分隔符。默认值为 ','(comma)。

5. 根据您如何应用 OAuth 2.0 身份验证以及所使用的授权服务器类型，添加额外的配置设置：

   listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
     # ...
     oauth.check.issuer=false \ 

   1

   
     oauth.fallback.username.claim="CLIENT-ID" \ 

   2

   
     oauth.fallback.username.prefix="CLIENT-ACCOUNT" \ 

   3

   
     oauth.valid.token.type="bearer" \ 

   4

   
     oauth.userinfo.endpoint.uri="https://AUTH-SERVER-ADDRESS/auth/realms/REALM-NAME/protocol/openid-connect/userinfo" ; 

   5

   Copy to Clipboard Toggle word wrap

   1

   如果您的授权服务器不提供 iss 声明，则无法执行签发者检查。在这种情况下，将 oauth.check.issuer 设置为 false，且不指定 oauth.valid.issuer.uri。默认为 true。

   2

   授权服务器可能无法提供单个属性来识别常规用户和客户端。当客户端使用自己的名称进行身份验证时，服务器可能会提供 客户端 ID。当用户使用用户名和密码进行身份验证时，要获取刷新令牌或访问令牌时，除了客户端 ID 外，服务器可能会提供 username 属性。如果主用户 ID 属性不可用，则使用此回退选项指定用户名声明(attribute)。

   3

   当 oauth.fallback.username.claim 适用的情况下，可能需要防止用户名声明的值和回退用户名声明之间的名称冲突。例如，存在名为 producer 的客户端，但也存在名为 producer 的常规用户。为了区分两者，您可以使用此属性向客户端的用户 ID 添加前缀。

   4

   （仅在使用 oauth.introspection.endpoint.uri）取决于您使用的授权服务器时，内省端点可能会也可能不会返回 令牌类型 属性，或者可能包含不同的值。您可以指定来自内省端点的响应的有效令牌类型值。

   5

   （仅在使用 oauth.introspection.endpoint.uri时使用，可以配置或实施授权服务器，以便在内省端点响应中不提供任何可识别的信息。要获取用户 ID，您可以将 userinfo 端点的 URI 配置为回退。oauth.fallback.username.claim、oauth.fallback.username.claim 和 oauth.fallback.username.prefix 设置应用到 userinfo 端点的响应。

流程

1. 将带有 OAuth 2.0 支持的客户端库添加到 Kafka 客户端的 pom.xml 文件中：

   <dependency>
    <groupId>io.strimzi</groupId>
    <artifactId>kafka-oauth-client</artifactId>
    <version>0.12.0.redhat-00006</version>
   </dependency>

   Copy to Clipboard Toggle word wrap

2. 通过在属性文件中指定以下配置来配置客户端属性：

   • 安全协议
   • SASL 机制

   • JAAS 模块和身份验证属性，取决于所使用的方法

     例如，我们可以将以下内容添加到 client.properties 文件中：

     客户端凭证机制属性

     security.protocol=SASL_SSL 

     1

     
     sasl.mechanism=OAUTHBEARER 

     2

     
     ssl.truststore.location=/tmp/truststore.p12 

     3

     
     ssl.truststore.password=$STOREPASS
     ssl.truststore.type=PKCS12
     sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
       oauth.token.endpoint.uri="<token_endpoint_url>" \ 

     4

     
       oauth.client.id="<client_id>" \ 

     5

     
       oauth.client.secret="<client_secret>" \ 

     6

     
       oauth.ssl.truststore.location="/tmp/oauth-truststore.p12" \ 

     7

     
       oauth.ssl.truststore.password="$STOREPASS" \ 

     8

     
       oauth.ssl.truststore.type="PKCS12" \ 

     9

     
       oauth.scope="<scope>" \ 

     10

     
       oauth.audience="<audience>" ; 

     11

     Copy to Clipboard Toggle word wrap

     1

     用于 TLS 加密连接的 SASL_SSL 安全协议。使用 SASL_PLAINTEXT over unencrypted 连接仅进行本地开发。

     2

     指定为 OAUTHBEARER 或 PLAIN 的 SASL 机制。

     3

     用于安全访问 Kafka 集群的信任存储配置。

     4

     授权服务器令牌端点的 URI。

     5

     客户端 ID，这是在授权服务器中创建客户端时使用的名称。

     6

     在授权服务器中创建 客户端 时创建的客户端 secret。

     7

     位置包含授权服务器的公钥证书(truststore.p12)。

     8

     用于访问 truststore 的密码。

     9

     truststore 类型。

     10

     （可选）从令牌端点请求令牌的范围。授权服务器可能需要客户端来指定范围。

     11

     （可选）从令牌端点请求令牌的听众。授权服务器可能需要客户端来指定受众。

     密码授予机制属性

     security.protocol=SASL_SSL
     sasl.mechanism=OAUTHBEARER
     ssl.truststore.location=/tmp/truststore.p12
     ssl.truststore.password=$STOREPASS
     ssl.truststore.type=PKCS12
     sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
       oauth.token.endpoint.uri="<token_endpoint_url>" \
       oauth.client.id="<client_id>" \ 

     1

     
       oauth.client.secret="<client_secret>" \ 

     2

     
       oauth.password.grant.username="<username>" \ 

     3

     
       oauth.password.grant.password="<password>" \ 

     4

     
       oauth.ssl.truststore.location="/tmp/oauth-truststore.p12" \
       oauth.ssl.truststore.password="$STOREPASS" \
       oauth.ssl.truststore.type="PKCS12" \
       oauth.scope="<scope>" \
       oauth.audience="<audience>" ;

     Copy to Clipboard Toggle word wrap

     1

     客户端 ID，这是在授权服务器中创建客户端时使用的名称。

     2

     （可选）在授权服务器中创建客户端时创建的客户端 secret。

     3

     password 授权身份验证的用户名。OAuth 密码授权配置（用户名和密码）使用 OAuth 2.0 密码授权方法。要使用密码授权，请在您的授权服务器上为客户端创建一个有限权限的用户帐户。帐户应像服务帐户一样操作。在进行身份验证需要用户帐户的环境中使用，但首先考虑使用刷新令牌。

     4

     密码以授予密码身份验证。

     注意

     SASL PLAIN 不支持使用 OAuth 2.0 密码授权方法传递用户名和密码（密码授予）。

     访问令牌属性

     security.protocol=SASL_SSL
     sasl.mechanism=OAUTHBEARER
     ssl.truststore.location=/tmp/truststore.p12
     ssl.truststore.password=$STOREPASS
     ssl.truststore.type=PKCS12
     sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
       oauth.token.endpoint.uri="<token_endpoint_url>" \
       oauth.access.token="<access_token>" ; 

     1

     
       oauth.ssl.truststore.location="/tmp/oauth-truststore.p12" \
       oauth.ssl.truststore.password="$STOREPASS" \
       oauth.ssl.truststore.type="PKCS12" \

     Copy to Clipboard Toggle word wrap

     1

     Kafka 客户端的长期访问令牌。

     刷新令牌属性

     security.protocol=SASL_SSL
     sasl.mechanism=OAUTHBEARER
     ssl.truststore.location=/tmp/truststore.p12
     ssl.truststore.password=$STOREPASS
     ssl.truststore.type=PKCS12
     sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
       oauth.token.endpoint.uri="<token_endpoint_url>" \
       oauth.client.id="<client_id>" \ 

     1

     
       oauth.client.secret="<client_secret>" \ 

     2

     
       oauth.refresh.token="<refresh_token>" ; 

     3

     
       oauth.ssl.truststore.location="/tmp/oauth-truststore.p12" \
       oauth.ssl.truststore.password="$STOREPASS" \
       oauth.ssl.truststore.type="PKCS12" \

     Copy to Clipboard Toggle word wrap

     1

     客户端 ID，这是在授权服务器中创建客户端时使用的名称。

     2

     （可选）在授权服务器中创建客户端时创建的客户端 secret。

     3

     Kafka 客户端的长期刷新令牌。

3. 将 OAUTH 2.0 身份验证的客户端属性输入到 Java 客户端代码中。

   显示客户端属性输入示例

   Properties props = new Properties();
   try (FileReader reader = new FileReader("client.properties", StandardCharsets.UTF_8)) {
     props.load(reader);
   }

   Copy to Clipboard Toggle word wrap

4. 验证 Kafka 客户端可以访问 Kafka 代理。