Repository: kroxylicious/kroxylicious (java)

Issue:
Kafka 4.0.0 : Oauthbearer fail without change properties
**Describe the bug**
Kafka 4.0.0 introduces a new "mandatory" system property when using `SASL_SSL` with `OAUTHBEARER`.
The property is described here : https://kafka.apache.org/documentation/#systemproperties_org.apache.kafka.sasl.oauthbearer.allowed.urls
Because of that, the proxy fails to start when upgrading from `0.11.0` to `0.12.0`

**To Reproduce**
```yaml
filterDefinitions:
  - name: jwt
    type: OauthBearerValidation
    config:
      jwksEndpointUrl: https://my.jwks.url
defaultFilters:
- jwt
```

**Expected behavior**
As a proxy owner, I want to upgrade from `0.11` to `0.12` without disruption.

**YAML files and logs**

```
io.kroxylicious.proxy.plugin.PluginConfigurationException: Exception initializing filter factory jwt with config Config[jwksEndpointUrl=https://my.jwks.url, jwksEndpointRefreshMs=null, jwksEndpointRetryBackoffMs=null, jwksEndpointRetryBackoffMaxMs=null, scopeClaimName=null, subClaimName=null, authenticateBackOffMaxMs=null, authenticateCacheMaxSize=null, expectedAudience=null, expectedIssuer=null]: https://my.jwks.url is not allowed. Update system property 'org.apache.kafka.sasl.oauthbearer.allowed.urls' to allow https://my.jwks.url
	at io.kroxylicious.proxy.bootstrap.FilterChainFactory$Wrapper.<init>(FilterChainFactory.java:55)
	at io.kroxylicious.proxy.bootstrap.FilterChainFactory.<init>(FilterChainFactory.java:114)
	at io.kroxylicious.proxy.KafkaProxy.startup(KafkaProxy.java:138)
	at io.kroxylicious.app.Kroxylicious.call(Kroxylicious.java:73)
	at io.kroxylicious.app.Kroxylicious.call(Kroxylicious.java:35)
	at picocli.CommandLine.executeUserObject(CommandLine.java:2031)
	at picocli.CommandLine.access$1500(CommandLine.java:148)
	at picocli.CommandLine$RunLast.executeUserObjectOfLastSubcommandWithSameParent(CommandLine.java:2469)
	at picocli.CommandLine$RunLast.handle(CommandLine.java:2461)
	at picocli.CommandLine$RunLast.handle(CommandLine.java:2423)
	at picocli.CommandLine$AbstractParseResultHandler.execute(CommandLine.java:2277)
	at picocli.CommandLine$RunLast.execute(CommandLine.java:2425)
	at picocli.CommandLine.execute(CommandLine.java:2174)
	at io.kroxylicious.app.Kroxylicious.main(Kroxylicious.java:123)
```

**Additional context**

- A test is missing to assert the `configure` step of `OAuthBearerValidatorCallbackHandler` in `OauthBearerValidation`
- We can certainly set the JWKS url defined by the user as authorized by default in JAVA Properties
```java
Config configWithDefaults = initConfigWithDefaults(config);
System.setProperty("org.apache.kafka.sasl.oauthbearer.allowed.urls", config.jwksEndpointUrl().toString());
oauthHandler.configure(
        createSaslConfigMap(configWithDefaults),
        OAUTHBEARER_MECHANISM,
        createDefaultJaasConfig());
```

Pull request description:
Fix #2091: OauthBearer factory needs to add the JWKS endpoint to the allow-list.
### Type of change

- Bugfix

### Description

In Kafka 4.0.0, the OauthBearer server side needs to add the JWKS endpoint URL to an allow list
in order for the Kafka libraries to interact with it.

With this change, the OauthBearer factory will automatically set the Kafka `org.apache.kafka.sasl.oauthbearer.allowed.urls` system property to include the JWKS endpoint (if it does not already include the value).  To support embedded use-cases, the factory will undo the system properties changes on closure (this is a best effort approach).

### Additional Context

_Why are you making this pull request?_

### Checklist

_Please go through this checklist and make sure all applicable tasks have been done_

- [x] PR raised from a fork of this repository and made from a branch rather than main. 
- [x] Write tests
- [ ] Update documentation
- [ ] Make sure all unit/integration tests pass
- [ ] Make sure all Sonarcloud warnings are addressed or are justifiably ignored.
- [ ] If applicable to the change, [trigger the system test suite](../blob/main/DEV_GUIDE.md#jenkins-pipeline-for-system-tests).  Make sure tests pass.
- [ ] If applicable to the change, [trigger the performance test suite](../blob/main/PERFORMANCE.md#jenkins-pipeline-for-performance). Ensure that any degradations to performance numbers are understood and justified.
- [ ] Ensure the PR references relevant issue(s) so they are closed on merging.
- [ ] For user facing changes, update CHANGELOG.md (remember to include changes affecting the API of the test artefacts too).

> **_NOTE:_**  You must be a member of `@kroxylicious/developers` to trigger the system test and performance test suites.  If you are not part of this group, comment on the PR requesting a trigger, tagging `@kroxylicious/developers`.


Patch:
diff --git a/kroxylicious-filters/kroxylicious-oauthbearer-validation/pom.xml b/kroxylicious-filters/kroxylicious-oauthbearer-validation/pom.xml
index 3519bebf..f510c130 100644
--- a/kroxylicious-filters/kroxylicious-oauthbearer-validation/pom.xml
+++ b/kroxylicious-filters/kroxylicious-oauthbearer-validation/pom.xml
@@ -60,6 +60,11 @@
             <artifactId>junit-jupiter-api</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-params</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.mockito</groupId>
             <artifactId>mockito-core</artifactId>
@@ -102,6 +107,10 @@
             <artifactId>kroxylicious-filter-test-support</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.junit-pioneer</groupId>
+            <artifactId>junit-pioneer</artifactId>
+        </dependency>
     </dependencies>
     <build>
         <plugins>
diff --git a/kroxylicious-filters/kroxylicious-oauthbearer-validation/src/main/java/io/kroxylicious/proxy/filter/oauthbearer/OauthBearerValidation.java b/kroxylicious-filters/kroxylicious-oauthbearer-validation/src/main/java/io/kroxylicious/proxy/filter/oauthbearer/OauthBearerValidation.java
index 14969bd7..04ff79c1 100644
--- a/kroxylicious-filters/kroxylicious-oauthbearer-validation/src/main/java/io/kroxylicious/proxy/filter/oauthbearer/OauthBearerValidation.java
+++ b/kroxylicious-filters/kroxylicious-oauthbearer-validation/src/main/java/io/kroxylicious/proxy/filter/oauthbearer/OauthBearerValidation.java
@@ -8,10 +8,14 @@ package io.kroxylicious.proxy.filter.oauthbearer;
 
 import java.net.URI;
 import java.time.Duration;
+import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Deque;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentLinkedDeque;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -43,10 +47,15 @@ public class OauthBearerValidation implements FilterFactory<OauthBearerValidatio
 
     private final OAuthBearerValidatorCallbackHandler oauthHandler;
 
+    @VisibleForTesting
+    static final String ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG = "org.apache.kafka.sasl.oauthbearer.allowed.urls";
+
     static {
         OAuthBearerSaslServerProvider.initialize();
     }
 
+    private final Deque<Runnable> oauthSystemPropertyCleanupTasks = new ConcurrentLinkedDeque<>();
+
     @SuppressWarnings("unused")
     public OauthBearerValidation() {
         this(new OAuthBearerValidatorCallbackHandler());
@@ -61,6 +70,7 @@ public class OauthBearerValidation implements FilterFactory<OauthBearerValidatio
     @SuppressWarnings("java:S2245") // secure randomization not needed for exponential backoff
     public SharedOauthBearerValidationContext initialize(FilterFactoryContext context, Config config) throws PluginConfigurationException {
         Plugins.requireConfig(this, config);
+        setAllowedSaslOauthbearerSysPropIfNecessary(config.jwksEndpointUrl().toString());
         Config configWithDefaults = initConfigWithDefaults(config);
         oauthHandler.configure(
                 createSaslConfigMap(configWithDefaults),
@@ -75,6 +85,39 @@ public class OauthBearerValidation implements FilterFactory<OauthBearerValidatio
         return new SharedOauthBearerValidationContext(configWithDefaults, backoffStrategy, rateLimiter, oauthHandler);
     }
 
+    private void setAllowedSaslOauthbearerSysPropIfNecessary(String jwkUrl) {
+        var allowedUrls = parseAllowedSaslOauthBearerProperty();
+        if (!allowedUrls.contains(jwkUrl)) {
+            allowedUrls.add(jwkUrl);
+            setOrClearAllowedSaslOauthBearerProperty(allowedUrls);
+            this.oauthSystemPropertyCleanupTasks.push(() -> {
+                var now = parseAllowedSaslOauthBearerProperty();
+                now.remove(jwkUrl);
+                setOrClearAllowedSaslOauthBearerProperty(now);
+            });
+        }
+    }
+
+    @NonNull
+    private List<String> parseAllowedSaslOauthBearerProperty() {
+        String property = System.getProperty(ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG);
+        var allowedList = Optional.ofNullable(property)
+                .map(p -> Arrays.stream(p.split(","))
+                        .map(String::trim)
+                        .toList())
+                .orElse(List.of());
+        return new ArrayList<>(allowedList);
+    }
+
+    private void setOrClearAllowedSaslOauthBearerProperty(List<String> allowedUrls) {
+        if (allowedUrls.isEmpty()) {
+            System.clearProperty(ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG);
+        }
+        else {
+            System.setProperty(ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG, String.join(",", allowedUrls));
+        }
+    }
+
     @NonNull
     @Override
     public OauthBearerValidationFilter createFilter(FilterFactoryContext context, SharedOauthBearerValidationContext sharedContext) {
@@ -84,6 +127,10 @@ public class OauthBearerValidation implements FilterFactory<OauthBearerValidatio
     @Override
     public void close(SharedOauthBearerValidationContext sharedContext) {
         oauthHandler.close();
+        while (!oauthSystemPropertyCleanupTasks.isEmpty()) {
+            oauthSystemPropertyCleanupTasks.pop().run();
+        }
+
     }
 
     public record Config(


Test patch:
diff --git a/kroxylicious-filters/kroxylicious-oauthbearer-validation/src/test/java/io/kroxylicious/proxy/filter/oauthbearer/OauthBearerValidationTest.java b/kroxylicious-filters/kroxylicious-oauthbearer-validation/src/test/java/io/kroxylicious/proxy/filter/oauthbearer/OauthBearerValidationTest.java
index 57d1d52a..a1605340 100644
--- a/kroxylicious-filters/kroxylicious-oauthbearer-validation/src/test/java/io/kroxylicious/proxy/filter/oauthbearer/OauthBearerValidationTest.java
+++ b/kroxylicious-filters/kroxylicious-oauthbearer-validation/src/test/java/io/kroxylicious/proxy/filter/oauthbearer/OauthBearerValidationTest.java
@@ -8,11 +8,19 @@ package io.kroxylicious.proxy.filter.oauthbearer;
 
 import java.net.URI;
 import java.util.List;
+import java.util.UUID;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+import java.util.stream.Stream;
 
 import org.apache.kafka.common.config.SaslConfigs;
 import org.apache.kafka.common.security.oauthbearer.OAuthBearerValidatorCallbackHandler;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.junitpioneer.jupiter.RestoreSystemProperties;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 
@@ -21,7 +29,9 @@ import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
 
 import io.kroxylicious.proxy.filter.FilterDispatchExecutor;
 import io.kroxylicious.proxy.filter.FilterFactoryContext;
+import io.kroxylicious.proxy.filter.oauthbearer.OauthBearerValidation.Config;
 
+import static io.kroxylicious.proxy.filter.oauthbearer.OauthBearerValidation.ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.anyList;
 import static org.mockito.ArgumentMatchers.assertArg;
@@ -30,6 +40,7 @@ import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 @ExtendWith(MockitoExtension.class)
+@RestoreSystemProperties
 class OauthBearerValidationTest {
 
     @Mock
@@ -51,7 +62,7 @@ class OauthBearerValidationTest {
     @Test
     void mustProvideDefaultValuesForConfig() throws Exception {
         ObjectMapper yamlMapper = new ObjectMapper(new YAMLFactory());
-        OauthBearerValidation.Config yamlConfig = yamlMapper.readerFor(OauthBearerValidation.Config.class).readValue("""
+        Config yamlConfig = yamlMapper.readerFor(Config.class).readValue("""
                 jwksEndpointUrl: https://jwks.endpoint
                 """);
 
@@ -60,14 +71,14 @@ class OauthBearerValidationTest {
 
     @Test
     @SuppressWarnings("java:S5838")
-    void mustInitAndCreateFilterWithDefaultsNull() throws Exception {
+    void mustInitAndCreateFilterWithDefaultsNull() {
         mustInitAndCreateFilter(defaultConfig());
     }
 
     @Test
-    void mustInitAndCreateFilterWithNegativeAndEmptyValues() throws Exception {
-        OauthBearerValidation.Config config = new OauthBearerValidation.Config(
-                new URI("https://jwks.endpoint"),
+    void mustInitAndCreateFilterWithNegativeAndEmptyValues() {
+        Config config = new Config(
+                URI.create("https://jwks.endpoint"),
                 -1L,
                 -1L,
                 -1L,
@@ -86,7 +97,7 @@ class OauthBearerValidationTest {
         // given
         when(ffc.filterDispatchExecutor()).thenReturn(executor);
         OauthBearerValidation oauthBearerValidation = new OauthBearerValidation(callbackHandler);
-        OauthBearerValidation.Config config = new OauthBearerValidation.Config(
+        Config config = new Config(
                 new URI("https://jwks.endpoint"),
                 10000L,
                 20000L,
@@ -132,8 +143,55 @@ class OauthBearerValidationTest {
         verify(callbackHandler).close();
     }
 
+    @Test
+    void initShouldAddJwksEndpointToOauthAllowList() {
+        // given
+        OauthBearerValidation oauthBearerValidation = new OauthBearerValidation(callbackHandler);
+        Config config = defaultConfig(URI.create("https://" + UUID.randomUUID() + ".invalid"));
+
+        // when
+        oauthBearerValidation.initialize(ffc, config);
+
+        // then
+        assertThat(System.getProperty(ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG))
+                .isNotEmpty()
+                .contains(config.jwksEndpointUrl().toString());
+    }
+
+    static Stream<Arguments> closeShouldRestoreOauthAllowList() {
+        return Stream.of(
+                Arguments.argumentSet("close clears system property",
+                        (Consumer<String>) unused -> System.clearProperty(ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG),
+                        (BiConsumer<String, String>) (initial, propValue) -> assertThat(propValue).isNull()),
+                Arguments.argumentSet("close preserves an existing system property that exactly matches config value",
+                        (Consumer<String>) configValue -> System.setProperty(ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG, configValue),
+                        (BiConsumer<String, String>) (initial, propValue) -> assertThat(propValue).isEqualTo(initial)),
+                Arguments.argumentSet("close preserves an existing system property that did not matches config value",
+                        (Consumer<String>) configValue -> System.setProperty(ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG, "https://another.invalid"),
+                        (BiConsumer<String, String>) (initial, propValue) -> assertThat(propValue).isEqualTo("https://another.invalid")));
+    }
+
+    @ParameterizedTest
+    @MethodSource
+    @SuppressWarnings("java:S6103") // false positive, the consumer does include an assertion.
+    void closeShouldRestoreOauthAllowList(Consumer<String> prepareSysProperties, BiConsumer<String, String> postCloseAssertions) {
+        // given
+        var jwksEndpointUrl = "https://" + UUID.randomUUID() + ".invalid";
+        prepareSysProperties.accept(jwksEndpointUrl);
+        OauthBearerValidation oauthBearerValidation = new OauthBearerValidation(callbackHandler);
+        Config config = defaultConfig(URI.create(jwksEndpointUrl));
+        var context = oauthBearerValidation.initialize(ffc, config);
+
+        // when
+        oauthBearerValidation.close(context);
+
+        // then
+        assertThat(System.getProperty(ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG))
+                .satisfies(current -> postCloseAssertions.accept(jwksEndpointUrl, current));
+    }
+
     @SuppressWarnings("java:S5838")
-    void mustInitAndCreateFilter(OauthBearerValidation.Config config) {
+    void mustInitAndCreateFilter(Config config) {
         // given
         OauthBearerValidation oauthBearerValidation = new OauthBearerValidation(callbackHandler);
         when(ffc.filterDispatchExecutor()).thenReturn(executor);
@@ -162,9 +220,13 @@ class OauthBearerValidationTest {
         assertThat(sharedContext.config().authenticateCacheMaxSize()).isEqualTo(1000);
     }
 
-    private OauthBearerValidation.Config defaultConfig() throws Exception {
-        return new OauthBearerValidation.Config(
-                new URI("https://jwks.endpoint"),
+    private Config defaultConfig() {
+        return defaultConfig(URI.create("https://jwks.endpoint"));
+    }
+
+    private Config defaultConfig(URI jwksEndpointUrl) {
+        return new Config(
+                jwksEndpointUrl,
                 null,
                 null,
                 null,
diff --git a/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/filter/oauthbearer/OauthBearerValidationIT.java b/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/filter/oauthbearer/OauthBearerValidationIT.java
index 0eb27b58..6bca324a 100644
--- a/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/filter/oauthbearer/OauthBearerValidationIT.java
+++ b/kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/filter/oauthbearer/OauthBearerValidationIT.java
@@ -94,8 +94,10 @@ class OauthBearerValidationIT {
     private static OauthServerContainer oauthServer;
 
     @BeforeAll
-    public static void beforeAll() {
-        // required to permit Kafka to interact with our endpoint.
+    static void beforeAll() {
+        // Kafka 4.0 requires that the org.apache.kafka.sasl.oauthbearer.allowed.urls sys property is set in order to use Oauth Bearer.
+        // The Kafka Broker and Proxy requires that JWKS_ENDPOINT_URL is in the allow list.
+        // The Kafka Client requires that TOKEN_ENDPOINT_URL is in the allow list.
         System.setProperty(ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG, JWKS_ENDPOINT_URL + "," + TOKEN_ENDPOINT_URL);
 
         oauthServer = new OauthServerContainer(OauthBearerValidationIT.DOCKER_IMAGE_NAME);
@@ -107,14 +109,14 @@ class OauthBearerValidationIT {
     }
 
     @AfterAll
-    public static void afterAll() {
+    static void afterAll() {
         if (oauthServer != null) {
             oauthServer.close();
         }
     }
 
     @AfterEach
-    public void afterEach() throws Exception {
+    void afterEach() throws Exception {
         workaroundKafka17134();
     }