This is an automated email from the ASF dual-hosted git repository.
mimaison pushed a commit to branch 3.8
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/3.8 by this push:
new b92ecb12556 MINOR: Cleanups in ConfigurationUtils (#18769)
b92ecb12556 is described below
commit b92ecb1255652925533af210524a053eb2ce4fa8
Author: Mickael Maison <[email protected]>
AuthorDate: Sat Feb 1 09:51:51 2025 +0100
MINOR: Cleanups in ConfigurationUtils (#18769)
Reviewers: Luke Chen <[email protected]>
---
.../config/internals/BrokerSecurityConfigs.java | 3 +
.../secured/AccessTokenRetrieverFactory.java | 1 +
.../internals/secured/ConfigurationUtils.java | 23 +++++++
.../secured/VerificationKeyResolverFactory.java | 1 +
.../secured/AccessTokenRetrieverFactoryTest.java | 27 ++++++--
.../internals/secured/ConfigurationUtilsTest.java | 46 +++++++++++--
.../VerificationKeyResolverFactoryTest.java | 80 ++++++++++++++++++++++
7 files changed, 173 insertions(+), 8 deletions(-)
diff --git
a/clients/src/main/java/org/apache/kafka/common/config/internals/BrokerSecurityConfigs.java
b/clients/src/main/java/org/apache/kafka/common/config/internals/BrokerSecurityConfigs.java
index 767fa9aca01..ca3105c6123 100644
---
a/clients/src/main/java/org/apache/kafka/common/config/internals/BrokerSecurityConfigs.java
+++
b/clients/src/main/java/org/apache/kafka/common/config/internals/BrokerSecurityConfigs.java
@@ -115,4 +115,7 @@ public class BrokerSecurityConfigs {
public final static String SASL_MECHANISM_INTER_BROKER_PROTOCOL_CONFIG =
"sasl.mechanism.inter.broker.protocol";
public final static String SASL_MECHANISM_INTER_BROKER_PROTOCOL_DOC =
"SASL mechanism used for inter-broker communication. Default is GSSAPI.";
+ // The allowlist of the SASL OAUTHBEARER endpoints
+ public static final String ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG =
"org.apache.kafka.sasl.oauthbearer.allowed.urls";
+
}
diff --git
a/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/AccessTokenRetrieverFactory.java
b/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/AccessTokenRetrieverFactory.java
index ac224730597..1a9a2864a12 100644
---
a/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/AccessTokenRetrieverFactory.java
+++
b/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/AccessTokenRetrieverFactory.java
@@ -53,6 +53,7 @@ public class AccessTokenRetrieverFactory {
String saslMechanism,
Map<String, Object> jaasConfig) {
ConfigurationUtils cu = new ConfigurationUtils(configs, saslMechanism);
+ cu.throwIfURLIsNotAllowed(SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL);
URL tokenEndpointUrl =
cu.validateUrl(SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL);
if
(tokenEndpointUrl.getProtocol().toLowerCase(Locale.ROOT).equals("file")) {
diff --git
a/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/ConfigurationUtils.java
b/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/ConfigurationUtils.java
index 99bc20661bb..499d0c4f37f 100644
---
a/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/ConfigurationUtils.java
+++
b/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/ConfigurationUtils.java
@@ -22,11 +22,17 @@ import java.net.MalformedURLException;
import java.net.URISyntaxException;
import java.net.URL;
import java.nio.file.Path;
+import java.util.Arrays;
import java.util.Locale;
import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
import org.apache.kafka.common.config.ConfigException;
import org.apache.kafka.common.network.ListenerName;
+import static
org.apache.kafka.common.config.internals.BrokerSecurityConfigs.ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG;
+
/**
* <code>ConfigurationUtils</code> is a utility class to perform basic
configuration-related
* logic and is separated out here for easier, more direct testing.
@@ -218,4 +224,21 @@ public class ConfigurationUtils {
return (T) configs.get(name);
}
+ // make sure the url is in the
"org.apache.kafka.sasl.oauthbearer.allowed.urls" system property
+ public void throwIfURLIsNotAllowed(String urlConfig) {
+ String allowedUrlsProp =
System.getProperty(ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG);
+ if (allowedUrlsProp == null) {
+ // by default, we accept all URLs
+ return;
+ }
+ Set<String> allowedUrlsList = Arrays.stream(allowedUrlsProp.split(","))
+ .map(String::trim)
+ .collect(Collectors.toSet());
+
+ String value = get(urlConfig);
+ if (!allowedUrlsList.contains(value)) {
+ throw new IllegalArgumentException(value + " is not allowed.
Update system property '"
+ + ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG + "' to allow " +
value);
+ }
+ }
}
diff --git
a/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/VerificationKeyResolverFactory.java
b/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/VerificationKeyResolverFactory.java
index 3eb8dd64ba1..de0bf4b7e0a 100644
---
a/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/VerificationKeyResolverFactory.java
+++
b/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/VerificationKeyResolverFactory.java
@@ -55,6 +55,7 @@ public class VerificationKeyResolverFactory {
String saslMechanism,
Map<String, Object> jaasConfig) {
ConfigurationUtils cu = new ConfigurationUtils(configs, saslMechanism);
+ cu.throwIfURLIsNotAllowed(SASL_OAUTHBEARER_JWKS_ENDPOINT_URL);
URL jwksEndpointUrl =
cu.validateUrl(SASL_OAUTHBEARER_JWKS_ENDPOINT_URL);
if
(jwksEndpointUrl.getProtocol().toLowerCase(Locale.ROOT).equals("file")) {
diff --git
a/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/internals/secured/AccessTokenRetrieverFactoryTest.java
b/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/internals/secured/AccessTokenRetrieverFactoryTest.java
index 741b4d2566d..40d5c0ea6ea 100644
---
a/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/internals/secured/AccessTokenRetrieverFactoryTest.java
+++
b/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/internals/secured/AccessTokenRetrieverFactoryTest.java
@@ -17,17 +17,25 @@
package org.apache.kafka.common.security.oauthbearer.internals.secured;
-import static
org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL;
-import static org.junit.jupiter.api.Assertions.assertEquals;
import java.io.File;
import java.util.Collections;
import java.util.Map;
import org.apache.kafka.common.config.ConfigException;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
+import static
org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL;
+import static
org.apache.kafka.common.config.internals.BrokerSecurityConfigs.ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
public class AccessTokenRetrieverFactoryTest extends OAuthBearerTest {
+ @AfterEach
+ public void tearDown() throws Exception {
+ System.clearProperty(ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG);
+ }
+
@Test
public void testConfigureRefreshingFileAccessTokenRetriever() throws
Exception {
String expected = "{}";
@@ -47,14 +55,15 @@ public class AccessTokenRetrieverFactoryTest extends
OAuthBearerTest {
@Test
public void
testConfigureRefreshingFileAccessTokenRetrieverWithInvalidDirectory() {
// Should fail because the parent path doesn't exist.
- Map<String, ?> configs =
getSaslConfigs(SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL, new
File("/tmp/this-directory-does-not-exist/foo.json").toURI().toString());
+ String file = new
File("/tmp/this-directory-does-not-exist/foo.json").toURI().toString();
+ Map<String, ?> configs =
getSaslConfigs(SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL, file);
Map<String, Object> jaasConfig = Collections.emptyMap();
assertThrowsWithMessage(ConfigException.class, () ->
AccessTokenRetrieverFactory.create(configs, jaasConfig), "that doesn't exist");
}
@Test
public void
testConfigureRefreshingFileAccessTokenRetrieverWithInvalidFile() throws
Exception {
- // Should fail because the while the parent path exists, the file
itself doesn't.
+ // Should fail because while the parent path exists, the file itself
doesn't.
File tmpDir = createTempDir("this-directory-does-exist");
File accessTokenFile = new File(tmpDir,
"this-file-does-not-exist.json");
Map<String, ?> configs =
getSaslConfigs(SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL,
accessTokenFile.toURI().toString());
@@ -62,4 +71,14 @@ public class AccessTokenRetrieverFactoryTest extends
OAuthBearerTest {
assertThrowsWithMessage(ConfigException.class, () ->
AccessTokenRetrieverFactory.create(configs, jaasConfig), "that doesn't exist");
}
+ @Test
+ public void testSaslOauthbearerTokenEndpointUrlIsNotAllowed() throws
Exception {
+ // Should fail because the file is not in the allowed list
+ File tmpDir = createTempDir("not_allowed");
+ File accessTokenFile = new File(tmpDir, "not_allowed.json");
+ System.setProperty(ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG, "nothing");
+ Map<String, ?> configs =
getSaslConfigs(SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL,
accessTokenFile.toURI().toString());
+ assertThrowsWithMessage(IllegalArgumentException.class, () ->
AccessTokenRetrieverFactory.create(configs, Collections.emptyMap()),
+ ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG);
+ }
}
diff --git
a/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/internals/secured/ConfigurationUtilsTest.java
b/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/internals/secured/ConfigurationUtilsTest.java
index b1ed7c16529..0d3df183ae3 100644
---
a/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/internals/secured/ConfigurationUtilsTest.java
+++
b/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/internals/secured/ConfigurationUtilsTest.java
@@ -17,17 +17,30 @@
package org.apache.kafka.common.security.oauthbearer.internals.secured;
+import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.test.TestUtils;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
import java.io.File;
import java.io.IOException;
import java.util.Collections;
+import java.util.HashMap;
import java.util.Map;
-import org.apache.kafka.common.config.ConfigException;
-import org.apache.kafka.test.TestUtils;
-import org.junit.jupiter.api.Test;
+
+import static
org.apache.kafka.common.config.internals.BrokerSecurityConfigs.ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
public class ConfigurationUtilsTest extends OAuthBearerTest {
- private final static String URL_CONFIG_NAME = "url";
+ private static final String URL_CONFIG_NAME = "url";
+ private static final String FILE_CONFIG_NAME = "file";
+
+ @AfterEach
+ public void tearDown() throws Exception {
+ System.clearProperty(ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG);
+ }
@Test
public void testUrl() {
@@ -127,6 +140,31 @@ public class ConfigurationUtilsTest extends
OAuthBearerTest {
assertThrowsWithMessage(ConfigException.class, () -> testFile(" "),
"must not contain only whitespace");
}
+ @Test
+ public void testAllowedSaslOauthbearerUrlSystemProperty() {
+ String url = "http://www.example.com";;
+ String fileUrl = "file:///etc/passwd";
+ Map<String, Object> configs = new HashMap<>();
+ configs.put(URL_CONFIG_NAME, url);
+ configs.put(FILE_CONFIG_NAME, fileUrl);
+ ConfigurationUtils cu = new ConfigurationUtils(configs);
+
+ // By default, all URLs are allowed
+ assertDoesNotThrow(() -> cu.throwIfURLIsNotAllowed(URL_CONFIG_NAME));
+ assertDoesNotThrow(() -> cu.throwIfURLIsNotAllowed(FILE_CONFIG_NAME));
+
+ // add one url into allowed list
+ System.setProperty(ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG, url);
+ assertDoesNotThrow(() -> cu.throwIfURLIsNotAllowed(URL_CONFIG_NAME));
+ assertThrowsWithMessage(IllegalArgumentException.class, () ->
cu.throwIfURLIsNotAllowed(FILE_CONFIG_NAME),
+ ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG);
+
+ // add all urls into allowed list
+ System.setProperty(ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG, url + "," +
fileUrl);
+ assertDoesNotThrow(() -> cu.throwIfURLIsNotAllowed(URL_CONFIG_NAME));
+ assertDoesNotThrow(() -> cu.throwIfURLIsNotAllowed(FILE_CONFIG_NAME));
+ }
+
protected void testFile(String value) {
Map<String, Object> configs =
Collections.singletonMap(URL_CONFIG_NAME, value);
ConfigurationUtils cu = new ConfigurationUtils(configs);
diff --git
a/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/internals/secured/VerificationKeyResolverFactoryTest.java
b/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/internals/secured/VerificationKeyResolverFactoryTest.java
new file mode 100644
index 00000000000..0c4abd3cbe1
--- /dev/null
+++
b/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/internals/secured/VerificationKeyResolverFactoryTest.java
@@ -0,0 +1,80 @@
+/*
+ * 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.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.File;
+import java.util.Collections;
+import java.util.Map;
+
+import static
org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_JWKS_ENDPOINT_URL;
+import static
org.apache.kafka.common.config.internals.BrokerSecurityConfigs.ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG;
+
+public class VerificationKeyResolverFactoryTest extends OAuthBearerTest {
+
+ @AfterEach
+ public void tearDown() throws Exception {
+ System.clearProperty(ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG);
+ }
+
+ @Test
+ public void testConfigureRefreshingFileVerificationKeyResolver() throws
Exception {
+ File tmpDir = createTempDir("access-token");
+ File verificationKeyFile = createTempFile(tmpDir, "access-token-",
".json", "{}");
+
+ Map<String, ?> configs =
Collections.singletonMap(SASL_OAUTHBEARER_JWKS_ENDPOINT_URL,
verificationKeyFile.toURI().toString());
+ Map<String, Object> jaasConfig = Collections.emptyMap();
+
+ // verify it won't throw exception
+ try (CloseableVerificationKeyResolver verificationKeyResolver =
VerificationKeyResolverFactory.create(configs, jaasConfig)) { }
+ }
+
+ @Test
+ public void
testConfigureRefreshingFileVerificationKeyResolverWithInvalidDirectory() {
+ // Should fail because the parent path doesn't exist.
+ String file = new
File("/tmp/this-directory-does-not-exist/foo.json").toURI().toString();
+ Map<String, ?> configs =
getSaslConfigs(SASL_OAUTHBEARER_JWKS_ENDPOINT_URL, file);
+ Map<String, Object> jaasConfig = Collections.emptyMap();
+ assertThrowsWithMessage(ConfigException.class, () ->
VerificationKeyResolverFactory.create(configs, jaasConfig), "that doesn't
exist");
+ }
+
+ @Test
+ public void
testConfigureRefreshingFileVerificationKeyResolverWithInvalidFile() throws
Exception {
+ // Should fail because while the parent path exists, the file itself
doesn't.
+ File tmpDir = createTempDir("this-directory-does-exist");
+ File verificationKeyFile = new File(tmpDir,
"this-file-does-not-exist.json");
+ Map<String, ?> configs =
getSaslConfigs(SASL_OAUTHBEARER_JWKS_ENDPOINT_URL,
verificationKeyFile.toURI().toString());
+ Map<String, Object> jaasConfig = Collections.emptyMap();
+ assertThrowsWithMessage(ConfigException.class, () ->
VerificationKeyResolverFactory.create(configs, jaasConfig), "that doesn't
exist");
+ }
+
+ @Test
+ public void testSaslOauthbearerTokenEndpointUrlIsNotAllowed() throws
Exception {
+ // Should fail because the file is not in the allowed list
+ File tmpDir = createTempDir("not_allowed");
+ File verificationKeyFile = new File(tmpDir, "not_allowed.json");
+ System.setProperty(ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG, "nothing");
+ Map<String, ?> configs =
getSaslConfigs(SASL_OAUTHBEARER_JWKS_ENDPOINT_URL,
verificationKeyFile.toURI().toString());
+ assertThrowsWithMessage(IllegalArgumentException.class, () ->
VerificationKeyResolverFactory.create(configs, Collections.emptyMap()),
+ ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG);
+ }
+}
