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 作为用户名，以及一个 secret 作为密码。
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. 安装 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. 创建并启用 realm。

   您可以使用现有的 master 域。

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

6. 在 Settings 选项卡中设置：

   • 访问类型 以 机密
   • 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://<oauth_server_address>/auth/realms/<realm_name>" \
     oauth.jwks.endpoint.uri="https://<oauth_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://<oauth_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://<oauth_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://<oauth_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。当使用 KeycloakAuthorizer 或启用了 OAuth 2.0 的监听程序时，需要用于代理间通信。

   2

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

   3

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

   4

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

   5

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

   6

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

   7

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

   8

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

   9

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

   10

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

   11

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

   12

   尝试另一个到授权服务器的 HTTP 请求重试前等待的时间。默认情况下，这个时间被设置为零，这意味着不会应用暂停。这是因为导致失败请求的许多问题是针对每个请求的网络粘合或代理问题，可以快速解决。但是，如果您的授权服务器处于压力下或高流量，您可能希望将此选项设置为值 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://<oauth_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 属性不可用，则使用这个 fallback 选项指定要使用的用户名声明(attribute)。如果需要，您可以使用 JsonPath 表达式，如 "['client.info'].['client.id']"，从令牌中的嵌套 JSON 属性检索回退用户名。

   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.13.0.redhat-00015</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

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

     2

     指定为 OAUTHBEARER 或 PLAIN 的 SASL 机制。

     3

     用于安全访问 Kafka 集群的 truststore 配置。

     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. 在 Java 客户端代码中输入 OAUTH 2.0 身份验证的客户端属性。

   显示客户端属性输入示例

   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 代理。