KAFKA-20101: Support wildcard OAuth bearer URLs

clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/ConfigurationUtils.java

@@ -51,6 +51,7 @@
 public class ConfigurationUtils {
 
     private static final Logger LOG = LoggerFactory.getLogger(ConfigurationUtils.class);
+    private static final String WILDCARD = "*";
 
     private final Map<String, ?> configs;
 
@@ -378,7 +379,8 @@ void throwIfURLIsNotAllowed(String configName, String configValue) {
             configName,
             configValue,
             ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG,
-            ALLOWED_SASL_OAUTHBEARER_URLS_DEFAULT
+            ALLOWED_SASL_OAUTHBEARER_URLS_DEFAULT,
+            true
         );
     }
 
@@ -390,21 +392,23 @@ void throwIfFileIsNotAllowed(String configName, String configValue) {
             configName,
             configValue,
             ALLOWED_SASL_OAUTHBEARER_FILES_CONFIG,
-            ALLOWED_SASL_OAUTHBEARER_FILES_DEFAULT
+            ALLOWED_SASL_OAUTHBEARER_FILES_DEFAULT,
+            false
         );
     }
 
     private void throwIfResourceIsNotAllowed(String resourceType,
                                              String configName,
                                              String configValue,
                                              String propertyName,
-                                             String propertyDefault) {
+                                             String propertyDefault,
+                                             boolean allowWildcard) {
         String[] allowedArray = System.getProperty(propertyName, propertyDefault).split(",");
         Set<String> allowed = Arrays.stream(allowedArray)
             .map(String::trim)
             .collect(Collectors.toSet());
 
-        if (!allowed.contains(configValue)) {
+        if (!allowed.contains(configValue) && !(allowWildcard && allowed.contains(WILDCARD))) {
             String message = String.format(
                 "The %s cannot be accessed due to restrictions. Update the system property '%s' to allow the %s to be accessed.",
                 resourceType,

clients/src/test/java/org/apache/kafka/common/security/oauthbearer/internals/secured/ConfigurationUtilsTest.java

@@ -40,6 +40,7 @@ public class ConfigurationUtilsTest extends OAuthBearerTest {
     @AfterEach
     public void tearDown() throws Exception {
         System.clearProperty(ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG);
+        System.clearProperty(ALLOWED_SASL_OAUTHBEARER_FILES_CONFIG);
     }
 
     @Test
@@ -158,6 +159,24 @@ public void testThrowIfURLIsNotAllowed() {
         assertDoesNotThrow(() -> cu.throwIfURLIsNotAllowed(FILE_CONFIG_NAME, fileUrl));
     }
 
+    @Test
+    public void testUrlWithWildcardAllowList() {
+        System.setProperty(ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG, " * ");
+        Map<String, Object> configs = Collections.singletonMap(URL_CONFIG_NAME, "https://another.example.com");
+        ConfigurationUtils cu = new ConfigurationUtils(configs);
+
+        assertDoesNotThrow(() -> cu.validateUrl(URL_CONFIG_NAME));
+    }
+
+    @Test
+    public void testFileAllowListDoesNotSupportWildcard() {
+        ConfigurationUtils cu = new ConfigurationUtils(Map.of());
+        System.setProperty(ALLOWED_SASL_OAUTHBEARER_FILES_CONFIG, "*");
+
+        assertThrowsWithMessage(ConfigException.class, () -> cu.throwIfFileIsNotAllowed(FILE_CONFIG_NAME, "/tmp/token"),
+            ALLOWED_SASL_OAUTHBEARER_FILES_CONFIG);
+    }
+
     @Test
     public void testThrowIfFileIsNotAllowed() {
         String file1 = "file1";

docs/configuration/system-properties.md

@@ -62,6 +62,8 @@ Default Value:
 
 This system property is used to set the allowed URLs as SASL OAUTHBEARER token or jwks endpoints. This property accepts comma-separated list of URLs. By default the value is an empty list. 
 
+The special value `*` allows any URL. Because this disables URL allow-listing, it should only be used when the endpoint configuration is trusted.
+
 If users want to enable some URLs, users need to explicitly set the system property like below. 
 
 ```bash
@@ -198,4 +200,3 @@ Default Value:
 All built-in ConfigProviders are enabled
 </td></tr> </table>
 
-