4.10. 使用基于 OAuth 2.0 令牌的身份验证

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://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.refresh.token="REFRESH-TOKEN-FOR-KAFKA-BROKERS" \ 

   9

   
     oauth.access.token="ACCESS-TOKEN-FOR-KAFKA-BROKERS" ; 

   10

   
     oauth.connect.timeout.seconds=60 

   11

   
     oauth.read.timeout.seconds=60 

   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。它还在使用 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的客户端。当内省端点用于令牌验证时，或使用 KeycloakRBACAuthorizer 时是必需的。

   8

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

   9

   （可选） Kafka 代理的长期刷新令牌。

   10

   （可选） Kafka 代理的长期访问令牌。

   11

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

   12

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

   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 属性不可用，则使用这个 fallback 选项指定要使用的用户名声明(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.10.0.redhat-00002</version>
   </dependency>

   Copy to Clipboard Toggle word wrap

2. 为回调配置系统属性：

   例如：

   System.setProperty(ClientConfig.OAUTH_TOKEN_ENDPOINT_URI, “https://<auth-server-address>/auth/realms/master/protocol/openid-connect/token”); 

   1

   
   System.setProperty(ClientConfig.OAUTH_CLIENT_ID, "<client_name>"); 

   2

   
   System.setProperty(ClientConfig.OAUTH_CLIENT_SECRET, "<client_secret>"); 

   3

   
   System.setProperty(ClientConfig.OAUTH_SCOPE, "<scope_value>") 

   4

   
   System.setProperty(ClientConfig.OAUTH_AUDIENCE, "<audience_value") 

   5

   Copy to Clipboard Toggle word wrap

   1

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

   2

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

   3

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

   4

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

   5

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

3. 在 Kafka 客户端配置中的 TLS 加密连接上启用 OAUTHBEARER 或 PLAIN 机制。

   例如：

   为 Kafka 客户端启用 OAUTHBEARER

   props.put("sasl.jaas.config", "org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required;");
   props.put("security.protocol", "SASL_SSL");
   props.put("sasl.mechanism", "OAUTHBEARER");
   props.put("sasl.login.callback.handler.class", "io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler");

   Copy to Clipboard Toggle word wrap

   为 Kafka 客户端启用 PLAIN

   props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"$CLIENT_ID_OR_ACCOUNT_NAME\" password=\"$SECRET_OR_ACCESS_TOKEN\" ;");
   props.put("security.protocol", "SASL_SSL"); 

   1

   
   props.put("sasl.mechanism", "PLAIN");

   Copy to Clipboard Toggle word wrap

   1

   在这里，我们使用 SASL_SSL 进行 TLS 连接。仅对本地开发使用 SASL_PLAINTEXT。

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