kirktrue commented on code in PR #21483: URL: https://github.com/apache/kafka/pull/21483#discussion_r2881053480
########## clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/assertion/AssertionSupplierFactory.java: ########## @@ -0,0 +1,133 @@ +/* + * 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.assertion; + +import org.apache.kafka.common.security.oauthbearer.JwtRetrieverException; +import org.apache.kafka.common.security.oauthbearer.internals.secured.ConfigurationUtils; +import org.apache.kafka.common.utils.Time; +import org.apache.kafka.common.utils.Utils; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.IOException; +import java.util.Optional; + +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_ALGORITHM; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_FILE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_FILE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_PASSPHRASE; +import static org.apache.kafka.common.security.oauthbearer.internals.secured.assertion.AssertionUtils.layeredAssertionJwtTemplate; + +/** + * Factory class for creating JWT assertion suppliers used in OAuth 2.0 client authentication. + * + * <p> + * This factory supports two methods of obtaining JWT assertions for client authentication: + * <ul> + * <li><b>File-based assertions:</b> Pre-generated JWT assertions read from a file specified by + * {@link org.apache.kafka.common.config.SaslConfigs#SASL_OAUTHBEARER_ASSERTION_FILE}. Review Comment: I can't remember how it works off the top of my head, but can the packaging information be dropped because there's already an `import static` reference to `SASL_OAUTHBEARER_ASSERTION_FILE`? For example: ```suggestion * {@link SASL_OAUTHBEARER_ASSERTION_FILE}. ``` ########## clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/assertion/AssertionSupplierFactory.java: ########## @@ -0,0 +1,133 @@ +/* + * 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.assertion; + +import org.apache.kafka.common.security.oauthbearer.JwtRetrieverException; +import org.apache.kafka.common.security.oauthbearer.internals.secured.ConfigurationUtils; +import org.apache.kafka.common.utils.Time; +import org.apache.kafka.common.utils.Utils; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.IOException; +import java.util.Optional; + +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_ALGORITHM; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_FILE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_FILE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_PASSPHRASE; +import static org.apache.kafka.common.security.oauthbearer.internals.secured.assertion.AssertionUtils.layeredAssertionJwtTemplate; + +/** + * Factory class for creating JWT assertion suppliers used in OAuth 2.0 client authentication. + * + * <p> + * This factory supports two methods of obtaining JWT assertions for client authentication: + * <ul> + * <li><b>File-based assertions:</b> Pre-generated JWT assertions read from a file specified by + * {@link org.apache.kafka.common.config.SaslConfigs#SASL_OAUTHBEARER_ASSERTION_FILE}. + * This is useful for testing or when assertions are managed externally.</li> + * <li><b>Locally-generated assertions:</b> JWTs dynamically created and signed using a private key + * specified by {@link org.apache.kafka.common.config.SaslConfigs#SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_FILE}. + * This is the recommended approach for production use.</li> + * </ul> + * </p> + * + * <p> + * The created supplier can be invoked repeatedly to obtain assertions as needed (for example, when + * refreshing tokens). For file-based assertions, the assertion is cached and reloaded automatically when the file changes on disk. + * For locally-generated assertions, a new assertion with updated timestamps is created on each invocation. + * </p> + */ +public class AssertionSupplierFactory { + private static final Logger LOG = LoggerFactory.getLogger(AssertionSupplierFactory.class); + + /** + * Creates a closeable supplier that provides JWT assertions based on the provided configuration. + * + * <p> + * If {@link org.apache.kafka.common.config.SaslConfigs#SASL_OAUTHBEARER_ASSERTION_FILE} is configured, + * the supplier will read assertions from that file. Otherwise, it will create assertions dynamically + * using the configured private key and algorithm. + * </p> + * + * <p> + * <b>Important:</b> The returned {@link CloseableSupplier} must be closed when no longer needed + * to properly release resources such as file handles and cryptographic resources. + * </p> + * + * @param cu The configuration utilities containing assertion configuration + * @param time The time source for generating timestamps in locally-created assertions Review Comment: I prefer dynamically over locally. ########## clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/assertion/AssertionSupplierFactory.java: ########## @@ -0,0 +1,133 @@ +/* + * 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.assertion; + +import org.apache.kafka.common.security.oauthbearer.JwtRetrieverException; +import org.apache.kafka.common.security.oauthbearer.internals.secured.ConfigurationUtils; +import org.apache.kafka.common.utils.Time; +import org.apache.kafka.common.utils.Utils; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.IOException; +import java.util.Optional; + +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_ALGORITHM; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_FILE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_FILE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_PASSPHRASE; +import static org.apache.kafka.common.security.oauthbearer.internals.secured.assertion.AssertionUtils.layeredAssertionJwtTemplate; + +/** + * Factory class for creating JWT assertion suppliers used in OAuth 2.0 client authentication. + * + * <p> + * This factory supports two methods of obtaining JWT assertions for client authentication: + * <ul> + * <li><b>File-based assertions:</b> Pre-generated JWT assertions read from a file specified by + * {@link org.apache.kafka.common.config.SaslConfigs#SASL_OAUTHBEARER_ASSERTION_FILE}. + * This is useful for testing or when assertions are managed externally.</li> + * <li><b>Locally-generated assertions:</b> JWTs dynamically created and signed using a private key + * specified by {@link org.apache.kafka.common.config.SaslConfigs#SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_FILE}. + * This is the recommended approach for production use.</li> + * </ul> + * </p> + * + * <p> + * The created supplier can be invoked repeatedly to obtain assertions as needed (for example, when + * refreshing tokens). For file-based assertions, the assertion is cached and reloaded automatically when the file changes on disk. + * For locally-generated assertions, a new assertion with updated timestamps is created on each invocation. + * </p> + */ +public class AssertionSupplierFactory { + private static final Logger LOG = LoggerFactory.getLogger(AssertionSupplierFactory.class); + + /** + * Creates a closeable supplier that provides JWT assertions based on the provided configuration. + * + * <p> + * If {@link org.apache.kafka.common.config.SaslConfigs#SASL_OAUTHBEARER_ASSERTION_FILE} is configured, + * the supplier will read assertions from that file. Otherwise, it will create assertions dynamically + * using the configured private key and algorithm. + * </p> + * + * <p> + * <b>Important:</b> The returned {@link CloseableSupplier} must be closed when no longer needed + * to properly release resources such as file handles and cryptographic resources. + * </p> + * + * @param cu The configuration utilities containing assertion configuration + * @param time The time source for generating timestamps in locally-created assertions + * @return A closeable supplier that provides JWT assertion strings when invoked + * @throws org.apache.kafka.common.config.ConfigException if required configuration is missing or invalid + * @throws JwtRetrieverException if assertion creation fails (wrapped in the returned supplier) + */ + public static CloseableSupplier<String> create(ConfigurationUtils cu, Time time) { + AssertionJwtTemplate assertionJwtTemplate; + AssertionCreator assertionCreator; + + if (cu.validateString(SASL_OAUTHBEARER_ASSERTION_FILE, false) != null) { + File assertionFile = cu.validateFile(SASL_OAUTHBEARER_ASSERTION_FILE); + LOG.info("Configuring File based assertion using file: {}", assertionFile.getAbsolutePath()); + assertionCreator = new FileAssertionCreator(assertionFile); + assertionJwtTemplate = new StaticAssertionJwtTemplate(); + } else { + String algorithm = cu.validateString(SASL_OAUTHBEARER_ASSERTION_ALGORITHM); + File privateKeyFile = cu.validateFile(SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_FILE); + Optional<String> passphrase = cu.containsKey(SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_PASSPHRASE) ? + Optional.of(cu.validatePassword(SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_PASSPHRASE)) : + Optional.empty(); + LOG.info("Configuring local assertion creation"); Review Comment: Can we provide some of the non-sensitive information (algorithm and private key file name) in the log? ########## clients/src/test/java/org/apache/kafka/common/security/oauthbearer/internals/secured/assertion/AssertionSupplierFactoryTest.java: ########## @@ -0,0 +1,327 @@ +/* + * 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.assertion; + +import org.apache.kafka.common.security.oauthbearer.JwtRetrieverException; +import org.apache.kafka.common.security.oauthbearer.internals.secured.ConfigurationUtils; +import org.apache.kafka.common.security.oauthbearer.internals.secured.OAuthBearerTest; +import org.apache.kafka.common.utils.MockTime; +import org.apache.kafka.common.utils.Time; + +import org.jose4j.jwt.consumer.JwtConsumer; +import org.jose4j.jwt.consumer.JwtConsumerBuilder; +import org.jose4j.jwt.consumer.JwtContext; +import org.junit.jupiter.api.Test; + +import java.io.File; +import java.security.KeyPair; +import java.security.PublicKey; + +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_ALGORITHM; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_AUD; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_EXP_SECONDS; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_ISS; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_JTI_INCLUDE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_NBF_SECONDS; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_SUB; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_FILE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_FILE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_PASSPHRASE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_TEMPLATE_FILE; +import static org.apache.kafka.test.TestUtils.tempFile; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +public class AssertionSupplierFactoryTest extends OAuthBearerTest { + + /** + * When {@code SASL_OAUTHBEARER_ASSERTION_FILE} is configured, the factory should use file-based + * assertion creation and return the JWT from the file. + */ + @Test + public void testCreateWithAssertionFile() throws Exception { + String expectedJwt = createJwt("test-subject"); + File jwtFile = tempFile(expectedJwt); + + ConfigurationUtils cu = mock(ConfigurationUtils.class); + when(cu.validateString(SASL_OAUTHBEARER_ASSERTION_FILE, false)).thenReturn(jwtFile.getAbsolutePath()); + when(cu.validateFile(SASL_OAUTHBEARER_ASSERTION_FILE)).thenReturn(jwtFile); + + Time time = new MockTime(); + + try (CloseableSupplier<String> supplier = AssertionSupplierFactory.create(cu, time)) { + String assertion = supplier.get(); + assertEquals(expectedJwt, assertion); + } + } + + /** + * When no assertion file is configured, the factory should fall back to locally-generated + * assertions using the private key and signing algorithm. + */ + @Test + public void testCreateWithPrivateKeyFile() throws Exception { + KeyPair keyPair = generateKeyPair(); + File privateKeyFile = generatePrivateKey(keyPair.getPrivate()); + + ConfigurationUtils cu = mockConfigForLocalAssertion(privateKeyFile); + + Time time = new MockTime(); + + try (CloseableSupplier<String> supplier = AssertionSupplierFactory.create(cu, time)) { + String assertion = supplier.get(); + assertNotNull(assertion); + assertValidJwtFormat(assertion); + assertClaims(keyPair.getPublic(), assertion); + } + } + + /** + * When static claims (aud, iss, sub) are configured alongside the private key, + * they should be included in the generated JWT assertion. + */ + @Test + public void testCreateWithPrivateKeyFileAndStaticClaims() throws Exception { + KeyPair keyPair = generateKeyPair(); + File privateKeyFile = generatePrivateKey(keyPair.getPrivate()); + + ConfigurationUtils cu = mockConfigForLocalAssertion(privateKeyFile); + // Override static claims to be present + when(cu.containsKey(SASL_OAUTHBEARER_ASSERTION_CLAIM_AUD)).thenReturn(true); + when(cu.validateString(SASL_OAUTHBEARER_ASSERTION_CLAIM_AUD)).thenReturn("https://auth.example.com";); + when(cu.containsKey(SASL_OAUTHBEARER_ASSERTION_CLAIM_ISS)).thenReturn(true); + when(cu.validateString(SASL_OAUTHBEARER_ASSERTION_CLAIM_ISS)).thenReturn("my-client"); + when(cu.containsKey(SASL_OAUTHBEARER_ASSERTION_CLAIM_SUB)).thenReturn(true); + when(cu.validateString(SASL_OAUTHBEARER_ASSERTION_CLAIM_SUB)).thenReturn("service-account"); Review Comment: Out of curiosity, do these need to be set for this particular test to work? ########## clients/src/test/java/org/apache/kafka/common/security/oauthbearer/internals/secured/assertion/AssertionSupplierFactoryTest.java: ########## @@ -0,0 +1,327 @@ +/* + * 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.assertion; + +import org.apache.kafka.common.security.oauthbearer.JwtRetrieverException; +import org.apache.kafka.common.security.oauthbearer.internals.secured.ConfigurationUtils; +import org.apache.kafka.common.security.oauthbearer.internals.secured.OAuthBearerTest; +import org.apache.kafka.common.utils.MockTime; +import org.apache.kafka.common.utils.Time; + +import org.jose4j.jwt.consumer.JwtConsumer; +import org.jose4j.jwt.consumer.JwtConsumerBuilder; +import org.jose4j.jwt.consumer.JwtContext; +import org.junit.jupiter.api.Test; + +import java.io.File; +import java.security.KeyPair; +import java.security.PublicKey; + +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_ALGORITHM; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_AUD; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_EXP_SECONDS; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_ISS; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_JTI_INCLUDE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_NBF_SECONDS; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_SUB; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_FILE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_FILE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_PASSPHRASE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_TEMPLATE_FILE; +import static org.apache.kafka.test.TestUtils.tempFile; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +public class AssertionSupplierFactoryTest extends OAuthBearerTest { + + /** + * When {@code SASL_OAUTHBEARER_ASSERTION_FILE} is configured, the factory should use file-based + * assertion creation and return the JWT from the file. + */ + @Test + public void testCreateWithAssertionFile() throws Exception { + String expectedJwt = createJwt("test-subject"); + File jwtFile = tempFile(expectedJwt); + + ConfigurationUtils cu = mock(ConfigurationUtils.class); + when(cu.validateString(SASL_OAUTHBEARER_ASSERTION_FILE, false)).thenReturn(jwtFile.getAbsolutePath()); + when(cu.validateFile(SASL_OAUTHBEARER_ASSERTION_FILE)).thenReturn(jwtFile); + + Time time = new MockTime(); + + try (CloseableSupplier<String> supplier = AssertionSupplierFactory.create(cu, time)) { + String assertion = supplier.get(); + assertEquals(expectedJwt, assertion); + } + } + + /** + * When no assertion file is configured, the factory should fall back to locally-generated + * assertions using the private key and signing algorithm. + */ + @Test + public void testCreateWithPrivateKeyFile() throws Exception { + KeyPair keyPair = generateKeyPair(); + File privateKeyFile = generatePrivateKey(keyPair.getPrivate()); + + ConfigurationUtils cu = mockConfigForLocalAssertion(privateKeyFile); + + Time time = new MockTime(); + + try (CloseableSupplier<String> supplier = AssertionSupplierFactory.create(cu, time)) { + String assertion = supplier.get(); + assertNotNull(assertion); + assertValidJwtFormat(assertion); + assertClaims(keyPair.getPublic(), assertion); + } + } + + /** + * When static claims (aud, iss, sub) are configured alongside the private key, + * they should be included in the generated JWT assertion. + */ + @Test + public void testCreateWithPrivateKeyFileAndStaticClaims() throws Exception { + KeyPair keyPair = generateKeyPair(); + File privateKeyFile = generatePrivateKey(keyPair.getPrivate()); + + ConfigurationUtils cu = mockConfigForLocalAssertion(privateKeyFile); + // Override static claims to be present + when(cu.containsKey(SASL_OAUTHBEARER_ASSERTION_CLAIM_AUD)).thenReturn(true); + when(cu.validateString(SASL_OAUTHBEARER_ASSERTION_CLAIM_AUD)).thenReturn("https://auth.example.com";); + when(cu.containsKey(SASL_OAUTHBEARER_ASSERTION_CLAIM_ISS)).thenReturn(true); + when(cu.validateString(SASL_OAUTHBEARER_ASSERTION_CLAIM_ISS)).thenReturn("my-client"); + when(cu.containsKey(SASL_OAUTHBEARER_ASSERTION_CLAIM_SUB)).thenReturn(true); + when(cu.validateString(SASL_OAUTHBEARER_ASSERTION_CLAIM_SUB)).thenReturn("service-account"); + + Time time = new MockTime(); + + try (CloseableSupplier<String> supplier = AssertionSupplierFactory.create(cu, time)) { + String assertion = supplier.get(); + assertNotNull(assertion); + assertValidJwtFormat(assertion); + // Use a consumer that expects the configured audience + assertClaimsWithAudience(keyPair.getPublic(), assertion, "https://auth.example.com";); + } + } + + /** + * When a passphrase is configured, the factory should read it via + * {@code validatePassword} and pass it to the {@code DefaultAssertionCreator}. + * Providing a passphrase for an unencrypted key is an invalid combination and + * should result in a {@code JwtRetrieverException} because the raw key bytes + * cannot be parsed as an {@code EncryptedPrivateKeyInfo} structure. + */ + @Test + public void testCreateWithPassphraseReadsPasswordConfig() throws Exception { + KeyPair keyPair = generateKeyPair(); + File privateKeyFile = generatePrivateKey(keyPair.getPrivate()); + + ConfigurationUtils cu = mockConfigForLocalAssertion(privateKeyFile); + // Enable passphrase config + when(cu.containsKey(SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_PASSPHRASE)).thenReturn(true); + when(cu.validatePassword(SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_PASSPHRASE)).thenReturn("my-passphrase"); + + Time time = new MockTime(); + + // Providing a passphrase for an unencrypted key triggers an error during key loading + assertThrows(JwtRetrieverException.class, () -> AssertionSupplierFactory.create(cu, time)); Review Comment: It might be helpful to capture the returned exception of the `assertThrows()` call and perform some more verification on it because `JwtRetrieverException` is a bit of a catch-all. ########## clients/src/test/java/org/apache/kafka/common/security/oauthbearer/internals/secured/assertion/AssertionSupplierFactoryTest.java: ########## @@ -0,0 +1,327 @@ +/* + * 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.assertion; + +import org.apache.kafka.common.security.oauthbearer.JwtRetrieverException; +import org.apache.kafka.common.security.oauthbearer.internals.secured.ConfigurationUtils; +import org.apache.kafka.common.security.oauthbearer.internals.secured.OAuthBearerTest; +import org.apache.kafka.common.utils.MockTime; +import org.apache.kafka.common.utils.Time; + +import org.jose4j.jwt.consumer.JwtConsumer; +import org.jose4j.jwt.consumer.JwtConsumerBuilder; +import org.jose4j.jwt.consumer.JwtContext; +import org.junit.jupiter.api.Test; + +import java.io.File; +import java.security.KeyPair; +import java.security.PublicKey; + +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_ALGORITHM; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_AUD; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_EXP_SECONDS; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_ISS; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_JTI_INCLUDE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_NBF_SECONDS; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_SUB; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_FILE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_FILE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_PASSPHRASE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_TEMPLATE_FILE; +import static org.apache.kafka.test.TestUtils.tempFile; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +public class AssertionSupplierFactoryTest extends OAuthBearerTest { + + /** + * When {@code SASL_OAUTHBEARER_ASSERTION_FILE} is configured, the factory should use file-based + * assertion creation and return the JWT from the file. + */ + @Test + public void testCreateWithAssertionFile() throws Exception { + String expectedJwt = createJwt("test-subject"); + File jwtFile = tempFile(expectedJwt); + + ConfigurationUtils cu = mock(ConfigurationUtils.class); + when(cu.validateString(SASL_OAUTHBEARER_ASSERTION_FILE, false)).thenReturn(jwtFile.getAbsolutePath()); + when(cu.validateFile(SASL_OAUTHBEARER_ASSERTION_FILE)).thenReturn(jwtFile); + + Time time = new MockTime(); + + try (CloseableSupplier<String> supplier = AssertionSupplierFactory.create(cu, time)) { + String assertion = supplier.get(); + assertEquals(expectedJwt, assertion); + } + } + + /** + * When no assertion file is configured, the factory should fall back to locally-generated + * assertions using the private key and signing algorithm. + */ + @Test + public void testCreateWithPrivateKeyFile() throws Exception { + KeyPair keyPair = generateKeyPair(); + File privateKeyFile = generatePrivateKey(keyPair.getPrivate()); + + ConfigurationUtils cu = mockConfigForLocalAssertion(privateKeyFile); + + Time time = new MockTime(); + + try (CloseableSupplier<String> supplier = AssertionSupplierFactory.create(cu, time)) { + String assertion = supplier.get(); + assertNotNull(assertion); + assertValidJwtFormat(assertion); + assertClaims(keyPair.getPublic(), assertion); + } + } + + /** + * When static claims (aud, iss, sub) are configured alongside the private key, + * they should be included in the generated JWT assertion. + */ + @Test + public void testCreateWithPrivateKeyFileAndStaticClaims() throws Exception { + KeyPair keyPair = generateKeyPair(); + File privateKeyFile = generatePrivateKey(keyPair.getPrivate()); + + ConfigurationUtils cu = mockConfigForLocalAssertion(privateKeyFile); + // Override static claims to be present + when(cu.containsKey(SASL_OAUTHBEARER_ASSERTION_CLAIM_AUD)).thenReturn(true); + when(cu.validateString(SASL_OAUTHBEARER_ASSERTION_CLAIM_AUD)).thenReturn("https://auth.example.com";); + when(cu.containsKey(SASL_OAUTHBEARER_ASSERTION_CLAIM_ISS)).thenReturn(true); + when(cu.validateString(SASL_OAUTHBEARER_ASSERTION_CLAIM_ISS)).thenReturn("my-client"); + when(cu.containsKey(SASL_OAUTHBEARER_ASSERTION_CLAIM_SUB)).thenReturn(true); + when(cu.validateString(SASL_OAUTHBEARER_ASSERTION_CLAIM_SUB)).thenReturn("service-account"); + + Time time = new MockTime(); + + try (CloseableSupplier<String> supplier = AssertionSupplierFactory.create(cu, time)) { + String assertion = supplier.get(); + assertNotNull(assertion); + assertValidJwtFormat(assertion); + // Use a consumer that expects the configured audience + assertClaimsWithAudience(keyPair.getPublic(), assertion, "https://auth.example.com";); + } + } + + /** + * When a passphrase is configured, the factory should read it via + * {@code validatePassword} and pass it to the {@code DefaultAssertionCreator}. + * Providing a passphrase for an unencrypted key is an invalid combination and + * should result in a {@code JwtRetrieverException} because the raw key bytes + * cannot be parsed as an {@code EncryptedPrivateKeyInfo} structure. + */ + @Test + public void testCreateWithPassphraseReadsPasswordConfig() throws Exception { + KeyPair keyPair = generateKeyPair(); + File privateKeyFile = generatePrivateKey(keyPair.getPrivate()); + + ConfigurationUtils cu = mockConfigForLocalAssertion(privateKeyFile); + // Enable passphrase config + when(cu.containsKey(SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_PASSPHRASE)).thenReturn(true); + when(cu.validatePassword(SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_PASSPHRASE)).thenReturn("my-passphrase"); + + Time time = new MockTime(); + + // Providing a passphrase for an unencrypted key triggers an error during key loading + assertThrows(JwtRetrieverException.class, () -> AssertionSupplierFactory.create(cu, time)); + + // Verify that the password config was read from configuration + verify(cu).validatePassword(SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_PASSPHRASE); + } + + /** + * When the passphrase config key is absent, the factory should not attempt to read + * a password and should produce a working supplier with the unencrypted private key. + */ + @Test + public void testCreateWithoutPassphraseDoesNotReadPasswordConfig() throws Exception { + KeyPair keyPair = generateKeyPair(); + File privateKeyFile = generatePrivateKey(keyPair.getPrivate()); + + ConfigurationUtils cu = mockConfigForLocalAssertion(privateKeyFile); + // Passphrase not configured (already the default from mockConfigForLocalAssertion) + + Time time = new MockTime(); + + try (CloseableSupplier<String> supplier = AssertionSupplierFactory.create(cu, time)) { + String assertion = supplier.get(); + assertNotNull(assertion); + assertClaims(keyPair.getPublic(), assertion); + } + + // Verify that validatePassword was never called + verify(cu, never()).validatePassword(SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_PASSPHRASE); + } + + /** + * The file-based supplier should return the same value on consecutive calls since + * the file contents haven't changed. + */ + @Test + public void testFileBasedSupplierReturnsSameValueOnMultipleCalls() throws Exception { + String expectedJwt = createJwt("repeated-calls"); + File jwtFile = tempFile(expectedJwt); + + ConfigurationUtils cu = mock(ConfigurationUtils.class); + when(cu.validateString(SASL_OAUTHBEARER_ASSERTION_FILE, false)).thenReturn(jwtFile.getAbsolutePath()); + when(cu.validateFile(SASL_OAUTHBEARER_ASSERTION_FILE)).thenReturn(jwtFile); + + Time time = new MockTime(); + + try (CloseableSupplier<String> supplier = AssertionSupplierFactory.create(cu, time)) { + String first = supplier.get(); + String second = supplier.get(); + assertEquals(first, second); + assertEquals(expectedJwt, first); + } + } + + /** + * The locally-generated supplier should produce different assertions on each call + * since timestamps (iat, exp, nbf) are regenerated. + */ + @Test + public void testLocallyGeneratedSupplierProducesConsistentAssertions() throws Exception { + KeyPair keyPair = generateKeyPair(); + File privateKeyFile = generatePrivateKey(keyPair.getPrivate()); + + ConfigurationUtils cu = mockConfigForLocalAssertion(privateKeyFile); + Time time = new MockTime(); + + try (CloseableSupplier<String> supplier = AssertionSupplierFactory.create(cu, time)) { + String first = supplier.get(); + String second = supplier.get(); + assertNotNull(first); + assertNotNull(second); + // Both should be valid JWTs + assertClaims(keyPair.getPublic(), first); + assertClaims(keyPair.getPublic(), second); + } + } + + /** + * When the underlying file is deleted after factory creation, calling get() should + * throw a JwtRetrieverException since the assertion creator cannot read the file. + */ + @Test + public void testSupplierGetWrapsExceptionInJwtRetrieverException() throws Exception { + String jwt = createJwt("will-be-deleted"); + File jwtFile = tempFile(jwt); + + ConfigurationUtils cu = mock(ConfigurationUtils.class); + when(cu.validateString(SASL_OAUTHBEARER_ASSERTION_FILE, false)).thenReturn(jwtFile.getAbsolutePath()); + when(cu.validateFile(SASL_OAUTHBEARER_ASSERTION_FILE)).thenReturn(jwtFile); + + Time time = new MockTime(); + + try (CloseableSupplier<String> supplier = AssertionSupplierFactory.create(cu, time)) { + // Delete the file so the read will fail + assertTrue(jwtFile.delete()); + assertThrows(JwtRetrieverException.class, supplier::get); + } + } + + /** + * Closing the supplier should not throw any exceptions. + */ + @Test + public void testSupplierCloseDoesNotThrow() throws Exception { + String expectedJwt = createJwt("close-test"); + File jwtFile = tempFile(expectedJwt); + + ConfigurationUtils cu = mock(ConfigurationUtils.class); + when(cu.validateString(SASL_OAUTHBEARER_ASSERTION_FILE, false)).thenReturn(jwtFile.getAbsolutePath()); + when(cu.validateFile(SASL_OAUTHBEARER_ASSERTION_FILE)).thenReturn(jwtFile); + + Time time = new MockTime(); + + CloseableSupplier<String> supplier = AssertionSupplierFactory.create(cu, time); + // Verify close does not throw + assertDoesNotThrow(supplier::close); + } + + /** + * After closing the supplier, the underlying resources should be released. + * Verify that the supplier can be closed multiple times without error. + */ + @Test + public void testSupplierCanBeClosedMultipleTimes() throws Exception { + String expectedJwt = createJwt("multi-close"); + File jwtFile = tempFile(expectedJwt); + + ConfigurationUtils cu = mock(ConfigurationUtils.class); + when(cu.validateString(SASL_OAUTHBEARER_ASSERTION_FILE, false)).thenReturn(jwtFile.getAbsolutePath()); + when(cu.validateFile(SASL_OAUTHBEARER_ASSERTION_FILE)).thenReturn(jwtFile); + + Time time = new MockTime(); + + CloseableSupplier<String> supplier = AssertionSupplierFactory.create(cu, time); + assertDoesNotThrow(supplier::close); + assertDoesNotThrow(supplier::close); + } + + private void assertValidJwtFormat(String jwt) { + String[] parts = jwt.split("\\."); + assertEquals(3, parts.length, "JWT should have 3 parts (header.payload.signature), but was: " + jwt); Review Comment: This is very rudimentary validation. Is that all that's needed? If so, maybe a comment to that effect to allay concerns. ########## clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/assertion/AssertionSupplierFactory.java: ########## @@ -0,0 +1,133 @@ +/* + * 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.assertion; + +import org.apache.kafka.common.security.oauthbearer.JwtRetrieverException; +import org.apache.kafka.common.security.oauthbearer.internals.secured.ConfigurationUtils; +import org.apache.kafka.common.utils.Time; +import org.apache.kafka.common.utils.Utils; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.IOException; +import java.util.Optional; + +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_ALGORITHM; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_FILE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_FILE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_PASSPHRASE; +import static org.apache.kafka.common.security.oauthbearer.internals.secured.assertion.AssertionUtils.layeredAssertionJwtTemplate; + +/** + * Factory class for creating JWT assertion suppliers used in OAuth 2.0 client authentication. + * + * <p> + * This factory supports two methods of obtaining JWT assertions for client authentication: + * <ul> + * <li><b>File-based assertions:</b> Pre-generated JWT assertions read from a file specified by + * {@link org.apache.kafka.common.config.SaslConfigs#SASL_OAUTHBEARER_ASSERTION_FILE}. + * This is useful for testing or when assertions are managed externally.</li> + * <li><b>Locally-generated assertions:</b> JWTs dynamically created and signed using a private key Review Comment: "Locally" is a bit confusing. Local to what? ########## clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/assertion/DefaultAssertionCreator.java: ########## @@ -85,7 +85,10 @@ public PrivateKey transform(File file, String contents) { try { contents = contents.replace("-----BEGIN PRIVATE KEY-----", "") .replace("-----END PRIVATE KEY-----", "") - .replace("\n", ""); + .replace("-----BEGIN ENCRYPTED PRIVATE KEY-----", "") + .replace("-----END ENCRYPTED PRIVATE KEY-----", "") + .replace("\n", "") + .replace("\r", ""); Review Comment: Does it makes sense for this replacement logic to live in `AssertionUtils`? ########## clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/assertion/AssertionSupplierFactory.java: ########## @@ -0,0 +1,133 @@ +/* + * 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.assertion; + +import org.apache.kafka.common.security.oauthbearer.JwtRetrieverException; +import org.apache.kafka.common.security.oauthbearer.internals.secured.ConfigurationUtils; +import org.apache.kafka.common.utils.Time; +import org.apache.kafka.common.utils.Utils; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.IOException; +import java.util.Optional; + +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_ALGORITHM; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_FILE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_FILE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_PASSPHRASE; +import static org.apache.kafka.common.security.oauthbearer.internals.secured.assertion.AssertionUtils.layeredAssertionJwtTemplate; + +/** + * Factory class for creating JWT assertion suppliers used in OAuth 2.0 client authentication. + * + * <p> + * This factory supports two methods of obtaining JWT assertions for client authentication: + * <ul> + * <li><b>File-based assertions:</b> Pre-generated JWT assertions read from a file specified by + * {@link org.apache.kafka.common.config.SaslConfigs#SASL_OAUTHBEARER_ASSERTION_FILE}. + * This is useful for testing or when assertions are managed externally.</li> + * <li><b>Locally-generated assertions:</b> JWTs dynamically created and signed using a private key + * specified by {@link org.apache.kafka.common.config.SaslConfigs#SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_FILE}. + * This is the recommended approach for production use.</li> + * </ul> + * </p> + * + * <p> + * The created supplier can be invoked repeatedly to obtain assertions as needed (for example, when + * refreshing tokens). For file-based assertions, the assertion is cached and reloaded automatically when the file changes on disk. + * For locally-generated assertions, a new assertion with updated timestamps is created on each invocation. + * </p> + */ +public class AssertionSupplierFactory { + private static final Logger LOG = LoggerFactory.getLogger(AssertionSupplierFactory.class); + + /** + * Creates a closeable supplier that provides JWT assertions based on the provided configuration. + * + * <p> + * If {@link org.apache.kafka.common.config.SaslConfigs#SASL_OAUTHBEARER_ASSERTION_FILE} is configured, + * the supplier will read assertions from that file. Otherwise, it will create assertions dynamically + * using the configured private key and algorithm. + * </p> + * + * <p> + * <b>Important:</b> The returned {@link CloseableSupplier} must be closed when no longer needed + * to properly release resources such as file handles and cryptographic resources. Review Comment: Just out of curiosity, what file handles do we need to close explicitly? ########## clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/ClientSecretRequestFormatter.java: ########## @@ -28,7 +28,36 @@ import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_CLIENT_CREDENTIALS_CLIENT_ID; import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_CLIENT_CREDENTIALS_CLIENT_SECRET; -public class ClientCredentialsRequestFormatter implements HttpRequestFormatter { +/** + * {@code ClientSecretRequestFormatter} is an {@link HttpRequestFormatter} that formats HTTP requests + * for OAuth 2.0 token requests using the <code>client_credentials</code> grant type with client secret + * authentication. + * + * <p> + * This formatter implements the traditional OAuth 2.0 client authentication method where the client + * authenticates using a client ID and client secret. The credentials are sent in the HTTP Authorization + * header using HTTP Basic authentication (Base64-encoded). + * </p> + * + * <p> + * The formatter creates HTTP requests with: + * <ul> + * <li>Authorization header: Basic <base64(clientId:clientSecret)></li> + * <li>Content-Type: application/x-www-form-urlencoded</li> + * <li>grant_type=client_credentials</li> + * <li>Optional scope parameter</li> + * </ul> + * </p> + * + * <p> + * This class was renamed from {@code ClientCredentialsRequestFormatter} to better distinguish it from + * client assertion-based authentication ({@link ClientAssertionRequestFormatter}). + * </p> + * + * @see ClientAssertionRequestFormatter + * @see <a href="https://datatracker.ietf.org/doc/html/rfc6749#section-2.3.1";>RFC 6749 Section 2.3.1: Client Password</a> + */ +public class ClientSecretRequestFormatter implements HttpRequestFormatter { Review Comment: Again, thanks for adding comments! ########## clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/assertion/AssertionSupplierFactory.java: ########## @@ -0,0 +1,133 @@ +/* + * 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.assertion; + +import org.apache.kafka.common.security.oauthbearer.JwtRetrieverException; +import org.apache.kafka.common.security.oauthbearer.internals.secured.ConfigurationUtils; +import org.apache.kafka.common.utils.Time; +import org.apache.kafka.common.utils.Utils; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.IOException; +import java.util.Optional; + +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_ALGORITHM; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_FILE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_FILE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_PASSPHRASE; +import static org.apache.kafka.common.security.oauthbearer.internals.secured.assertion.AssertionUtils.layeredAssertionJwtTemplate; + +/** + * Factory class for creating JWT assertion suppliers used in OAuth 2.0 client authentication. + * + * <p> + * This factory supports two methods of obtaining JWT assertions for client authentication: + * <ul> + * <li><b>File-based assertions:</b> Pre-generated JWT assertions read from a file specified by + * {@link org.apache.kafka.common.config.SaslConfigs#SASL_OAUTHBEARER_ASSERTION_FILE}. + * This is useful for testing or when assertions are managed externally.</li> + * <li><b>Locally-generated assertions:</b> JWTs dynamically created and signed using a private key + * specified by {@link org.apache.kafka.common.config.SaslConfigs#SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_FILE}. + * This is the recommended approach for production use.</li> + * </ul> + * </p> + * + * <p> + * The created supplier can be invoked repeatedly to obtain assertions as needed (for example, when + * refreshing tokens). For file-based assertions, the assertion is cached and reloaded automatically when the file changes on disk. + * For locally-generated assertions, a new assertion with updated timestamps is created on each invocation. + * </p> + */ +public class AssertionSupplierFactory { + private static final Logger LOG = LoggerFactory.getLogger(AssertionSupplierFactory.class); + + /** + * Creates a closeable supplier that provides JWT assertions based on the provided configuration. + * + * <p> + * If {@link org.apache.kafka.common.config.SaslConfigs#SASL_OAUTHBEARER_ASSERTION_FILE} is configured, + * the supplier will read assertions from that file. Otherwise, it will create assertions dynamically + * using the configured private key and algorithm. + * </p> + * + * <p> + * <b>Important:</b> The returned {@link CloseableSupplier} must be closed when no longer needed + * to properly release resources such as file handles and cryptographic resources. + * </p> + * + * @param cu The configuration utilities containing assertion configuration + * @param time The time source for generating timestamps in locally-created assertions + * @return A closeable supplier that provides JWT assertion strings when invoked + * @throws org.apache.kafka.common.config.ConfigException if required configuration is missing or invalid Review Comment: ```suggestion * @throws ConfigException if required configuration is missing or invalid ``` ########## clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/assertion/AssertionUtils.java: ########## @@ -64,30 +64,41 @@ public class AssertionUtils { */ public static PrivateKey privateKey(byte[] privateKeyContents, Optional<String> passphrase) throws GeneralSecurityException, IOException { + byte[] derEncodedBytes = Base64.getDecoder().decode(privateKeyContents); PKCS8EncodedKeySpec keySpec; - if (passphrase.isPresent()) { - EncryptedPrivateKeyInfo keyInfo = new EncryptedPrivateKeyInfo(privateKeyContents); + EncryptedPrivateKeyInfo keyInfo = new EncryptedPrivateKeyInfo(derEncodedBytes); String algorithm = keyInfo.getAlgName(); SecretKeyFactory secretKeyFactory = SecretKeyFactory.getInstance(algorithm); SecretKey pbeKey = secretKeyFactory.generateSecret(new PBEKeySpec(passphrase.get().toCharArray())); Cipher cipher = Cipher.getInstance(algorithm); cipher.init(Cipher.DECRYPT_MODE, pbeKey, keyInfo.getAlgParameters()); keySpec = keyInfo.getKeySpec(cipher); } else { - byte[] pkcs8EncodedBytes = Base64.getDecoder().decode(privateKeyContents); - keySpec = new PKCS8EncodedKeySpec(pkcs8EncodedBytes); + keySpec = new PKCS8EncodedKeySpec(derEncodedBytes); + } + + // Try RSA first, then EC. PKCS#8 encoded keys are algorithm-agnostic, so we need to + // attempt multiple key factories to determine the correct key type. + for (String keyAlgorithm : new String[]{"RSA", "EC"}) { Review Comment: Don't we already know the algorithm in use from the configuration? Can we refactor this method to pass in the algorithm so we don't need to try to guess it? ########## clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/assertion/AssertionUtils.java: ########## @@ -64,30 +64,41 @@ public class AssertionUtils { */ public static PrivateKey privateKey(byte[] privateKeyContents, Optional<String> passphrase) throws GeneralSecurityException, IOException { + byte[] derEncodedBytes = Base64.getDecoder().decode(privateKeyContents); Review Comment: What does the "der" prefix in the `derEncodedBytes` variable name refer to? ########## clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/ClientAssertionRequestFormatter.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.internals.secured.assertion.CloseableSupplier; +import org.apache.kafka.common.utils.Utils; + +import java.io.Closeable; +import java.io.IOException; +import java.net.URLEncoder; +import java.nio.charset.StandardCharsets; +import java.util.HashMap; +import java.util.Map; + +/** + * {@code ClientAssertionRequestFormatter} is an {@link HttpRequestFormatter} that formats HTTP requests + * for OAuth 2.0 token requests using the <code>client_credentials</code> grant type with client assertion + * authentication (RFC 7521 and RFC 7523). + * + * <p> + * This formatter is used when authenticating to an OAuth provider using a client assertion (JWT) + * instead of a client secret. The assertion is typically a self-signed JWT that proves the client's + * identity. This method is more secure than client secrets as the assertion is short-lived and the + * private key used to sign it never leaves the client. + * </p> + * + * <p> + * The formatter creates HTTP requests with: + * <ul> + * <li>Content-Type: application/x-www-form-urlencoded</li> + * <li>grant_type=client_credentials</li> + * <li>client_assertion_type=urn:ietf:params:oauth:client-assertion-type:jwt-bearer</li> + * <li>client_assertion=<JWT assertion></li> + * <li>Optional client_id parameter</li> + * <li>Optional scope parameter</li> + * </ul> + * </p> + * + * @see ClientSecretRequestFormatter + * @see <a href="https://datatracker.ietf.org/doc/html/rfc7521";>RFC 7521: Assertion Framework for OAuth 2.0</a> + * @see <a href="https://datatracker.ietf.org/doc/html/rfc7523";>RFC 7523: JWT Profile for OAuth 2.0 Client Authentication</a> + */ +public class ClientAssertionRequestFormatter implements HttpRequestFormatter, Closeable { + public static final String GRANT_TYPE = "client_credentials"; + private final String clientId; + private final String scope; + private final CloseableSupplier<String> assertionSupplier; + + /** + * Creates a new {@code ClientAssertionRequestFormatter} instance. + * + * @param clientId The OAuth client ID; may be {@code null} if the assertion contains the client ID + * @param scope The OAuth scope to request; may be {@code null} for default scope + * @param assertionSupplier A closeable supplier that provides the JWT assertion string when invoked; + * this formatter takes ownership and will close it when {@link #close()} is called + */ + public ClientAssertionRequestFormatter(String clientId, String scope, CloseableSupplier<String> assertionSupplier) { + this.clientId = clientId; + this.scope = scope; + this.assertionSupplier = assertionSupplier; + } + + @Override + public Map<String, String> formatHeaders() { + Map<String, String> headers = new HashMap<>(); + headers.put("Accept", "application/json"); + headers.put("Cache-Control", "no-cache"); + headers.put("Content-Type", "application/x-www-form-urlencoded"); + return headers; + } + + @Override + public String formatBody() { + StringBuilder requestParameters = new StringBuilder(); + requestParameters.append("client_assertion_type=").append("urn%3Aietf%3Aparams%3Aoauth%3Aclient-assertion-type%3Ajwt-bearer"); + requestParameters.append("&client_assertion=").append(assertionSupplier.get()); + requestParameters.append("&grant_type=").append(URLEncoder.encode(GRANT_TYPE, StandardCharsets.UTF_8)); Review Comment: I forget now why `grant_type` and `scope` need to be URL encoded but `client_id` doesn't 🤔 Granted, I assume this is based on the code that I wrote. Can you add some comments here about URL encoding and any encoding needed for the assertion itself? ########## clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/ClientAssertionRequestFormatter.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.internals.secured.assertion.CloseableSupplier; +import org.apache.kafka.common.utils.Utils; + +import java.io.Closeable; +import java.io.IOException; +import java.net.URLEncoder; +import java.nio.charset.StandardCharsets; +import java.util.HashMap; +import java.util.Map; + +/** + * {@code ClientAssertionRequestFormatter} is an {@link HttpRequestFormatter} that formats HTTP requests + * for OAuth 2.0 token requests using the <code>client_credentials</code> grant type with client assertion + * authentication (RFC 7521 and RFC 7523). + * + * <p> + * This formatter is used when authenticating to an OAuth provider using a client assertion (JWT) + * instead of a client secret. The assertion is typically a self-signed JWT that proves the client's + * identity. This method is more secure than client secrets as the assertion is short-lived and the + * private key used to sign it never leaves the client. + * </p> + * + * <p> + * The formatter creates HTTP requests with: + * <ul> + * <li>Content-Type: application/x-www-form-urlencoded</li> + * <li>grant_type=client_credentials</li> + * <li>client_assertion_type=urn:ietf:params:oauth:client-assertion-type:jwt-bearer</li> + * <li>client_assertion=<JWT assertion></li> + * <li>Optional client_id parameter</li> + * <li>Optional scope parameter</li> + * </ul> + * </p> + * + * @see ClientSecretRequestFormatter + * @see <a href="https://datatracker.ietf.org/doc/html/rfc7521";>RFC 7521: Assertion Framework for OAuth 2.0</a> + * @see <a href="https://datatracker.ietf.org/doc/html/rfc7523";>RFC 7523: JWT Profile for OAuth 2.0 Client Authentication</a> + */ Review Comment: Nice comments! ########## clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/ConfigOrJaas.java: ########## @@ -0,0 +1,195 @@ +/* + * 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.config.ConfigException; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.function.Function; + +import static org.apache.kafka.common.config.SaslConfigs.SASL_JAAS_CONFIG; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_CLIENT_CREDENTIALS_CLIENT_ID; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_CLIENT_CREDENTIALS_CLIENT_SECRET; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_SCOPE; +import static org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler.CLIENT_ID_CONFIG; +import static org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler.CLIENT_SECRET_CONFIG; +import static org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler.SCOPE_CONFIG; + +/** + * Utility class that retrieves OAuth configuration values from either modern configuration properties + * or legacy JAAS options, providing a unified interface with proper deprecation handling. + * + * <p> + * Kafka historically configured OAuth parameters (client ID, client secret, scope) via JAAS options. + * These have been deprecated in favor of dedicated configuration properties. This class: + * <ul> + * <li>Checks modern configuration first</li> + * <li>Falls back to JAAS options if not found (with deprecation warnings)</li> + * <li>Throws {@link ConfigException} if required values are missing from both sources</li> + * <li>Logs warnings when both are present (modern config takes precedence)</li> + * </ul> + * </p> + * + * <p> + * This approach maintains backward compatibility while guiding users toward the recommended + * configuration method. + * </p> + * + * @see ConfigurationUtils + * @see JaasOptionsUtils Review Comment: Thanks for adding comments 😄 ########## clients/src/test/java/org/apache/kafka/common/security/oauthbearer/internals/secured/assertion/AssertionSupplierFactoryTest.java: ########## @@ -0,0 +1,327 @@ +/* + * 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.assertion; + +import org.apache.kafka.common.security.oauthbearer.JwtRetrieverException; +import org.apache.kafka.common.security.oauthbearer.internals.secured.ConfigurationUtils; +import org.apache.kafka.common.security.oauthbearer.internals.secured.OAuthBearerTest; +import org.apache.kafka.common.utils.MockTime; +import org.apache.kafka.common.utils.Time; + +import org.jose4j.jwt.consumer.JwtConsumer; +import org.jose4j.jwt.consumer.JwtConsumerBuilder; +import org.jose4j.jwt.consumer.JwtContext; +import org.junit.jupiter.api.Test; + +import java.io.File; +import java.security.KeyPair; +import java.security.PublicKey; + +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_ALGORITHM; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_AUD; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_EXP_SECONDS; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_ISS; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_JTI_INCLUDE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_NBF_SECONDS; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_SUB; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_FILE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_FILE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_PASSPHRASE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_TEMPLATE_FILE; +import static org.apache.kafka.test.TestUtils.tempFile; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +public class AssertionSupplierFactoryTest extends OAuthBearerTest { + + /** + * When {@code SASL_OAUTHBEARER_ASSERTION_FILE} is configured, the factory should use file-based + * assertion creation and return the JWT from the file. + */ + @Test + public void testCreateWithAssertionFile() throws Exception { + String expectedJwt = createJwt("test-subject"); + File jwtFile = tempFile(expectedJwt); + + ConfigurationUtils cu = mock(ConfigurationUtils.class); + when(cu.validateString(SASL_OAUTHBEARER_ASSERTION_FILE, false)).thenReturn(jwtFile.getAbsolutePath()); + when(cu.validateFile(SASL_OAUTHBEARER_ASSERTION_FILE)).thenReturn(jwtFile); + + Time time = new MockTime(); + + try (CloseableSupplier<String> supplier = AssertionSupplierFactory.create(cu, time)) { + String assertion = supplier.get(); + assertEquals(expectedJwt, assertion); + } + } + + /** + * When no assertion file is configured, the factory should fall back to locally-generated + * assertions using the private key and signing algorithm. + */ + @Test + public void testCreateWithPrivateKeyFile() throws Exception { + KeyPair keyPair = generateKeyPair(); + File privateKeyFile = generatePrivateKey(keyPair.getPrivate()); + + ConfigurationUtils cu = mockConfigForLocalAssertion(privateKeyFile); + + Time time = new MockTime(); + + try (CloseableSupplier<String> supplier = AssertionSupplierFactory.create(cu, time)) { + String assertion = supplier.get(); + assertNotNull(assertion); + assertValidJwtFormat(assertion); + assertClaims(keyPair.getPublic(), assertion); + } + } + + /** + * When static claims (aud, iss, sub) are configured alongside the private key, + * they should be included in the generated JWT assertion. + */ + @Test + public void testCreateWithPrivateKeyFileAndStaticClaims() throws Exception { + KeyPair keyPair = generateKeyPair(); + File privateKeyFile = generatePrivateKey(keyPair.getPrivate()); + + ConfigurationUtils cu = mockConfigForLocalAssertion(privateKeyFile); + // Override static claims to be present + when(cu.containsKey(SASL_OAUTHBEARER_ASSERTION_CLAIM_AUD)).thenReturn(true); + when(cu.validateString(SASL_OAUTHBEARER_ASSERTION_CLAIM_AUD)).thenReturn("https://auth.example.com";); + when(cu.containsKey(SASL_OAUTHBEARER_ASSERTION_CLAIM_ISS)).thenReturn(true); + when(cu.validateString(SASL_OAUTHBEARER_ASSERTION_CLAIM_ISS)).thenReturn("my-client"); + when(cu.containsKey(SASL_OAUTHBEARER_ASSERTION_CLAIM_SUB)).thenReturn(true); + when(cu.validateString(SASL_OAUTHBEARER_ASSERTION_CLAIM_SUB)).thenReturn("service-account"); + + Time time = new MockTime(); + + try (CloseableSupplier<String> supplier = AssertionSupplierFactory.create(cu, time)) { + String assertion = supplier.get(); + assertNotNull(assertion); + assertValidJwtFormat(assertion); + // Use a consumer that expects the configured audience + assertClaimsWithAudience(keyPair.getPublic(), assertion, "https://auth.example.com";); + } + } + + /** + * When a passphrase is configured, the factory should read it via + * {@code validatePassword} and pass it to the {@code DefaultAssertionCreator}. + * Providing a passphrase for an unencrypted key is an invalid combination and + * should result in a {@code JwtRetrieverException} because the raw key bytes + * cannot be parsed as an {@code EncryptedPrivateKeyInfo} structure. + */ + @Test + public void testCreateWithPassphraseReadsPasswordConfig() throws Exception { + KeyPair keyPair = generateKeyPair(); + File privateKeyFile = generatePrivateKey(keyPair.getPrivate()); + + ConfigurationUtils cu = mockConfigForLocalAssertion(privateKeyFile); + // Enable passphrase config + when(cu.containsKey(SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_PASSPHRASE)).thenReturn(true); + when(cu.validatePassword(SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_PASSPHRASE)).thenReturn("my-passphrase"); + + Time time = new MockTime(); + + // Providing a passphrase for an unencrypted key triggers an error during key loading + assertThrows(JwtRetrieverException.class, () -> AssertionSupplierFactory.create(cu, time)); + + // Verify that the password config was read from configuration + verify(cu).validatePassword(SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_PASSPHRASE); + } + + /** + * When the passphrase config key is absent, the factory should not attempt to read + * a password and should produce a working supplier with the unencrypted private key. + */ + @Test + public void testCreateWithoutPassphraseDoesNotReadPasswordConfig() throws Exception { + KeyPair keyPair = generateKeyPair(); + File privateKeyFile = generatePrivateKey(keyPair.getPrivate()); + + ConfigurationUtils cu = mockConfigForLocalAssertion(privateKeyFile); + // Passphrase not configured (already the default from mockConfigForLocalAssertion) + + Time time = new MockTime(); + + try (CloseableSupplier<String> supplier = AssertionSupplierFactory.create(cu, time)) { + String assertion = supplier.get(); + assertNotNull(assertion); + assertClaims(keyPair.getPublic(), assertion); + } + + // Verify that validatePassword was never called + verify(cu, never()).validatePassword(SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_PASSPHRASE); + } + + /** + * The file-based supplier should return the same value on consecutive calls since + * the file contents haven't changed. + */ + @Test + public void testFileBasedSupplierReturnsSameValueOnMultipleCalls() throws Exception { + String expectedJwt = createJwt("repeated-calls"); + File jwtFile = tempFile(expectedJwt); + + ConfigurationUtils cu = mock(ConfigurationUtils.class); + when(cu.validateString(SASL_OAUTHBEARER_ASSERTION_FILE, false)).thenReturn(jwtFile.getAbsolutePath()); + when(cu.validateFile(SASL_OAUTHBEARER_ASSERTION_FILE)).thenReturn(jwtFile); + + Time time = new MockTime(); + + try (CloseableSupplier<String> supplier = AssertionSupplierFactory.create(cu, time)) { + String first = supplier.get(); + String second = supplier.get(); + assertEquals(first, second); + assertEquals(expectedJwt, first); + } + } + + /** + * The locally-generated supplier should produce different assertions on each call + * since timestamps (iat, exp, nbf) are regenerated. + */ + @Test + public void testLocallyGeneratedSupplierProducesConsistentAssertions() throws Exception { + KeyPair keyPair = generateKeyPair(); + File privateKeyFile = generatePrivateKey(keyPair.getPrivate()); + + ConfigurationUtils cu = mockConfigForLocalAssertion(privateKeyFile); + Time time = new MockTime(); + + try (CloseableSupplier<String> supplier = AssertionSupplierFactory.create(cu, time)) { + String first = supplier.get(); + String second = supplier.get(); Review Comment: Is there any way this test could be potentially flaky? There's a way to ensure that subsequent calls to `MockTime` methods produce different results, but I can't remember how 🤔 ########## clients/src/test/java/org/apache/kafka/common/security/oauthbearer/internals/secured/assertion/AssertionSupplierFactoryTest.java: ########## @@ -0,0 +1,327 @@ +/* + * 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.assertion; + +import org.apache.kafka.common.security.oauthbearer.JwtRetrieverException; +import org.apache.kafka.common.security.oauthbearer.internals.secured.ConfigurationUtils; +import org.apache.kafka.common.security.oauthbearer.internals.secured.OAuthBearerTest; +import org.apache.kafka.common.utils.MockTime; +import org.apache.kafka.common.utils.Time; + +import org.jose4j.jwt.consumer.JwtConsumer; +import org.jose4j.jwt.consumer.JwtConsumerBuilder; +import org.jose4j.jwt.consumer.JwtContext; +import org.junit.jupiter.api.Test; + +import java.io.File; +import java.security.KeyPair; +import java.security.PublicKey; + +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_ALGORITHM; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_AUD; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_EXP_SECONDS; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_ISS; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_JTI_INCLUDE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_NBF_SECONDS; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_SUB; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_FILE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_FILE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_PASSPHRASE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_TEMPLATE_FILE; +import static org.apache.kafka.test.TestUtils.tempFile; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +public class AssertionSupplierFactoryTest extends OAuthBearerTest { + + /** + * When {@code SASL_OAUTHBEARER_ASSERTION_FILE} is configured, the factory should use file-based + * assertion creation and return the JWT from the file. + */ + @Test + public void testCreateWithAssertionFile() throws Exception { + String expectedJwt = createJwt("test-subject"); + File jwtFile = tempFile(expectedJwt); + + ConfigurationUtils cu = mock(ConfigurationUtils.class); + when(cu.validateString(SASL_OAUTHBEARER_ASSERTION_FILE, false)).thenReturn(jwtFile.getAbsolutePath()); + when(cu.validateFile(SASL_OAUTHBEARER_ASSERTION_FILE)).thenReturn(jwtFile); + + Time time = new MockTime(); + + try (CloseableSupplier<String> supplier = AssertionSupplierFactory.create(cu, time)) { + String assertion = supplier.get(); + assertEquals(expectedJwt, assertion); + } + } + + /** + * When no assertion file is configured, the factory should fall back to locally-generated + * assertions using the private key and signing algorithm. + */ + @Test + public void testCreateWithPrivateKeyFile() throws Exception { + KeyPair keyPair = generateKeyPair(); + File privateKeyFile = generatePrivateKey(keyPair.getPrivate()); + + ConfigurationUtils cu = mockConfigForLocalAssertion(privateKeyFile); + + Time time = new MockTime(); + + try (CloseableSupplier<String> supplier = AssertionSupplierFactory.create(cu, time)) { + String assertion = supplier.get(); + assertNotNull(assertion); + assertValidJwtFormat(assertion); + assertClaims(keyPair.getPublic(), assertion); + } + } + + /** + * When static claims (aud, iss, sub) are configured alongside the private key, + * they should be included in the generated JWT assertion. + */ + @Test + public void testCreateWithPrivateKeyFileAndStaticClaims() throws Exception { + KeyPair keyPair = generateKeyPair(); + File privateKeyFile = generatePrivateKey(keyPair.getPrivate()); + + ConfigurationUtils cu = mockConfigForLocalAssertion(privateKeyFile); + // Override static claims to be present + when(cu.containsKey(SASL_OAUTHBEARER_ASSERTION_CLAIM_AUD)).thenReturn(true); + when(cu.validateString(SASL_OAUTHBEARER_ASSERTION_CLAIM_AUD)).thenReturn("https://auth.example.com";); + when(cu.containsKey(SASL_OAUTHBEARER_ASSERTION_CLAIM_ISS)).thenReturn(true); + when(cu.validateString(SASL_OAUTHBEARER_ASSERTION_CLAIM_ISS)).thenReturn("my-client"); + when(cu.containsKey(SASL_OAUTHBEARER_ASSERTION_CLAIM_SUB)).thenReturn(true); + when(cu.validateString(SASL_OAUTHBEARER_ASSERTION_CLAIM_SUB)).thenReturn("service-account"); + + Time time = new MockTime(); + + try (CloseableSupplier<String> supplier = AssertionSupplierFactory.create(cu, time)) { + String assertion = supplier.get(); + assertNotNull(assertion); + assertValidJwtFormat(assertion); + // Use a consumer that expects the configured audience + assertClaimsWithAudience(keyPair.getPublic(), assertion, "https://auth.example.com";); + } + } + + /** + * When a passphrase is configured, the factory should read it via + * {@code validatePassword} and pass it to the {@code DefaultAssertionCreator}. + * Providing a passphrase for an unencrypted key is an invalid combination and + * should result in a {@code JwtRetrieverException} because the raw key bytes + * cannot be parsed as an {@code EncryptedPrivateKeyInfo} structure. + */ + @Test + public void testCreateWithPassphraseReadsPasswordConfig() throws Exception { + KeyPair keyPair = generateKeyPair(); + File privateKeyFile = generatePrivateKey(keyPair.getPrivate()); + + ConfigurationUtils cu = mockConfigForLocalAssertion(privateKeyFile); + // Enable passphrase config + when(cu.containsKey(SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_PASSPHRASE)).thenReturn(true); + when(cu.validatePassword(SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_PASSPHRASE)).thenReturn("my-passphrase"); + + Time time = new MockTime(); + + // Providing a passphrase for an unencrypted key triggers an error during key loading + assertThrows(JwtRetrieverException.class, () -> AssertionSupplierFactory.create(cu, time)); + + // Verify that the password config was read from configuration + verify(cu).validatePassword(SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_PASSPHRASE); + } + + /** + * When the passphrase config key is absent, the factory should not attempt to read + * a password and should produce a working supplier with the unencrypted private key. + */ + @Test + public void testCreateWithoutPassphraseDoesNotReadPasswordConfig() throws Exception { + KeyPair keyPair = generateKeyPair(); + File privateKeyFile = generatePrivateKey(keyPair.getPrivate()); + + ConfigurationUtils cu = mockConfigForLocalAssertion(privateKeyFile); + // Passphrase not configured (already the default from mockConfigForLocalAssertion) + + Time time = new MockTime(); + + try (CloseableSupplier<String> supplier = AssertionSupplierFactory.create(cu, time)) { + String assertion = supplier.get(); + assertNotNull(assertion); + assertClaims(keyPair.getPublic(), assertion); + } + + // Verify that validatePassword was never called + verify(cu, never()).validatePassword(SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_PASSPHRASE); + } + + /** + * The file-based supplier should return the same value on consecutive calls since + * the file contents haven't changed. + */ + @Test + public void testFileBasedSupplierReturnsSameValueOnMultipleCalls() throws Exception { + String expectedJwt = createJwt("repeated-calls"); + File jwtFile = tempFile(expectedJwt); + + ConfigurationUtils cu = mock(ConfigurationUtils.class); + when(cu.validateString(SASL_OAUTHBEARER_ASSERTION_FILE, false)).thenReturn(jwtFile.getAbsolutePath()); + when(cu.validateFile(SASL_OAUTHBEARER_ASSERTION_FILE)).thenReturn(jwtFile); + + Time time = new MockTime(); + + try (CloseableSupplier<String> supplier = AssertionSupplierFactory.create(cu, time)) { + String first = supplier.get(); + String second = supplier.get(); + assertEquals(first, second); + assertEquals(expectedJwt, first); + } + } + + /** + * The locally-generated supplier should produce different assertions on each call + * since timestamps (iat, exp, nbf) are regenerated. + */ + @Test + public void testLocallyGeneratedSupplierProducesConsistentAssertions() throws Exception { + KeyPair keyPair = generateKeyPair(); + File privateKeyFile = generatePrivateKey(keyPair.getPrivate()); + + ConfigurationUtils cu = mockConfigForLocalAssertion(privateKeyFile); + Time time = new MockTime(); + + try (CloseableSupplier<String> supplier = AssertionSupplierFactory.create(cu, time)) { + String first = supplier.get(); + String second = supplier.get(); + assertNotNull(first); + assertNotNull(second); + // Both should be valid JWTs + assertClaims(keyPair.getPublic(), first); + assertClaims(keyPair.getPublic(), second); + } + } + + /** + * When the underlying file is deleted after factory creation, calling get() should + * throw a JwtRetrieverException since the assertion creator cannot read the file. + */ + @Test + public void testSupplierGetWrapsExceptionInJwtRetrieverException() throws Exception { + String jwt = createJwt("will-be-deleted"); + File jwtFile = tempFile(jwt); + + ConfigurationUtils cu = mock(ConfigurationUtils.class); + when(cu.validateString(SASL_OAUTHBEARER_ASSERTION_FILE, false)).thenReturn(jwtFile.getAbsolutePath()); + when(cu.validateFile(SASL_OAUTHBEARER_ASSERTION_FILE)).thenReturn(jwtFile); + + Time time = new MockTime(); + + try (CloseableSupplier<String> supplier = AssertionSupplierFactory.create(cu, time)) { + // Delete the file so the read will fail + assertTrue(jwtFile.delete()); Review Comment: Sinister! 😄 ########## clients/src/main/java/org/apache/kafka/common/security/oauthbearer/JwtBearerJwtRetriever.java: ########## Review Comment: Did we want to add the log line for the delegate creation similar to what was added in `ClientCredentialsJwtRetriever`? ########## clients/src/test/java/org/apache/kafka/common/security/oauthbearer/internals/secured/assertion/AssertionSupplierFactoryTest.java: ########## @@ -0,0 +1,327 @@ +/* + * 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.assertion; + +import org.apache.kafka.common.security.oauthbearer.JwtRetrieverException; +import org.apache.kafka.common.security.oauthbearer.internals.secured.ConfigurationUtils; +import org.apache.kafka.common.security.oauthbearer.internals.secured.OAuthBearerTest; +import org.apache.kafka.common.utils.MockTime; +import org.apache.kafka.common.utils.Time; + +import org.jose4j.jwt.consumer.JwtConsumer; +import org.jose4j.jwt.consumer.JwtConsumerBuilder; +import org.jose4j.jwt.consumer.JwtContext; +import org.junit.jupiter.api.Test; + +import java.io.File; +import java.security.KeyPair; +import java.security.PublicKey; + +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_ALGORITHM; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_AUD; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_EXP_SECONDS; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_ISS; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_JTI_INCLUDE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_NBF_SECONDS; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_SUB; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_FILE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_FILE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_PASSPHRASE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_TEMPLATE_FILE; +import static org.apache.kafka.test.TestUtils.tempFile; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +public class AssertionSupplierFactoryTest extends OAuthBearerTest { + + /** + * When {@code SASL_OAUTHBEARER_ASSERTION_FILE} is configured, the factory should use file-based + * assertion creation and return the JWT from the file. + */ + @Test + public void testCreateWithAssertionFile() throws Exception { + String expectedJwt = createJwt("test-subject"); + File jwtFile = tempFile(expectedJwt); + + ConfigurationUtils cu = mock(ConfigurationUtils.class); + when(cu.validateString(SASL_OAUTHBEARER_ASSERTION_FILE, false)).thenReturn(jwtFile.getAbsolutePath()); + when(cu.validateFile(SASL_OAUTHBEARER_ASSERTION_FILE)).thenReturn(jwtFile); + + Time time = new MockTime(); + + try (CloseableSupplier<String> supplier = AssertionSupplierFactory.create(cu, time)) { + String assertion = supplier.get(); + assertEquals(expectedJwt, assertion); + } + } + + /** + * When no assertion file is configured, the factory should fall back to locally-generated + * assertions using the private key and signing algorithm. + */ + @Test + public void testCreateWithPrivateKeyFile() throws Exception { + KeyPair keyPair = generateKeyPair(); + File privateKeyFile = generatePrivateKey(keyPair.getPrivate()); + + ConfigurationUtils cu = mockConfigForLocalAssertion(privateKeyFile); + + Time time = new MockTime(); + + try (CloseableSupplier<String> supplier = AssertionSupplierFactory.create(cu, time)) { + String assertion = supplier.get(); + assertNotNull(assertion); + assertValidJwtFormat(assertion); + assertClaims(keyPair.getPublic(), assertion); + } + } + + /** + * When static claims (aud, iss, sub) are configured alongside the private key, + * they should be included in the generated JWT assertion. + */ + @Test + public void testCreateWithPrivateKeyFileAndStaticClaims() throws Exception { + KeyPair keyPair = generateKeyPair(); + File privateKeyFile = generatePrivateKey(keyPair.getPrivate()); + + ConfigurationUtils cu = mockConfigForLocalAssertion(privateKeyFile); + // Override static claims to be present + when(cu.containsKey(SASL_OAUTHBEARER_ASSERTION_CLAIM_AUD)).thenReturn(true); + when(cu.validateString(SASL_OAUTHBEARER_ASSERTION_CLAIM_AUD)).thenReturn("https://auth.example.com";); + when(cu.containsKey(SASL_OAUTHBEARER_ASSERTION_CLAIM_ISS)).thenReturn(true); + when(cu.validateString(SASL_OAUTHBEARER_ASSERTION_CLAIM_ISS)).thenReturn("my-client"); + when(cu.containsKey(SASL_OAUTHBEARER_ASSERTION_CLAIM_SUB)).thenReturn(true); + when(cu.validateString(SASL_OAUTHBEARER_ASSERTION_CLAIM_SUB)).thenReturn("service-account"); + + Time time = new MockTime(); + + try (CloseableSupplier<String> supplier = AssertionSupplierFactory.create(cu, time)) { + String assertion = supplier.get(); + assertNotNull(assertion); + assertValidJwtFormat(assertion); + // Use a consumer that expects the configured audience + assertClaimsWithAudience(keyPair.getPublic(), assertion, "https://auth.example.com";); + } + } + + /** + * When a passphrase is configured, the factory should read it via + * {@code validatePassword} and pass it to the {@code DefaultAssertionCreator}. + * Providing a passphrase for an unencrypted key is an invalid combination and + * should result in a {@code JwtRetrieverException} because the raw key bytes + * cannot be parsed as an {@code EncryptedPrivateKeyInfo} structure. + */ + @Test + public void testCreateWithPassphraseReadsPasswordConfig() throws Exception { + KeyPair keyPair = generateKeyPair(); + File privateKeyFile = generatePrivateKey(keyPair.getPrivate()); + + ConfigurationUtils cu = mockConfigForLocalAssertion(privateKeyFile); + // Enable passphrase config + when(cu.containsKey(SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_PASSPHRASE)).thenReturn(true); + when(cu.validatePassword(SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_PASSPHRASE)).thenReturn("my-passphrase"); + + Time time = new MockTime(); + + // Providing a passphrase for an unencrypted key triggers an error during key loading + assertThrows(JwtRetrieverException.class, () -> AssertionSupplierFactory.create(cu, time)); + + // Verify that the password config was read from configuration + verify(cu).validatePassword(SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_PASSPHRASE); + } + + /** + * When the passphrase config key is absent, the factory should not attempt to read + * a password and should produce a working supplier with the unencrypted private key. + */ + @Test + public void testCreateWithoutPassphraseDoesNotReadPasswordConfig() throws Exception { + KeyPair keyPair = generateKeyPair(); + File privateKeyFile = generatePrivateKey(keyPair.getPrivate()); + + ConfigurationUtils cu = mockConfigForLocalAssertion(privateKeyFile); + // Passphrase not configured (already the default from mockConfigForLocalAssertion) + + Time time = new MockTime(); + + try (CloseableSupplier<String> supplier = AssertionSupplierFactory.create(cu, time)) { + String assertion = supplier.get(); + assertNotNull(assertion); + assertClaims(keyPair.getPublic(), assertion); + } + + // Verify that validatePassword was never called + verify(cu, never()).validatePassword(SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_PASSPHRASE); + } + + /** + * The file-based supplier should return the same value on consecutive calls since + * the file contents haven't changed. + */ + @Test + public void testFileBasedSupplierReturnsSameValueOnMultipleCalls() throws Exception { + String expectedJwt = createJwt("repeated-calls"); + File jwtFile = tempFile(expectedJwt); + + ConfigurationUtils cu = mock(ConfigurationUtils.class); + when(cu.validateString(SASL_OAUTHBEARER_ASSERTION_FILE, false)).thenReturn(jwtFile.getAbsolutePath()); + when(cu.validateFile(SASL_OAUTHBEARER_ASSERTION_FILE)).thenReturn(jwtFile); + + Time time = new MockTime(); + + try (CloseableSupplier<String> supplier = AssertionSupplierFactory.create(cu, time)) { + String first = supplier.get(); + String second = supplier.get(); + assertEquals(first, second); + assertEquals(expectedJwt, first); + } + } + + /** + * The locally-generated supplier should produce different assertions on each call + * since timestamps (iat, exp, nbf) are regenerated. + */ + @Test + public void testLocallyGeneratedSupplierProducesConsistentAssertions() throws Exception { + KeyPair keyPair = generateKeyPair(); + File privateKeyFile = generatePrivateKey(keyPair.getPrivate()); + + ConfigurationUtils cu = mockConfigForLocalAssertion(privateKeyFile); + Time time = new MockTime(); + + try (CloseableSupplier<String> supplier = AssertionSupplierFactory.create(cu, time)) { + String first = supplier.get(); + String second = supplier.get(); + assertNotNull(first); + assertNotNull(second); + // Both should be valid JWTs + assertClaims(keyPair.getPublic(), first); + assertClaims(keyPair.getPublic(), second); + } + } + + /** + * When the underlying file is deleted after factory creation, calling get() should + * throw a JwtRetrieverException since the assertion creator cannot read the file. + */ + @Test + public void testSupplierGetWrapsExceptionInJwtRetrieverException() throws Exception { + String jwt = createJwt("will-be-deleted"); + File jwtFile = tempFile(jwt); + + ConfigurationUtils cu = mock(ConfigurationUtils.class); + when(cu.validateString(SASL_OAUTHBEARER_ASSERTION_FILE, false)).thenReturn(jwtFile.getAbsolutePath()); + when(cu.validateFile(SASL_OAUTHBEARER_ASSERTION_FILE)).thenReturn(jwtFile); + + Time time = new MockTime(); + + try (CloseableSupplier<String> supplier = AssertionSupplierFactory.create(cu, time)) { + // Delete the file so the read will fail + assertTrue(jwtFile.delete()); + assertThrows(JwtRetrieverException.class, supplier::get); + } + } + + /** + * Closing the supplier should not throw any exceptions. + */ + @Test + public void testSupplierCloseDoesNotThrow() throws Exception { + String expectedJwt = createJwt("close-test"); + File jwtFile = tempFile(expectedJwt); + + ConfigurationUtils cu = mock(ConfigurationUtils.class); + when(cu.validateString(SASL_OAUTHBEARER_ASSERTION_FILE, false)).thenReturn(jwtFile.getAbsolutePath()); + when(cu.validateFile(SASL_OAUTHBEARER_ASSERTION_FILE)).thenReturn(jwtFile); + + Time time = new MockTime(); + + CloseableSupplier<String> supplier = AssertionSupplierFactory.create(cu, time); + // Verify close does not throw + assertDoesNotThrow(supplier::close); Review Comment: I think this could be convincing if we concocted a test case where the underlying call to `close()` does throw an exception. As in, create a test resources of type `Foo` whereby we force `close()` to fail, and we can verify that in the test. ########## clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/HttpRequestFormatterFactory.java: ########## @@ -0,0 +1,115 @@ +/* + * 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.config.SaslConfigs; +import org.apache.kafka.common.security.oauthbearer.internals.secured.assertion.AssertionSupplierFactory; +import org.apache.kafka.common.security.oauthbearer.internals.secured.assertion.CloseableSupplier; +import org.apache.kafka.common.utils.Time; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Map; +import java.util.Objects; + +import static org.apache.kafka.common.config.SaslConfigs.DEFAULT_SASL_OAUTHBEARER_HEADER_URLENCODE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_CLAIM_ISS; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_FILE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_HEADER_URLENCODE; + +/** + * Factory class for creating {@link HttpRequestFormatter} instances based on the OAuth authentication + * method configured. + * + * <p> + * This factory implements a three-tier fallback mechanism: file-based assertion (first), locally-generated + * assertion (second), or client secret (third/fallback). When multiple authentication methods are configured, + * the first preference takes precedence and other configurations are ignored with a WARN log message. + * </p> + * + * <p> + * The factory handles reading configuration values from both the main configuration and JAAS options + * (for backward compatibility) using {@link ConfigOrJaas}. + * </p> + */ +public class HttpRequestFormatterFactory { Review Comment: What about renaming it, though? This is specific to the `client_credentials` grant type. ########## clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/assertion/AssertionSupplierFactory.java: ########## @@ -0,0 +1,133 @@ +/* + * 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.assertion; + +import org.apache.kafka.common.security.oauthbearer.JwtRetrieverException; +import org.apache.kafka.common.security.oauthbearer.internals.secured.ConfigurationUtils; +import org.apache.kafka.common.utils.Time; +import org.apache.kafka.common.utils.Utils; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.IOException; +import java.util.Optional; + +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_ALGORITHM; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_FILE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_FILE; +import static org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_PASSPHRASE; +import static org.apache.kafka.common.security.oauthbearer.internals.secured.assertion.AssertionUtils.layeredAssertionJwtTemplate; + +/** + * Factory class for creating JWT assertion suppliers used in OAuth 2.0 client authentication. + * + * <p> + * This factory supports two methods of obtaining JWT assertions for client authentication: + * <ul> + * <li><b>File-based assertions:</b> Pre-generated JWT assertions read from a file specified by + * {@link org.apache.kafka.common.config.SaslConfigs#SASL_OAUTHBEARER_ASSERTION_FILE}. + * This is useful for testing or when assertions are managed externally.</li> + * <li><b>Locally-generated assertions:</b> JWTs dynamically created and signed using a private key + * specified by {@link org.apache.kafka.common.config.SaslConfigs#SASL_OAUTHBEARER_ASSERTION_PRIVATE_KEY_FILE}. + * This is the recommended approach for production use.</li> + * </ul> + * </p> + * + * <p> + * The created supplier can be invoked repeatedly to obtain assertions as needed (for example, when + * refreshing tokens). For file-based assertions, the assertion is cached and reloaded automatically when the file changes on disk. + * For locally-generated assertions, a new assertion with updated timestamps is created on each invocation. + * </p> + */ +public class AssertionSupplierFactory { + private static final Logger LOG = LoggerFactory.getLogger(AssertionSupplierFactory.class); + + /** + * Creates a closeable supplier that provides JWT assertions based on the provided configuration. + * + * <p> + * If {@link org.apache.kafka.common.config.SaslConfigs#SASL_OAUTHBEARER_ASSERTION_FILE} is configured, + * the supplier will read assertions from that file. Otherwise, it will create assertions dynamically + * using the configured private key and algorithm. + * </p> + * + * <p> + * <b>Important:</b> The returned {@link CloseableSupplier} must be closed when no longer needed + * to properly release resources such as file handles and cryptographic resources. + * </p> + * + * @param cu The configuration utilities containing assertion configuration + * @param time The time source for generating timestamps in locally-created assertions + * @return A closeable supplier that provides JWT assertion strings when invoked + * @throws org.apache.kafka.common.config.ConfigException if required configuration is missing or invalid Review Comment: You'll need to import `ConfigException`. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
