You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "kirktrue (via GitHub)" <gi...@apache.org> on 2025/05/02 15:31:29 UTC
[PR] KAFKA-18847: Refactor OAuth layer to improve reusability 1/N [kafka]
kirktrue opened a new pull request, #19622:
URL: https://github.com/apache/kafka/pull/19622
Rename `AccessTokenRetriever` and `AccessTokenValidator` to `JwtRetriever` and `JwtValidator`, respectively. Also converting the factory pattern classes `AccessTokenRetrieverFactory` and `AccessTokenValidatorFactory` into delegate/wrapper classes `DefaultJwtRetriever` and `DefaultJwtValidator`, respectively.
These are all internal changes, no configuration, user APIs, RPCs, etc. were changed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18847: Refactor OAuth layer to improve reusability 1/N [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19622:
URL: https://github.com/apache/kafka/pull/19622#discussion_r2086872075
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/DefaultJwtValidator.java:
##########
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.common.security.oauthbearer.internals.secured;
+
+import org.apache.kafka.common.security.oauthbearer.OAuthBearerToken;
+import org.apache.kafka.common.utils.Utils;
+
+import org.jose4j.keys.resolvers.VerificationKeyResolver;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_CLOCK_SKEW_SECONDS;
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_EXPECTED_AUDIENCE;
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_EXPECTED_ISSUER;
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_SCOPE_CLAIM_NAME;
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_SUB_CLAIM_NAME;
+
+/**
+ * This {@link JwtValidator} uses the delegation approach, instantiating and delegating calls to a
+ * more concrete implementation. The underlying implementation is determined by the presence/absence
+ * of the {@link VerificationKeyResolver}: if it's present, a {@link BrokerJwtValidator} is
+ * created, otherwise a {@link ClientJwtValidator} is created.
+ */
+public class DefaultJwtValidator implements JwtValidator {
+
+ private final Map<String, ?> configs;
+ private final String saslMechanism;
+ private final Optional<VerificationKeyResolver> verificationKeyResolver;
+
+ private JwtValidator delegate;
+
+ public DefaultJwtValidator(Map<String, ?> configs, String saslMechanism) {
+ this.configs = configs;
+ this.saslMechanism = saslMechanism;
+ this.verificationKeyResolver = Optional.empty();
+ }
+
+ public DefaultJwtValidator(Map<String, ?> configs,
+ String saslMechanism,
+ VerificationKeyResolver verificationKeyResolver) {
+ this.configs = configs;
+ this.saslMechanism = saslMechanism;
+ this.verificationKeyResolver = Optional.of(verificationKeyResolver);
+ }
+
+ @Override
+ public void init() throws IOException {
+ ConfigurationUtils cu = new ConfigurationUtils(configs, saslMechanism);
+
+ if (verificationKeyResolver.isPresent()) {
+ List<String> expectedAudiencesList = cu.get(SASL_OAUTHBEARER_EXPECTED_AUDIENCE);
+ Set<String> expectedAudiences = expectedAudiencesList != null ? Set.copyOf(expectedAudiencesList) : null;
+ Integer clockSkew = cu.validateInteger(SASL_OAUTHBEARER_CLOCK_SKEW_SECONDS, false);
+ String expectedIssuer = cu.validateString(SASL_OAUTHBEARER_EXPECTED_ISSUER, false);
+ String scopeClaimName = cu.validateString(SASL_OAUTHBEARER_SCOPE_CLAIM_NAME);
+ String subClaimName = cu.validateString(SASL_OAUTHBEARER_SUB_CLAIM_NAME);
+
+ delegate = new BrokerJwtValidator(clockSkew,
+ expectedAudiences,
+ expectedIssuer,
+ verificationKeyResolver.get(),
+ scopeClaimName,
+ subClaimName);
+ } else {
+ String scopeClaimName = cu.get(SASL_OAUTHBEARER_SCOPE_CLAIM_NAME);
+ String subClaimName = cu.get(SASL_OAUTHBEARER_SUB_CLAIM_NAME);
+ delegate = new ClientJwtValidator(scopeClaimName, subClaimName);
+ }
+
+ delegate.init();
+ }
+
+ @Override
+ public OAuthBearerToken validate(String accessToken) throws ValidateException {
+ return Objects.requireNonNull(delegate).validate(accessToken);
Review Comment:
Updated to throw an `IllegalStateException` with a better message.
##########
clients/src/test/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerValidatorCallbackHandlerTest.java:
##########
@@ -81,9 +86,65 @@ public void testInvalidAccessToken() throws Exception {
assertInvalidAccessTokenFails(createAccessKey("{}", "{}", "{}"), substring);
}
+ @Test
+ public void testHandlerInitThrowsException() throws IOException {
+ AccessTokenBuilder builder = new AccessTokenBuilder()
+ .alg(AlgorithmIdentifiers.RSA_USING_SHA256);
+ CloseableVerificationKeyResolver verificationKeyResolver = createVerificationKeyResolver(builder);
+ JwtValidator jwtValidator = new JwtValidator() {
+ @Override
+ public void init() throws IOException {
+ throw new IOException("My init had an error!");
+ }
+
+ @Override
+ public OAuthBearerToken validate(String accessToken) throws ValidateException {
+ return null;
+ }
+ };
+
+ OAuthBearerValidatorCallbackHandler handler = new OAuthBearerValidatorCallbackHandler();
+
+ // An error initializing the JwtValidator should cause OAuthBearerValidatorCallbackHandler.init() to fail.
+ assertThrowsWithMessage(
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18847: Refactor OAuth layer to improve reusability 1/N [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19622:
URL: https://github.com/apache/kafka/pull/19622#discussion_r2078299781
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/DefaultJwtValidator.java:
##########
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.common.security.oauthbearer.internals.secured;
+
+import org.apache.kafka.common.security.oauthbearer.OAuthBearerToken;
+import org.apache.kafka.common.utils.Utils;
+
+import org.jose4j.keys.resolvers.VerificationKeyResolver;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_CLOCK_SKEW_SECONDS;
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_EXPECTED_AUDIENCE;
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_EXPECTED_ISSUER;
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_SCOPE_CLAIM_NAME;
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_SUB_CLAIM_NAME;
+
+/**
+ * This {@link JwtValidator} uses the delegation approach, instantiating and delegating calls to a
+ * more concrete implementation. The underlying implementation is determined by the presence/absence
+ * of the {@link VerificationKeyResolver}: if it's present, a {@link BrokerJwtValidator} is
+ * created, otherwise a {@link ClientJwtValidator} is created.
+ */
+public class DefaultJwtValidator implements JwtValidator {
+
+ private final Map<String, ?> configs;
+ private final String saslMechanism;
+ private final Optional<VerificationKeyResolver> verificationKeyResolver;
+
+ private JwtValidator delegate;
+
+ public DefaultJwtValidator(Map<String, ?> configs, String saslMechanism) {
+ this.configs = configs;
+ this.saslMechanism = saslMechanism;
+ this.verificationKeyResolver = Optional.empty();
+ }
+
+ public DefaultJwtValidator(Map<String, ?> configs,
+ String saslMechanism,
+ VerificationKeyResolver verificationKeyResolver) {
+ this.configs = configs;
+ this.saslMechanism = saslMechanism;
+ this.verificationKeyResolver = Optional.of(verificationKeyResolver);
+ }
+
+ @Override
+ public void init() throws IOException {
+ ConfigurationUtils cu = new ConfigurationUtils(configs, saslMechanism);
+
+ if (verificationKeyResolver.isPresent()) {
+ Set<String> expectedAudiences = null;
+ List<String> l = cu.get(SASL_OAUTHBEARER_EXPECTED_AUDIENCE);
+
+ if (l != null)
+ expectedAudiences = Set.copyOf(l);
Review Comment:
Updated. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18847: Refactor OAuth layer to improve reusability 1/N [kafka]
Posted by "omkreddy (via GitHub)" <gi...@apache.org>.
omkreddy commented on code in PR #19622:
URL: https://github.com/apache/kafka/pull/19622#discussion_r2084977349
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerLoginCallbackHandler.java:
##########
@@ -179,55 +180,45 @@ public class OAuthBearerLoginCallbackHandler implements AuthenticateCallbackHand
private Map<String, Object> moduleOptions;
- private AccessTokenRetriever accessTokenRetriever;
+ private JwtRetriever jwtRetriever;
- private AccessTokenValidator accessTokenValidator;
-
- private boolean isInitialized = false;
+ private JwtValidator jwtValidator;
@Override
public void configure(Map<String, ?> configs, String saslMechanism, List<AppConfigurationEntry> jaasConfigEntries) {
- moduleOptions = JaasOptionsUtils.getOptions(saslMechanism, jaasConfigEntries);
- AccessTokenRetriever accessTokenRetriever = AccessTokenRetrieverFactory.create(configs, saslMechanism, moduleOptions);
- AccessTokenValidator accessTokenValidator = AccessTokenValidatorFactory.create(configs, saslMechanism);
- init(accessTokenRetriever, accessTokenValidator);
+ Map<String, Object> moduleOptions = JaasOptionsUtils.getOptions(saslMechanism, jaasConfigEntries);
+ JwtRetriever jwtRetriever = new DefaultJwtRetriever(configs, saslMechanism, moduleOptions);
+ JwtValidator jwtValidator = new DefaultJwtValidator(configs, saslMechanism);
+ configure(moduleOptions, jwtRetriever, jwtValidator);
}
- public void init(AccessTokenRetriever accessTokenRetriever, AccessTokenValidator accessTokenValidator) {
- this.accessTokenRetriever = accessTokenRetriever;
- this.accessTokenValidator = accessTokenValidator;
+ void configure(Map<String, Object> moduleOptions, JwtRetriever jwtRetriever, JwtValidator jwtValidator) {
Review Comment:
why don't we use same `init(` method name
also lets keep below comment:
```
/*
* Package-visible for testing.
*/
```
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerValidatorCallbackHandler.java:
##########
@@ -135,37 +134,36 @@ public void configure(Map<String, ?> configs, String saslMechanism, List<AppConf
new RefCountingVerificationKeyResolver(VerificationKeyResolverFactory.create(configs, saslMechanism, moduleOptions)));
}
- AccessTokenValidator accessTokenValidator = AccessTokenValidatorFactory.create(configs, saslMechanism, verificationKeyResolver);
- init(verificationKeyResolver, accessTokenValidator);
+ JwtValidator jwtValidator = new DefaultJwtValidator(configs, saslMechanism, verificationKeyResolver);
+ configure(verificationKeyResolver, jwtValidator);
}
- public void init(CloseableVerificationKeyResolver verificationKeyResolver, AccessTokenValidator accessTokenValidator) {
+ void configure(CloseableVerificationKeyResolver verificationKeyResolver, JwtValidator jwtValidator) {
Review Comment:
same as above
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18847: Refactor OAuth layer to improve reusability 1/N [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on PR #19622:
URL: https://github.com/apache/kafka/pull/19622#issuecomment-2856455789
Thanks for the review, @lianetm! I've fixed the issues you've mentioned and tested locally, but I'll watch the result of the CI job. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18847: Refactor OAuth layer to improve reusability 1/N [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19622:
URL: https://github.com/apache/kafka/pull/19622#discussion_r2086875757
##########
clients/src/test/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerValidatorCallbackHandlerTest.java:
##########
@@ -106,11 +109,13 @@ public OAuthBearerToken validate(String accessToken) throws ValidateException {
OAuthBearerValidatorCallbackHandler handler = new OAuthBearerValidatorCallbackHandler();
// An error initializing the JwtValidator should cause OAuthBearerValidatorCallbackHandler.init() to fail.
- assertThrowsWithMessage(
+ KafkaException root = assertThrows(
KafkaException.class,
- () -> handler.init(verificationKeyResolver, jwtValidator),
- "encountered an error when initializing"
+ () -> handler.init(verificationKeyResolver, jwtValidator)
);
+ assertNotNull(root);
Review Comment:
Removed the call to `assertNotNull()`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18847: Refactor OAuth layer to improve reusability 1/N [kafka]
Posted by "omkreddy (via GitHub)" <gi...@apache.org>.
omkreddy commented on code in PR #19622:
URL: https://github.com/apache/kafka/pull/19622#discussion_r2080166544
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerLoginCallbackHandler.java:
##########
@@ -179,28 +180,34 @@ public class OAuthBearerLoginCallbackHandler implements AuthenticateCallbackHand
private Map<String, Object> moduleOptions;
- private AccessTokenRetriever accessTokenRetriever;
+ private JwtRetriever jwtRetriever;
- private AccessTokenValidator accessTokenValidator;
+ private JwtValidator jwtValidator;
private boolean isInitialized = false;
@Override
public void configure(Map<String, ?> configs, String saslMechanism, List<AppConfigurationEntry> jaasConfigEntries) {
moduleOptions = JaasOptionsUtils.getOptions(saslMechanism, jaasConfigEntries);
- AccessTokenRetriever accessTokenRetriever = AccessTokenRetrieverFactory.create(configs, saslMechanism, moduleOptions);
- AccessTokenValidator accessTokenValidator = AccessTokenValidatorFactory.create(configs, saslMechanism);
- init(accessTokenRetriever, accessTokenValidator);
+ JwtRetriever jwtRetriever = new DefaultJwtRetriever(configs, saslMechanism, moduleOptions);
+ JwtValidator jwtValidator = new DefaultJwtValidator(configs, saslMechanism);
+ init(jwtRetriever, jwtValidator);
}
- public void init(AccessTokenRetriever accessTokenRetriever, AccessTokenValidator accessTokenValidator) {
- this.accessTokenRetriever = accessTokenRetriever;
- this.accessTokenValidator = accessTokenValidator;
+ public void init(JwtRetriever jwtRetriever, JwtValidator jwtValidator) {
Review Comment:
When we originally added these classes, the methods were package-protected for testing purposes:
[Initial PR](https://github.com/apache/kafka/pull/11284/files#diff-07a4bd9cb223c59540c76fb6a4fab2527de52a70cb4653ab5bb4262cc8ec736cR191-R196)
Later, when the packages were reorganized, the methods were made public:
[Follow-up PR](https://github.com/apache/kafka/pull/12039/files#diff-8feae72c98bbd4064b428d119607c80f5b30136b0e540cf8fa0de08b31321c37R194-R206)
I agree that since these methods are only used in tests and invoked exclusively for testing, itβs reasonable to update or delete them.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18847: Refactor OAuth layer to improve reusability 1/N [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19622:
URL: https://github.com/apache/kafka/pull/19622#discussion_r2076512554
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/HttpJwtRetriever.java:
##########
@@ -49,22 +49,22 @@
import javax.net.ssl.SSLSocketFactory;
/**
- * <code>HttpAccessTokenRetriever</code> is an {@link AccessTokenRetriever} that will
+ * <code>HttpJwtRetriever</code> is an {@link JwtRetriever} that will
* communicate with an OAuth/OIDC provider directly via HTTP to post client credentials
* ({@link OAuthBearerLoginCallbackHandler#CLIENT_ID_CONFIG}/{@link OAuthBearerLoginCallbackHandler#CLIENT_SECRET_CONFIG})
* to a publicized token endpoint URL
* ({@link SaslConfigs#SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL}).
*
- * @see AccessTokenRetriever
+ * @see JwtRetriever
* @see OAuthBearerLoginCallbackHandler#CLIENT_ID_CONFIG
* @see OAuthBearerLoginCallbackHandler#CLIENT_SECRET_CONFIG
* @see OAuthBearerLoginCallbackHandler#SCOPE_CONFIG
* @see SaslConfigs#SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL
Review Comment:
Removed the `@see` annotations.
##########
clients/src/test/java/org/apache/kafka/common/security/oauthbearer/internals/secured/JwtValidatorFactoryTest.java:
##########
@@ -19,39 +19,42 @@
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler;
+import org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule;
import org.junit.jupiter.api.Test;
import java.io.IOException;
import java.util.Map;
-public class AccessTokenValidatorFactoryTest extends OAuthBearerTest {
+public class JwtValidatorFactoryTest extends OAuthBearerTest {
Review Comment:
Moved to `OAuthBearerLoginCallbackHandlerTest`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18847: Refactor OAuth layer to improve reusability 1/N [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #19622:
URL: https://github.com/apache/kafka/pull/19622#discussion_r2077221358
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/VerificationKeyResolverFactory.java:
##########
@@ -37,7 +37,7 @@
public class VerificationKeyResolverFactory {
/**
- * Create an {@link AccessTokenRetriever} from the given
+ * Create an {@link JwtRetriever} from the given
Review Comment:
```suggestion
* Create a {@link JwtRetriever} from the given
```
##########
clients/src/test/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerValidatorCallbackHandlerTest.java:
##########
@@ -81,6 +84,52 @@ public void testInvalidAccessToken() throws Exception {
assertInvalidAccessTokenFails(createAccessKey("{}", "{}", "{}"), substring);
}
+ @Test
+ public void testConfigureThrowsExceptionOnJwtValidatorInit() throws IOException {
+ OAuthBearerLoginCallbackHandler handler = new OAuthBearerLoginCallbackHandler();
+ JwtRetriever jwtRetriever = new JwtRetriever() {
+ @Override
+ public void init() throws IOException {
+ throw new IOException("My init had an error!");
+ }
+
+ @Override
+ public String retrieve() {
+ return "dummy";
+ }
+ };
+
+ Map<String, ?> configs = getSaslConfigs();
+
+ try (JwtValidator jwtValidator = new DefaultJwtValidator(configs, OAuthBearerLoginModule.OAUTHBEARER_MECHANISM)) {
+ assertThrowsWithMessage(
+ KafkaException.class, () -> handler.init(jwtRetriever, jwtValidator), "encountered an error when initializing");
+ }
+ }
+
+ @Test
+ public void testConfigureThrowsExceptionOnJwtValidatorClose() throws IOException {
+ OAuthBearerLoginCallbackHandler handler = new OAuthBearerLoginCallbackHandler();
+ JwtRetriever jwtRetriever = new JwtRetriever() {
+ @Override
+ public void close() throws IOException {
+ throw new IOException("My close had an error!");
+ }
+ @Override
+ public String retrieve() {
+ return "dummy";
+ }
+ };
+
+ Map<String, ?> configs = getSaslConfigs();
+ try (JwtValidator jwtValidator = new DefaultJwtValidator(configs, OAuthBearerLoginModule.OAUTHBEARER_MECHANISM)) {
+ handler.init(jwtRetriever, jwtValidator);
+
+ // Basically asserting this doesn't throw an exception :(
+ handler.close();
Review Comment:
How about use the `assertDoesNotThrow`
```suggestion
assertDoesNotThrow(handler::close);
```
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/DefaultJwtValidator.java:
##########
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.common.security.oauthbearer.internals.secured;
+
+import org.apache.kafka.common.security.oauthbearer.OAuthBearerToken;
+import org.apache.kafka.common.utils.Utils;
+
+import org.jose4j.keys.resolvers.VerificationKeyResolver;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_CLOCK_SKEW_SECONDS;
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_EXPECTED_AUDIENCE;
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_EXPECTED_ISSUER;
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_SCOPE_CLAIM_NAME;
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_SUB_CLAIM_NAME;
+
+/**
+ * This {@link JwtValidator} uses the delegation approach, instantiating and delegating calls to a
+ * more concrete implementation. The underlying implementation is determined by the presence/absence
+ * of the {@link VerificationKeyResolver}: if it's present, a {@link BrokerJwtValidator} is
+ * created, otherwise a {@link ClientJwtValidator} is created.
+ */
+public class DefaultJwtValidator implements JwtValidator {
+
+ private final Map<String, ?> configs;
+ private final String saslMechanism;
+ private final Optional<VerificationKeyResolver> verificationKeyResolver;
+
+ private JwtValidator delegate;
+
+ public DefaultJwtValidator(Map<String, ?> configs, String saslMechanism) {
+ this.configs = configs;
+ this.saslMechanism = saslMechanism;
+ this.verificationKeyResolver = Optional.empty();
+ }
+
+ public DefaultJwtValidator(Map<String, ?> configs,
+ String saslMechanism,
+ VerificationKeyResolver verificationKeyResolver) {
+ this.configs = configs;
+ this.saslMechanism = saslMechanism;
+ this.verificationKeyResolver = Optional.of(verificationKeyResolver);
+ }
+
+ @Override
+ public void init() throws IOException {
+ ConfigurationUtils cu = new ConfigurationUtils(configs, saslMechanism);
+
+ if (verificationKeyResolver.isPresent()) {
+ Set<String> expectedAudiences = null;
+ List<String> l = cu.get(SASL_OAUTHBEARER_EXPECTED_AUDIENCE);
+
+ if (l != null)
+ expectedAudiences = Set.copyOf(l);
Review Comment:
Perhaps we could update the variable name `l` to something more descriptive for better readability.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18847: Refactor OAuth layer to improve reusability 1/N [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #19622:
URL: https://github.com/apache/kafka/pull/19622#discussion_r2085007139
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerValidatorCallbackHandler.java:
##########
@@ -135,37 +134,36 @@ public void configure(Map<String, ?> configs, String saslMechanism, List<AppConf
new RefCountingVerificationKeyResolver(VerificationKeyResolverFactory.create(configs, saslMechanism, moduleOptions)));
}
- AccessTokenValidator accessTokenValidator = AccessTokenValidatorFactory.create(configs, saslMechanism, verificationKeyResolver);
- init(verificationKeyResolver, accessTokenValidator);
+ JwtValidator jwtValidator = new DefaultJwtValidator(configs, saslMechanism, verificationKeyResolver);
+ configure(verificationKeyResolver, jwtValidator);
}
- public void init(CloseableVerificationKeyResolver verificationKeyResolver, AccessTokenValidator accessTokenValidator) {
+ void configure(CloseableVerificationKeyResolver verificationKeyResolver, JwtValidator jwtValidator) {
this.verificationKeyResolver = verificationKeyResolver;
- this.accessTokenValidator = accessTokenValidator;
+ this.jwtValidator = jwtValidator;
try {
verificationKeyResolver.init();
} catch (Exception e) {
- throw new KafkaException("The OAuth validator configuration encountered an error when initializing the VerificationKeyResolver", e);
+ throw new KafkaException("The OAuth validator callback encountered an error when initializing the VerificationKeyResolver", e);
}
- isInitialized = true;
+ try {
+ jwtValidator.init();
+ } catch (IOException e) {
+ throw new KafkaException("The OAuth validator callback encountered an error when initializing the JwtValidator", e);
+ }
}
@Override
public void close() {
- if (verificationKeyResolver != null) {
- try {
- verificationKeyResolver.close();
- } catch (Exception e) {
- log.error(e.getMessage(), e);
- }
- }
+ Utils.closeQuietly(jwtValidator, "The OAuth validator callback encountered an error when closing the JwtValidator");
+ Utils.closeQuietly(verificationKeyResolver, "The OAuth validator callback encountered an error when closing the VerificationKeyResolver");
Review Comment:
ditto
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18847: Refactor OAuth layer to improve reusability 1/N [kafka]
Posted by "omkreddy (via GitHub)" <gi...@apache.org>.
omkreddy commented on code in PR #19622:
URL: https://github.com/apache/kafka/pull/19622#discussion_r2078936871
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/DefaultJwtRetriever.java:
##########
@@ -36,32 +39,27 @@
import static org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler.CLIENT_SECRET_CONFIG;
import static org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler.SCOPE_CONFIG;
-public class AccessTokenRetrieverFactory {
+public class DefaultJwtRetriever implements JwtRetriever {
Review Comment:
Can we add Javadoc?
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerLoginCallbackHandler.java:
##########
@@ -177,57 +178,41 @@ public class OAuthBearerLoginCallbackHandler implements AuthenticateCallbackHand
private static final String EXTENSION_PREFIX = "extension_";
- private Map<String, Object> moduleOptions;
+ protected Map<String, Object> moduleOptions;
Review Comment:
Can we avoid protected in these public classes. maybe we can add package-private getters/etc.. for testing
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18847: Refactor OAuth layer to improve reusability 1/N [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm merged PR #19622:
URL: https://github.com/apache/kafka/pull/19622
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18847: Refactor OAuth layer to improve reusability 1/N [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19622:
URL: https://github.com/apache/kafka/pull/19622#discussion_r2076511435
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/FileJwtRetriever.java:
##########
@@ -23,19 +23,19 @@
import java.nio.file.Path;
/**
- * <code>FileTokenRetriever</code> is an {@link AccessTokenRetriever} that will load the contents,
+ * <code>FileJwtRetriever</code> is an {@link JwtRetriever} that will load the contents,
Review Comment:
Yep, not sure how long that's been missing a key part of the sentence π€¦ββοΈ
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/HttpJwtRetriever.java:
##########
@@ -49,22 +49,22 @@
import javax.net.ssl.SSLSocketFactory;
/**
- * <code>HttpAccessTokenRetriever</code> is an {@link AccessTokenRetriever} that will
+ * <code>HttpJwtRetriever</code> is an {@link JwtRetriever} that will
Review Comment:
Fixed an typo.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18847: Refactor OAuth layer to improve reusability 1/N [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #19622:
URL: https://github.com/apache/kafka/pull/19622#discussion_r2076162526
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerValidatorCallbackHandler.java:
##########
@@ -135,13 +136,19 @@ public void configure(Map<String, ?> configs, String saslMechanism, List<AppConf
new RefCountingVerificationKeyResolver(VerificationKeyResolverFactory.create(configs, saslMechanism, moduleOptions)));
}
- AccessTokenValidator accessTokenValidator = AccessTokenValidatorFactory.create(configs, saslMechanism, verificationKeyResolver);
- init(verificationKeyResolver, accessTokenValidator);
+ JwtValidator jwtValidator = new DefaultJwtValidator(configs, saslMechanism, verificationKeyResolver);
+ init(verificationKeyResolver, jwtValidator);
}
- public void init(CloseableVerificationKeyResolver verificationKeyResolver, AccessTokenValidator accessTokenValidator) {
+ public void init(CloseableVerificationKeyResolver verificationKeyResolver, JwtValidator jwtValidator) {
this.verificationKeyResolver = verificationKeyResolver;
- this.accessTokenValidator = accessTokenValidator;
+ this.jwtValidator = jwtValidator;
+
+ try {
+ this.jwtValidator.init();
+ } catch (IOException e) {
+ throw new KafkaException("The OAuth validator configuration encountered an error when initializing the JwtValidator", e);
Review Comment:
this reads a bit weird, doesn't it? Should it be "The OAuth validator callback..."??
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/ClientJwtValidator.java:
##########
@@ -123,6 +124,16 @@ else if (scopeRaw instanceof Collection)
issuedAt);
}
+ @Override
+ public void init() throws IOException {
+ // Do nothing...
+ }
Review Comment:
as above, needed?
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/FileJwtRetriever.java:
##########
@@ -23,19 +23,19 @@
import java.nio.file.Path;
/**
- * <code>FileTokenRetriever</code> is an {@link AccessTokenRetriever} that will load the contents,
+ * <code>FileJwtRetriever</code> is an {@link JwtRetriever} that will load the contents,
Review Comment:
```suggestion
* <code>FileJwtRetriever</code> is a {@link JwtRetriever} that will load the contents,
```
(and are we missing something there? "contents of a file")
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/DefaultJwtValidator.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.common.security.oauthbearer.internals.secured;
+
+import org.apache.kafka.common.security.oauthbearer.OAuthBearerToken;
+import org.apache.kafka.common.utils.Utils;
+
+import org.jose4j.keys.resolvers.VerificationKeyResolver;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_CLOCK_SKEW_SECONDS;
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_EXPECTED_AUDIENCE;
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_EXPECTED_ISSUER;
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_SCOPE_CLAIM_NAME;
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_SUB_CLAIM_NAME;
+
+public class DefaultJwtValidator implements JwtValidator {
Review Comment:
a java doc here would be helpful I expect? This seems to default to broker or client validators based on the VerificationKeyResolver, expected to be used for LoginCallbacks (client) or ValidatorCallbacks (broker), is my understanding correct?
##########
clients/src/test/java/org/apache/kafka/common/security/oauthbearer/internals/secured/JwtValidatorFactoryTest.java:
##########
@@ -19,39 +19,42 @@
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler;
+import org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule;
import org.junit.jupiter.api.Test;
import java.io.IOException;
import java.util.Map;
-public class AccessTokenValidatorFactoryTest extends OAuthBearerTest {
+public class JwtValidatorFactoryTest extends OAuthBearerTest {
Review Comment:
both test in this class seem to be testing the handler, so shouldn't they belong to `OAuthBearerLoginCallbackHandlerTest`?
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/HttpJwtRetriever.java:
##########
@@ -49,22 +49,22 @@
import javax.net.ssl.SSLSocketFactory;
/**
- * <code>HttpAccessTokenRetriever</code> is an {@link AccessTokenRetriever} that will
+ * <code>HttpJwtRetriever</code> is an {@link JwtRetriever} that will
Review Comment:
```suggestion
* <code>HttpJwtRetriever</code> is a {@link JwtRetriever} that will
```
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/JwtValidator.java:
##########
@@ -40,13 +42,12 @@
* <li><a href="https://datatracker.ietf.org/doc/html/draft-ietf-oauth-access-token-jwt">RFC 6750, Section 2.1</a></li>
* </ul>
*
- * @see LoginAccessTokenValidator A basic AccessTokenValidator used by client-side login
- * authentication
- * @see ValidatorAccessTokenValidator A more robust AccessTokenValidator that is used on the broker
- * to validate the token's contents and verify the signature
+ * @see ClientJwtValidator A basic JwtValidator used by client-side login authentication
+ * @see BrokerJwtValidator A more robust JwtValidator that is used on the broker to validate the token's
+ * contents and verify the signature
*/
-public interface AccessTokenValidator {
+public interface JwtValidator extends Initable, Closeable {
Review Comment:
with this we should update the java doc for the `Initiable` interface (reads that it's for `initialization of the retriever` but now it's also for the validator (I would probably just remove the "what" it initializes, it's really a generic interface)
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/BrokerJwtValidator.java:
##########
@@ -190,6 +191,16 @@ else if (scopeRaw instanceof Collection)
issuedAt);
}
+ @Override
+ public void init() throws IOException {
+ // Do nothing...
+ }
Review Comment:
do we need this? it has a default empty implementation so I would expect we don't (if it's that we expect we'll need it in the future let's just add it when needed I would say)
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/ClientJwtValidator.java:
##########
@@ -50,9 +51,9 @@
* </ol>
Review Comment:
nit: on the line above, seems inconsistent that `scope` and `subject` do not have the code tags as the other claims, intentional?
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/HttpJwtRetriever.java:
##########
@@ -49,22 +49,22 @@
import javax.net.ssl.SSLSocketFactory;
/**
- * <code>HttpAccessTokenRetriever</code> is an {@link AccessTokenRetriever} that will
+ * <code>HttpJwtRetriever</code> is an {@link JwtRetriever} that will
* communicate with an OAuth/OIDC provider directly via HTTP to post client credentials
* ({@link OAuthBearerLoginCallbackHandler#CLIENT_ID_CONFIG}/{@link OAuthBearerLoginCallbackHandler#CLIENT_SECRET_CONFIG})
* to a publicized token endpoint URL
* ({@link SaslConfigs#SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL}).
*
- * @see AccessTokenRetriever
+ * @see JwtRetriever
* @see OAuthBearerLoginCallbackHandler#CLIENT_ID_CONFIG
* @see OAuthBearerLoginCallbackHandler#CLIENT_SECRET_CONFIG
* @see OAuthBearerLoginCallbackHandler#SCOPE_CONFIG
* @see SaslConfigs#SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL
Review Comment:
all these are already linked in this same java doc. Do we get anything from adding them again as `@see`? (the scope_config is the only one not duplicated, relevant?)
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/DefaultJwtRetriever.java:
##########
@@ -96,7 +106,7 @@ public static AccessTokenRetriever create(Map<String, ?> configs,
* <p/>
*
* This utility method ensures that we have a non-{@code null} value to use in the
- * {@link HttpAccessTokenRetriever} constructor.
+ * {@link HttpJwtRetriever} constructor.
*/
static boolean validateUrlencodeHeader(ConfigurationUtils configurationUtils) {
Boolean urlencodeHeader = configurationUtils.validateBoolean(SASL_OAUTHBEARER_HEADER_URLENCODE, false);
Review Comment:
this `validateBoolean` only validates isRequired, so seems useless if we pass `false` (not required). Is there a reason why we need this or `.get` could do?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18847: Refactor OAuth layer to improve reusability 1/N [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19622:
URL: https://github.com/apache/kafka/pull/19622#discussion_r2076510031
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/DefaultJwtRetriever.java:
##########
@@ -96,7 +106,7 @@ public static AccessTokenRetriever create(Map<String, ?> configs,
* <p/>
*
* This utility method ensures that we have a non-{@code null} value to use in the
- * {@link HttpAccessTokenRetriever} constructor.
+ * {@link HttpJwtRetriever} constructor.
*/
static boolean validateUrlencodeHeader(ConfigurationUtils configurationUtils) {
Boolean urlencodeHeader = configurationUtils.validateBoolean(SASL_OAUTHBEARER_HEADER_URLENCODE, false);
Review Comment:
Changed to `validateBoolean()` as suggested.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18847: Refactor OAuth layer to improve reusability 1/N [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19622:
URL: https://github.com/apache/kafka/pull/19622#discussion_r2078204014
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/DefaultJwtValidator.java:
##########
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.common.security.oauthbearer.internals.secured;
+
+import org.apache.kafka.common.security.oauthbearer.OAuthBearerToken;
+import org.apache.kafka.common.utils.Utils;
+
+import org.jose4j.keys.resolvers.VerificationKeyResolver;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_CLOCK_SKEW_SECONDS;
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_EXPECTED_AUDIENCE;
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_EXPECTED_ISSUER;
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_SCOPE_CLAIM_NAME;
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_SUB_CLAIM_NAME;
+
+/**
+ * This {@link JwtValidator} uses the delegation approach, instantiating and delegating calls to a
+ * more concrete implementation. The underlying implementation is determined by the presence/absence
+ * of the {@link VerificationKeyResolver}: if it's present, a {@link BrokerJwtValidator} is
+ * created, otherwise a {@link ClientJwtValidator} is created.
+ */
+public class DefaultJwtValidator implements JwtValidator {
+
+ private final Map<String, ?> configs;
+ private final String saslMechanism;
+ private final Optional<VerificationKeyResolver> verificationKeyResolver;
+
+ private JwtValidator delegate;
+
+ public DefaultJwtValidator(Map<String, ?> configs, String saslMechanism) {
+ this.configs = configs;
+ this.saslMechanism = saslMechanism;
+ this.verificationKeyResolver = Optional.empty();
+ }
+
+ public DefaultJwtValidator(Map<String, ?> configs,
+ String saslMechanism,
+ VerificationKeyResolver verificationKeyResolver) {
+ this.configs = configs;
+ this.saslMechanism = saslMechanism;
+ this.verificationKeyResolver = Optional.of(verificationKeyResolver);
+ }
+
+ @Override
+ public void init() throws IOException {
+ ConfigurationUtils cu = new ConfigurationUtils(configs, saslMechanism);
+
+ if (verificationKeyResolver.isPresent()) {
+ Set<String> expectedAudiences = null;
+ List<String> l = cu.get(SASL_OAUTHBEARER_EXPECTED_AUDIENCE);
+
+ if (l != null)
+ expectedAudiences = Set.copyOf(l);
Review Comment:
π Yes, I'll update it. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18847: Refactor OAuth layer to improve reusability 1/N [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19622:
URL: https://github.com/apache/kafka/pull/19622#discussion_r2078589954
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerLoginCallbackHandler.java:
##########
@@ -179,28 +180,34 @@ public class OAuthBearerLoginCallbackHandler implements AuthenticateCallbackHand
private Map<String, Object> moduleOptions;
- private AccessTokenRetriever accessTokenRetriever;
+ private JwtRetriever jwtRetriever;
- private AccessTokenValidator accessTokenValidator;
+ private JwtValidator jwtValidator;
private boolean isInitialized = false;
@Override
public void configure(Map<String, ?> configs, String saslMechanism, List<AppConfigurationEntry> jaasConfigEntries) {
moduleOptions = JaasOptionsUtils.getOptions(saslMechanism, jaasConfigEntries);
- AccessTokenRetriever accessTokenRetriever = AccessTokenRetrieverFactory.create(configs, saslMechanism, moduleOptions);
- AccessTokenValidator accessTokenValidator = AccessTokenValidatorFactory.create(configs, saslMechanism);
- init(accessTokenRetriever, accessTokenValidator);
+ JwtRetriever jwtRetriever = new DefaultJwtRetriever(configs, saslMechanism, moduleOptions);
+ JwtValidator jwtValidator = new DefaultJwtValidator(configs, saslMechanism);
+ init(jwtRetriever, jwtValidator);
}
- public void init(AccessTokenRetriever accessTokenRetriever, AccessTokenValidator accessTokenValidator) {
- this.accessTokenRetriever = accessTokenRetriever;
- this.accessTokenValidator = accessTokenValidator;
+ public void init(JwtRetriever jwtRetriever, JwtValidator jwtValidator) {
Review Comment:
> So I would say that having this init public is what's wrong in the first place?
This method is only `public` by virtue of the awkwardness of the unit tests. I'd prefer to make the `init()` method package `protected`, or just remove it altogether. The existence of `init()` was not mentioned in the previous `KIP` ([KIP-768](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=186877575)), so my thought is that it can be changed without updating the KIP.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18847: Refactor OAuth layer to improve reusability 1/N [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19622:
URL: https://github.com/apache/kafka/pull/19622#discussion_r2085243145
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerLoginCallbackHandler.java:
##########
@@ -179,55 +180,45 @@ public class OAuthBearerLoginCallbackHandler implements AuthenticateCallbackHand
private Map<String, Object> moduleOptions;
- private AccessTokenRetriever accessTokenRetriever;
+ private JwtRetriever jwtRetriever;
- private AccessTokenValidator accessTokenValidator;
-
- private boolean isInitialized = false;
+ private JwtValidator jwtValidator;
@Override
public void configure(Map<String, ?> configs, String saslMechanism, List<AppConfigurationEntry> jaasConfigEntries) {
- moduleOptions = JaasOptionsUtils.getOptions(saslMechanism, jaasConfigEntries);
- AccessTokenRetriever accessTokenRetriever = AccessTokenRetrieverFactory.create(configs, saslMechanism, moduleOptions);
- AccessTokenValidator accessTokenValidator = AccessTokenValidatorFactory.create(configs, saslMechanism);
- init(accessTokenRetriever, accessTokenValidator);
+ Map<String, Object> moduleOptions = JaasOptionsUtils.getOptions(saslMechanism, jaasConfigEntries);
+ JwtRetriever jwtRetriever = new DefaultJwtRetriever(configs, saslMechanism, moduleOptions);
+ JwtValidator jwtValidator = new DefaultJwtValidator(configs, saslMechanism);
+ configure(moduleOptions, jwtRetriever, jwtValidator);
}
- public void init(AccessTokenRetriever accessTokenRetriever, AccessTokenValidator accessTokenValidator) {
- this.accessTokenRetriever = accessTokenRetriever;
- this.accessTokenValidator = accessTokenValidator;
+ void configure(Map<String, Object> moduleOptions, JwtRetriever jwtRetriever, JwtValidator jwtValidator) {
+ this.moduleOptions = moduleOptions;
+ this.jwtRetriever = jwtRetriever;
+ this.jwtValidator = jwtValidator;
try {
- this.accessTokenRetriever.init();
+ this.jwtRetriever.init();
} catch (IOException e) {
- throw new KafkaException("The OAuth login configuration encountered an error when initializing the AccessTokenRetriever", e);
+ throw new KafkaException("The OAuth login callback encountered an error when initializing the JwtRetriever", e);
}
- isInitialized = true;
- }
-
- /*
- * Package-visible for testing.
- */
-
- AccessTokenRetriever getAccessTokenRetriever() {
- return accessTokenRetriever;
+ try {
+ this.jwtValidator.init();
+ } catch (IOException e) {
+ throw new KafkaException("The OAuth login callback encountered an error when initializing the JwtValidator", e);
+ }
}
@Override
public void close() {
- if (accessTokenRetriever != null) {
- try {
- this.accessTokenRetriever.close();
- } catch (IOException e) {
- log.warn("The OAuth login configuration encountered an error when closing the AccessTokenRetriever", e);
- }
- }
+ Utils.closeQuietly(jwtRetriever, "The OAuth login callback encountered an error when closing the JwtRetriever");
+ Utils.closeQuietly(jwtValidator, "The OAuth login callback encountered an error when closing the JwtValidator");
Review Comment:
I updated the string that was passed in to work correctly with the default error message.
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerValidatorCallbackHandler.java:
##########
@@ -135,37 +134,36 @@ public void configure(Map<String, ?> configs, String saslMechanism, List<AppConf
new RefCountingVerificationKeyResolver(VerificationKeyResolverFactory.create(configs, saslMechanism, moduleOptions)));
}
- AccessTokenValidator accessTokenValidator = AccessTokenValidatorFactory.create(configs, saslMechanism, verificationKeyResolver);
- init(verificationKeyResolver, accessTokenValidator);
+ JwtValidator jwtValidator = new DefaultJwtValidator(configs, saslMechanism, verificationKeyResolver);
+ configure(verificationKeyResolver, jwtValidator);
}
- public void init(CloseableVerificationKeyResolver verificationKeyResolver, AccessTokenValidator accessTokenValidator) {
+ void configure(CloseableVerificationKeyResolver verificationKeyResolver, JwtValidator jwtValidator) {
this.verificationKeyResolver = verificationKeyResolver;
- this.accessTokenValidator = accessTokenValidator;
+ this.jwtValidator = jwtValidator;
try {
verificationKeyResolver.init();
} catch (Exception e) {
- throw new KafkaException("The OAuth validator configuration encountered an error when initializing the VerificationKeyResolver", e);
+ throw new KafkaException("The OAuth validator callback encountered an error when initializing the VerificationKeyResolver", e);
}
- isInitialized = true;
+ try {
+ jwtValidator.init();
+ } catch (IOException e) {
+ throw new KafkaException("The OAuth validator callback encountered an error when initializing the JwtValidator", e);
+ }
}
@Override
public void close() {
- if (verificationKeyResolver != null) {
- try {
- verificationKeyResolver.close();
- } catch (Exception e) {
- log.error(e.getMessage(), e);
- }
- }
+ Utils.closeQuietly(jwtValidator, "The OAuth validator callback encountered an error when closing the JwtValidator");
+ Utils.closeQuietly(verificationKeyResolver, "The OAuth validator callback encountered an error when closing the VerificationKeyResolver");
Review Comment:
I updated the string that was passed in to work correctly with the default error message.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18847: Refactor OAuth layer to improve reusability 1/N [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19622:
URL: https://github.com/apache/kafka/pull/19622#discussion_r2081986003
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerLoginCallbackHandler.java:
##########
@@ -177,57 +178,41 @@ public class OAuthBearerLoginCallbackHandler implements AuthenticateCallbackHand
private static final String EXTENSION_PREFIX = "extension_";
- private Map<String, Object> moduleOptions;
+ protected Map<String, Object> moduleOptions;
Review Comment:
Revised the tests so that access to the private instance variables wasn't necessarily.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18847: Refactor OAuth layer to improve reusability 1/N [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19622:
URL: https://github.com/apache/kafka/pull/19622#discussion_r2081983328
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/DefaultJwtRetriever.java:
##########
@@ -36,32 +39,27 @@
import static org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler.CLIENT_SECRET_CONFIG;
import static org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler.SCOPE_CONFIG;
-public class AccessTokenRetrieverFactory {
+public class DefaultJwtRetriever implements JwtRetriever {
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18847: Refactor OAuth layer to improve reusability 1/N [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19622:
URL: https://github.com/apache/kafka/pull/19622#discussion_r2076507643
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/JwtValidator.java:
##########
@@ -40,13 +42,12 @@
* <li><a href="https://datatracker.ietf.org/doc/html/draft-ietf-oauth-access-token-jwt">RFC 6750, Section 2.1</a></li>
* </ul>
*
- * @see LoginAccessTokenValidator A basic AccessTokenValidator used by client-side login
- * authentication
- * @see ValidatorAccessTokenValidator A more robust AccessTokenValidator that is used on the broker
- * to validate the token's contents and verify the signature
+ * @see ClientJwtValidator A basic JwtValidator used by client-side login authentication
+ * @see BrokerJwtValidator A more robust JwtValidator that is used on the broker to validate the token's
+ * contents and verify the signature
*/
-public interface AccessTokenValidator {
+public interface JwtValidator extends Initable, Closeable {
Review Comment:
I generalized the wording. PLMK if it needs further changes.
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerValidatorCallbackHandler.java:
##########
@@ -135,13 +136,19 @@ public void configure(Map<String, ?> configs, String saslMechanism, List<AppConf
new RefCountingVerificationKeyResolver(VerificationKeyResolverFactory.create(configs, saslMechanism, moduleOptions)));
}
- AccessTokenValidator accessTokenValidator = AccessTokenValidatorFactory.create(configs, saslMechanism, verificationKeyResolver);
- init(verificationKeyResolver, accessTokenValidator);
+ JwtValidator jwtValidator = new DefaultJwtValidator(configs, saslMechanism, verificationKeyResolver);
+ init(verificationKeyResolver, jwtValidator);
}
- public void init(CloseableVerificationKeyResolver verificationKeyResolver, AccessTokenValidator accessTokenValidator) {
+ public void init(CloseableVerificationKeyResolver verificationKeyResolver, JwtValidator jwtValidator) {
this.verificationKeyResolver = verificationKeyResolver;
- this.accessTokenValidator = accessTokenValidator;
+ this.jwtValidator = jwtValidator;
+
+ try {
+ this.jwtValidator.init();
+ } catch (IOException e) {
+ throw new KafkaException("The OAuth validator configuration encountered an error when initializing the JwtValidator", e);
Review Comment:
Changed to "The OAuth validator callback", as suggested.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18847: Refactor OAuth layer to improve reusability 1/N [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19622:
URL: https://github.com/apache/kafka/pull/19622#discussion_r2076510682
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/DefaultJwtValidator.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.common.security.oauthbearer.internals.secured;
+
+import org.apache.kafka.common.security.oauthbearer.OAuthBearerToken;
+import org.apache.kafka.common.utils.Utils;
+
+import org.jose4j.keys.resolvers.VerificationKeyResolver;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_CLOCK_SKEW_SECONDS;
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_EXPECTED_AUDIENCE;
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_EXPECTED_ISSUER;
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_SCOPE_CLAIM_NAME;
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_SUB_CLAIM_NAME;
+
+public class DefaultJwtValidator implements JwtValidator {
Review Comment:
Added clarifying JavaDoc.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18847: Refactor OAuth layer to improve reusability 1/N [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19622:
URL: https://github.com/apache/kafka/pull/19622#discussion_r2076509531
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/ClientJwtValidator.java:
##########
@@ -50,9 +51,9 @@
* </ol>
Review Comment:
Not intentional. I've made them all consistent.
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/ClientJwtValidator.java:
##########
@@ -123,6 +124,16 @@ else if (scopeRaw instanceof Collection)
issuedAt);
}
+ @Override
+ public void init() throws IOException {
+ // Do nothing...
+ }
Review Comment:
Removed unnecessary no-op overrides since `Initable` already has a default, no-op `init()` method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18847: Refactor OAuth layer to improve reusability 1/N [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19622:
URL: https://github.com/apache/kafka/pull/19622#discussion_r2085243526
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerLoginCallbackHandler.java:
##########
@@ -179,55 +180,45 @@ public class OAuthBearerLoginCallbackHandler implements AuthenticateCallbackHand
private Map<String, Object> moduleOptions;
- private AccessTokenRetriever accessTokenRetriever;
+ private JwtRetriever jwtRetriever;
- private AccessTokenValidator accessTokenValidator;
-
- private boolean isInitialized = false;
+ private JwtValidator jwtValidator;
@Override
public void configure(Map<String, ?> configs, String saslMechanism, List<AppConfigurationEntry> jaasConfigEntries) {
- moduleOptions = JaasOptionsUtils.getOptions(saslMechanism, jaasConfigEntries);
- AccessTokenRetriever accessTokenRetriever = AccessTokenRetrieverFactory.create(configs, saslMechanism, moduleOptions);
- AccessTokenValidator accessTokenValidator = AccessTokenValidatorFactory.create(configs, saslMechanism);
- init(accessTokenRetriever, accessTokenValidator);
+ Map<String, Object> moduleOptions = JaasOptionsUtils.getOptions(saslMechanism, jaasConfigEntries);
+ JwtRetriever jwtRetriever = new DefaultJwtRetriever(configs, saslMechanism, moduleOptions);
+ JwtValidator jwtValidator = new DefaultJwtValidator(configs, saslMechanism);
+ configure(moduleOptions, jwtRetriever, jwtValidator);
}
- public void init(AccessTokenRetriever accessTokenRetriever, AccessTokenValidator accessTokenValidator) {
- this.accessTokenRetriever = accessTokenRetriever;
- this.accessTokenValidator = accessTokenValidator;
+ void configure(Map<String, Object> moduleOptions, JwtRetriever jwtRetriever, JwtValidator jwtValidator) {
Review Comment:
Done and done.
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerValidatorCallbackHandler.java:
##########
@@ -135,37 +134,36 @@ public void configure(Map<String, ?> configs, String saslMechanism, List<AppConf
new RefCountingVerificationKeyResolver(VerificationKeyResolverFactory.create(configs, saslMechanism, moduleOptions)));
}
- AccessTokenValidator accessTokenValidator = AccessTokenValidatorFactory.create(configs, saslMechanism, verificationKeyResolver);
- init(verificationKeyResolver, accessTokenValidator);
+ JwtValidator jwtValidator = new DefaultJwtValidator(configs, saslMechanism, verificationKeyResolver);
+ configure(verificationKeyResolver, jwtValidator);
}
- public void init(CloseableVerificationKeyResolver verificationKeyResolver, AccessTokenValidator accessTokenValidator) {
+ void configure(CloseableVerificationKeyResolver verificationKeyResolver, JwtValidator jwtValidator) {
Review Comment:
Done and done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18847: Refactor OAuth layer to improve reusability 1/N [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19622:
URL: https://github.com/apache/kafka/pull/19622#discussion_r2086871765
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/DefaultJwtRetriever.java:
##########
@@ -86,6 +90,18 @@ public static AccessTokenRetriever create(Map<String, ?> configs,
cu.validateInteger(SASL_LOGIN_READ_TIMEOUT_MS, false),
urlencodeHeader);
}
+
+ delegate.init();
+ }
+
+ @Override
+ public String retrieve() throws IOException {
+ return Objects.requireNonNull(delegate).retrieve();
Review Comment:
Updated to throw an `IllegalStateException` with a better message.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18847: Refactor OAuth layer to improve reusability 1/N [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #19622:
URL: https://github.com/apache/kafka/pull/19622#discussion_r2085329910
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/DefaultJwtValidator.java:
##########
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.common.security.oauthbearer.internals.secured;
+
+import org.apache.kafka.common.security.oauthbearer.OAuthBearerToken;
+import org.apache.kafka.common.utils.Utils;
+
+import org.jose4j.keys.resolvers.VerificationKeyResolver;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_CLOCK_SKEW_SECONDS;
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_EXPECTED_AUDIENCE;
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_EXPECTED_ISSUER;
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_SCOPE_CLAIM_NAME;
+import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_SUB_CLAIM_NAME;
+
+/**
+ * This {@link JwtValidator} uses the delegation approach, instantiating and delegating calls to a
+ * more concrete implementation. The underlying implementation is determined by the presence/absence
+ * of the {@link VerificationKeyResolver}: if it's present, a {@link BrokerJwtValidator} is
+ * created, otherwise a {@link ClientJwtValidator} is created.
+ */
+public class DefaultJwtValidator implements JwtValidator {
+
+ private final Map<String, ?> configs;
+ private final String saslMechanism;
+ private final Optional<VerificationKeyResolver> verificationKeyResolver;
+
+ private JwtValidator delegate;
+
+ public DefaultJwtValidator(Map<String, ?> configs, String saslMechanism) {
+ this.configs = configs;
+ this.saslMechanism = saslMechanism;
+ this.verificationKeyResolver = Optional.empty();
+ }
+
+ public DefaultJwtValidator(Map<String, ?> configs,
+ String saslMechanism,
+ VerificationKeyResolver verificationKeyResolver) {
+ this.configs = configs;
+ this.saslMechanism = saslMechanism;
+ this.verificationKeyResolver = Optional.of(verificationKeyResolver);
+ }
+
+ @Override
+ public void init() throws IOException {
+ ConfigurationUtils cu = new ConfigurationUtils(configs, saslMechanism);
+
+ if (verificationKeyResolver.isPresent()) {
+ List<String> expectedAudiencesList = cu.get(SASL_OAUTHBEARER_EXPECTED_AUDIENCE);
+ Set<String> expectedAudiences = expectedAudiencesList != null ? Set.copyOf(expectedAudiencesList) : null;
+ Integer clockSkew = cu.validateInteger(SASL_OAUTHBEARER_CLOCK_SKEW_SECONDS, false);
+ String expectedIssuer = cu.validateString(SASL_OAUTHBEARER_EXPECTED_ISSUER, false);
+ String scopeClaimName = cu.validateString(SASL_OAUTHBEARER_SCOPE_CLAIM_NAME);
+ String subClaimName = cu.validateString(SASL_OAUTHBEARER_SUB_CLAIM_NAME);
+
+ delegate = new BrokerJwtValidator(clockSkew,
+ expectedAudiences,
+ expectedIssuer,
+ verificationKeyResolver.get(),
+ scopeClaimName,
+ subClaimName);
+ } else {
+ String scopeClaimName = cu.get(SASL_OAUTHBEARER_SCOPE_CLAIM_NAME);
+ String subClaimName = cu.get(SASL_OAUTHBEARER_SUB_CLAIM_NAME);
+ delegate = new ClientJwtValidator(scopeClaimName, subClaimName);
+ }
+
+ delegate.init();
+ }
+
+ @Override
+ public OAuthBearerToken validate(String accessToken) throws ValidateException {
+ return Objects.requireNonNull(delegate).validate(accessToken);
Review Comment:
ditto
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/DefaultJwtRetriever.java:
##########
@@ -86,6 +90,18 @@ public static AccessTokenRetriever create(Map<String, ?> configs,
cu.validateInteger(SASL_LOGIN_READ_TIMEOUT_MS, false),
urlencodeHeader);
}
+
+ delegate.init();
+ }
+
+ @Override
+ public String retrieve() throws IOException {
+ return Objects.requireNonNull(delegate).retrieve();
Review Comment:
the NPE on "delegate" here would be really obfuscated, should we add a message to clarify? it really means that the api user should call init() before calling retrieve()
##########
clients/src/test/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerValidatorCallbackHandlerTest.java:
##########
@@ -81,9 +86,65 @@ public void testInvalidAccessToken() throws Exception {
assertInvalidAccessTokenFails(createAccessKey("{}", "{}", "{}"), substring);
}
+ @Test
+ public void testHandlerInitThrowsException() throws IOException {
+ AccessTokenBuilder builder = new AccessTokenBuilder()
+ .alg(AlgorithmIdentifiers.RSA_USING_SHA256);
+ CloseableVerificationKeyResolver verificationKeyResolver = createVerificationKeyResolver(builder);
+ JwtValidator jwtValidator = new JwtValidator() {
+ @Override
+ public void init() throws IOException {
+ throw new IOException("My init had an error!");
+ }
+
+ @Override
+ public OAuthBearerToken validate(String accessToken) throws ValidateException {
+ return null;
+ }
+ };
+
+ OAuthBearerValidatorCallbackHandler handler = new OAuthBearerValidatorCallbackHandler();
+
+ // An error initializing the JwtValidator should cause OAuthBearerValidatorCallbackHandler.init() to fail.
+ assertThrowsWithMessage(
Review Comment:
would it be more robust to simply check that the getCause is the same exception we threw on ln 97? (the substring we're checking is really the same for several errors on init, not only the validator)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18847: Refactor OAuth layer to improve reusability 1/N [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #19622:
URL: https://github.com/apache/kafka/pull/19622#discussion_r2086793268
##########
clients/src/test/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerValidatorCallbackHandlerTest.java:
##########
@@ -106,11 +109,13 @@ public OAuthBearerToken validate(String accessToken) throws ValidateException {
OAuthBearerValidatorCallbackHandler handler = new OAuthBearerValidatorCallbackHandler();
// An error initializing the JwtValidator should cause OAuthBearerValidatorCallbackHandler.init() to fail.
- assertThrowsWithMessage(
+ KafkaException root = assertThrows(
KafkaException.class,
- () -> handler.init(verificationKeyResolver, jwtValidator),
- "encountered an error when initializing"
+ () -> handler.init(verificationKeyResolver, jwtValidator)
);
+ assertNotNull(root);
Review Comment:
I would expect this is not needed (and that `assertThrows` simply fails if nothing is thrown)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18847: Refactor OAuth layer to improve reusability 1/N [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19622:
URL: https://github.com/apache/kafka/pull/19622#discussion_r2076508668
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/BrokerJwtValidator.java:
##########
@@ -190,6 +191,16 @@ else if (scopeRaw instanceof Collection)
issuedAt);
}
+ @Override
+ public void init() throws IOException {
+ // Do nothing...
+ }
Review Comment:
Removed unnecessary no-op overrides since `Initable` already has a default, no-op `init()` method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18847: Refactor OAuth layer to improve reusability 1/N [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #19622:
URL: https://github.com/apache/kafka/pull/19622#discussion_r2085006260
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerLoginCallbackHandler.java:
##########
@@ -179,55 +180,45 @@ public class OAuthBearerLoginCallbackHandler implements AuthenticateCallbackHand
private Map<String, Object> moduleOptions;
- private AccessTokenRetriever accessTokenRetriever;
+ private JwtRetriever jwtRetriever;
- private AccessTokenValidator accessTokenValidator;
-
- private boolean isInitialized = false;
+ private JwtValidator jwtValidator;
@Override
public void configure(Map<String, ?> configs, String saslMechanism, List<AppConfigurationEntry> jaasConfigEntries) {
- moduleOptions = JaasOptionsUtils.getOptions(saslMechanism, jaasConfigEntries);
- AccessTokenRetriever accessTokenRetriever = AccessTokenRetrieverFactory.create(configs, saslMechanism, moduleOptions);
- AccessTokenValidator accessTokenValidator = AccessTokenValidatorFactory.create(configs, saslMechanism);
- init(accessTokenRetriever, accessTokenValidator);
+ Map<String, Object> moduleOptions = JaasOptionsUtils.getOptions(saslMechanism, jaasConfigEntries);
+ JwtRetriever jwtRetriever = new DefaultJwtRetriever(configs, saslMechanism, moduleOptions);
+ JwtValidator jwtValidator = new DefaultJwtValidator(configs, saslMechanism);
+ configure(moduleOptions, jwtRetriever, jwtValidator);
}
- public void init(AccessTokenRetriever accessTokenRetriever, AccessTokenValidator accessTokenValidator) {
- this.accessTokenRetriever = accessTokenRetriever;
- this.accessTokenValidator = accessTokenValidator;
+ void configure(Map<String, Object> moduleOptions, JwtRetriever jwtRetriever, JwtValidator jwtValidator) {
+ this.moduleOptions = moduleOptions;
+ this.jwtRetriever = jwtRetriever;
+ this.jwtValidator = jwtValidator;
try {
- this.accessTokenRetriever.init();
+ this.jwtRetriever.init();
} catch (IOException e) {
- throw new KafkaException("The OAuth login configuration encountered an error when initializing the AccessTokenRetriever", e);
+ throw new KafkaException("The OAuth login callback encountered an error when initializing the JwtRetriever", e);
}
- isInitialized = true;
- }
-
- /*
- * Package-visible for testing.
- */
-
- AccessTokenRetriever getAccessTokenRetriever() {
- return accessTokenRetriever;
+ try {
+ this.jwtValidator.init();
+ } catch (IOException e) {
+ throw new KafkaException("The OAuth login callback encountered an error when initializing the JwtValidator", e);
+ }
}
@Override
public void close() {
- if (accessTokenRetriever != null) {
- try {
- this.accessTokenRetriever.close();
- } catch (IOException e) {
- log.warn("The OAuth login configuration encountered an error when closing the AccessTokenRetriever", e);
- }
- }
+ Utils.closeQuietly(jwtRetriever, "The OAuth login callback encountered an error when closing the JwtRetriever");
+ Utils.closeQuietly(jwtValidator, "The OAuth login callback encountered an error when closing the JwtValidator");
Review Comment:
`Utils.closeQuietly` uses a default error message template, so we should update the string to match that default format.
https://github.com/apache/kafka/blob/62bec20aefcdb94e8a66f5e3f74e40916981912a/clients/src/main/java/org/apache/kafka/common/utils/Utils.java#L1114
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18847: Refactor OAuth layer to improve reusability 1/N [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #19622:
URL: https://github.com/apache/kafka/pull/19622#discussion_r2078400315
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerLoginCallbackHandler.java:
##########
@@ -179,28 +180,34 @@ public class OAuthBearerLoginCallbackHandler implements AuthenticateCallbackHand
private Map<String, Object> moduleOptions;
- private AccessTokenRetriever accessTokenRetriever;
+ private JwtRetriever jwtRetriever;
- private AccessTokenValidator accessTokenValidator;
+ private JwtValidator jwtValidator;
private boolean isInitialized = false;
@Override
public void configure(Map<String, ?> configs, String saslMechanism, List<AppConfigurationEntry> jaasConfigEntries) {
moduleOptions = JaasOptionsUtils.getOptions(saslMechanism, jaasConfigEntries);
- AccessTokenRetriever accessTokenRetriever = AccessTokenRetrieverFactory.create(configs, saslMechanism, moduleOptions);
- AccessTokenValidator accessTokenValidator = AccessTokenValidatorFactory.create(configs, saslMechanism);
- init(accessTokenRetriever, accessTokenValidator);
+ JwtRetriever jwtRetriever = new DefaultJwtRetriever(configs, saslMechanism, moduleOptions);
+ JwtValidator jwtValidator = new DefaultJwtValidator(configs, saslMechanism);
+ init(jwtRetriever, jwtValidator);
}
- public void init(AccessTokenRetriever accessTokenRetriever, AccessTokenValidator accessTokenValidator) {
- this.accessTokenRetriever = accessTokenRetriever;
- this.accessTokenValidator = accessTokenValidator;
+ public void init(JwtRetriever jwtRetriever, JwtValidator jwtValidator) {
Review Comment:
uhm the `OAuthBearerLoginCallbackHandler` class is part of the public API, so changing params in this `public` method would be a breaking change not accounted for in the KIP, right?
It's a tricky situation because the class of the params is not public API (`AccessTokenRetriever`).
This init is only used internally, and I expect is not intended to be called directly ever (only indirectly from the public API `configure`). So I would say that having this `init` public is what's wrong in the first place?
Anyways, fact is that it is public API at the moment (https://kafka.apache.org/40/javadoc/org/apache/kafka/common/security/oauthbearer/OAuthBearerLoginCallbackHandler.html), so we need to amend the KIP I would expect, to remove this init overload from the public API (move to package-private). Since it's based on non-public param types it's probably safe to assume no one should be really using it (but not sure if I'm missing something here). Thoughts?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18847: Refactor OAuth layer to improve reusability 1/N [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #19622:
URL: https://github.com/apache/kafka/pull/19622#discussion_r2081985108
##########
clients/src/main/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerLoginCallbackHandler.java:
##########
@@ -179,28 +180,34 @@ public class OAuthBearerLoginCallbackHandler implements AuthenticateCallbackHand
private Map<String, Object> moduleOptions;
- private AccessTokenRetriever accessTokenRetriever;
+ private JwtRetriever jwtRetriever;
- private AccessTokenValidator accessTokenValidator;
+ private JwtValidator jwtValidator;
private boolean isInitialized = false;
@Override
public void configure(Map<String, ?> configs, String saslMechanism, List<AppConfigurationEntry> jaasConfigEntries) {
moduleOptions = JaasOptionsUtils.getOptions(saslMechanism, jaasConfigEntries);
- AccessTokenRetriever accessTokenRetriever = AccessTokenRetrieverFactory.create(configs, saslMechanism, moduleOptions);
- AccessTokenValidator accessTokenValidator = AccessTokenValidatorFactory.create(configs, saslMechanism);
- init(accessTokenRetriever, accessTokenValidator);
+ JwtRetriever jwtRetriever = new DefaultJwtRetriever(configs, saslMechanism, moduleOptions);
+ JwtValidator jwtValidator = new DefaultJwtValidator(configs, saslMechanism);
+ init(jwtRetriever, jwtValidator);
}
- public void init(AccessTokenRetriever accessTokenRetriever, AccessTokenValidator accessTokenValidator) {
- this.accessTokenRetriever = accessTokenRetriever;
- this.accessTokenValidator = accessTokenValidator;
+ public void init(JwtRetriever jwtRetriever, JwtValidator jwtValidator) {
Review Comment:
Updated them to be package protected.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org