5.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 做为用户名和密码传递一个 secret。
2. Kafka 代理使用令牌端点将 clientId 和 secret 传递给授权服务器。
3. 如果客户端凭据无效，授权服务器会返回全新的访问令牌或错误。

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

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

1. Kafka 客户端通过用户名和密码。密码提供运行客户端之前手动获取的访问令牌的值。

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

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

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

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

流程

1. 安装红帽单点登录。

   您可以从 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 选项卡中设置：

   • 访问类型 到 机密
   • 启用标准流 到 OFF 以禁用这个客户端的 Web 登录
   • 已启用 到 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

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

   示例 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" \
     # ...

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" ;

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

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

      检查可确保客户端与授权服务器连接是 authentic。您可能需要在非生产环境中关闭验证。

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.groups.claim="$.groups" \ 

   11

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

   12

   1

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

   2

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

   3

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

   4

   （可选） 检查。如果您的授权服务器 提供一个ud(audience )声明，而您想要强制执行受众检查，请将 ouath.check.audience 设置为 true。使用者检查确定了令牌的预期接收者。因此，Kafka 代理将拒绝在其 aud 声明中没有 clientId 的令牌。默认为 false。

   5

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

   6

   有效签发者 URI。只有此签发者发布的访问令牌才会被接受。（始终需要。）

   7

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

   8

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

   9

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

   10

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

   11

   用于从 JWT 令牌或内省端点响应中提取组信息的 JsonPath 查询。默认不设置。自定义授权器可以使用它根据用户组进行授权决策。

   12

   作为分隔字符串返回时，用于解析组信息的分隔符。默认值为 ','(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

   1

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

   2

   授权服务器可能无法提供单一属性来识别常规用户和客户端。当客户端以自己的名称进行身份验证时，服务器可能会提供 客户端 ID。当用户使用用户名和密码进行身份验证时，要获取刷新令牌或访问令牌，服务器可能会除了客户端 ID 外提供 用户名 属性。如果主用户 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.11.0.redhat-00003</version>
   </dependency>

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

     1

     用于 TLS 加密连接的 SASL_SSL 安全协议。仅在本地开发中使用 SASL_PLAINTEXT。

     2

     指定为 OAUTHBEARER 或 PLAIN 的 SASL 机制。

     3

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

     4

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

     5

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

     6

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

     7

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

     8

     用于访问 truststore 的密码。

     9

     信任存储类型。

     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>" ;

     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" \

     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" \

     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);
   }

4. 验证 Kafka 客户端能否访问 Kafka 代理。