Fix for kroxylicious#3035 and new ITs demonstrating the virtual cluster/sasl inspection subject builder working end-to-end (kroxylicious#3036)

* fix(sasl inspection): add missing annotations that cause kroxylicious#3035
* Add unit tests for sasl inspector config deserialization

Signed-off-by: Keith Wall <[email protected]>
Signed-off-by: Robert Young <[email protected]>
Co-authored-by: Robert Young <[email protected]>

Author: k-wall

CHANGELOG.md

@@ -6,6 +6,9 @@ For changes that effect a public API, the [deprecation policy](./DEV_GUIDE.md#de
 Format `<github issue/pr number>: <short description>`.
 
 ## SNAPSHOT
+
+* [#3035](https://github.com/kroxylicious/kroxylicious/issues/3035): fix(sasl inspector): Fix config parsing error if SaslInspector with subject builder
+
 ## 0.18.0
 
 * [#2922](https://github.com/kroxylicious/kroxylicious/pull/2922): build(deps): bump kafka.version from 4.1.0 to 4.1.1

kroxylicious-filters/kroxylicious-sasl-inspection/pom.xml

@@ -81,6 +81,11 @@
             <artifactId>logcaptor</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
 </project>

kroxylicious-filters/kroxylicious-sasl-inspection/src/main/java/io/kroxylicious/filters/sasl/inspection/Config.java

@@ -8,6 +8,10 @@
 
 import java.util.Set;
 
+import io.kroxylicious.proxy.authentication.SaslSubjectBuilderService;
+import io.kroxylicious.proxy.plugin.PluginImplConfig;
+import io.kroxylicious.proxy.plugin.PluginImplName;
+
 import edu.umd.cs.findbugs.annotations.Nullable;
 
 /**
@@ -25,6 +29,6 @@
  * SASL authentication has been attempted or was successful. Defaults to false.
  */
 public record Config(@Nullable Set<String> enabledMechanisms,
-                     @Nullable String subjectBuilder,
-                     @Nullable Object subjectBuilderConfig,
+                     @Nullable @PluginImplName(SaslSubjectBuilderService.class) String subjectBuilder,
+                     @Nullable @PluginImplConfig(implNameProperty = "subjectBuilder") Object subjectBuilderConfig,
                      @Nullable Boolean requireAuthentication) {}

kroxylicious-filters/kroxylicious-sasl-inspection/src/test/java/io/kroxylicious/filters/sasl/inspection/ConfigurationSerializationTest.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.filters.sasl.inspection;
+
+import java.io.IOException;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+
+import io.kroxylicious.proxy.config.ConfigParser;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class ConfigurationSerializationTest {
+
+    @Test
+    void deserializeEmpty() throws IOException {
+        Config config = parseConfig("{}");
+        assertThat(config.enabledMechanisms()).isNull();
+        assertThat(config.requireAuthentication()).isNull();
+        assertThat(config.subjectBuilderConfig()).isNull();
+        assertThat(config.subjectBuilder()).isNull();
+    }
+
+    @Test
+    void deserializeEnabledMechanisms() throws IOException {
+        String configYaml = """
+                enabledMechanisms:
+                  - PLAIN
+                """;
+        Config config = parseConfig(configYaml);
+        assertThat(config.enabledMechanisms()).containsExactly("PLAIN");
+    }
+
+    @CsvSource({ "true", "false" })
+    @ParameterizedTest
+    void deserializeRequireAuthentication(boolean requireAuthentication) throws IOException {
+        String configYaml = """
+                requireAuthentication: %s
+                """.formatted(requireAuthentication);
+        Config config = parseConfig(configYaml);
+        assertThat(config.requireAuthentication()).isEqualTo(requireAuthentication);
+    }
+
+    @Test
+    void deserializeSubjectBuilder() throws IOException {
+        String configYaml = """
+                subjectBuilder: MockSubjectBuilderService
+                """;
+        Config config = parseConfig(configYaml);
+        assertThat(config.subjectBuilder()).isEqualTo("MockSubjectBuilderService");
+    }
+
+    // covers the plugin annotations on io.kroxylicious.filters.sasl.inspection.Config
+    @Test
+    void deserializeSubjectBuilderConfig() throws IOException {
+        String configYaml = """
+                subjectBuilder: MockSubjectBuilderService
+                subjectBuilderConfig:
+                  arbitraryProperty: foo
+                """;
+        Config config = parseConfig(configYaml);
+        assertThat(config.subjectBuilderConfig()).isInstanceOf(MockSubjectBuilderService.Config.class);
+    }
+
+    private static Config parseConfig(String input) throws IOException {
+        return ConfigParser.createObjectMapper().reader().readValue(input, Config.class);
+    }
+}

kroxylicious-filters/kroxylicious-sasl-inspection/src/test/java/io/kroxylicious/filters/sasl/inspection/MockSubjectBuilderService.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.filters.sasl.inspection;
+
+import io.kroxylicious.proxy.authentication.SaslSubjectBuilder;
+import io.kroxylicious.proxy.authentication.SaslSubjectBuilderService;
+import io.kroxylicious.proxy.plugin.Plugin;
+
+import edu.umd.cs.findbugs.annotations.NonNull;
+
+@Plugin(configType = MockSubjectBuilderService.Config.class)
+public class MockSubjectBuilderService implements SaslSubjectBuilderService<MockSubjectBuilderService.Config> {
+    @Override
+    public void initialize(@NonNull Config config) {
+        throw new IllegalStateException("not implemented");
+    }
+
+    @Override
+    public SaslSubjectBuilder build() {
+        throw new IllegalStateException("not implemented");
+    }
+
+    public record Config(String arbitraryProperty) {}
+
+}

kroxylicious-filters/kroxylicious-sasl-inspection/src/test/resources/META-INF/services/io.kroxylicious.proxy.authentication.SaslSubjectBuilderService

@@ -0,0 +1,6 @@
+#
+# Copyright Kroxylicious Authors.
+#
+# Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+#
+io.kroxylicious.filters.sasl.inspection.MockSubjectBuilderService

kroxylicious-integration-test-support/src/main/java/io/kroxylicious/proxy/config/BuilderConfig.java

@@ -18,6 +18,7 @@
         "io.kroxylicious.proxy.config.VirtualClusterGateway",
         "io.kroxylicious.proxy.config.PortIdentifiesNodeIdentificationStrategy",
         "io.kroxylicious.proxy.config.SniHostIdentifiesNodeIdentificationStrategy",
+        "io.kroxylicious.proxy.config.TransportSubjectBuilderConfig",
         "io.kroxylicious.proxy.config.NamedRange",
         "io.kroxylicious.proxy.config.admin.ManagementConfiguration",
         "io.kroxylicious.proxy.config.admin.EndpointsConfiguration",

kroxylicious-integration-test-support/src/main/java/io/kroxylicious/proxy/config/TransportSubjectBuilderDefinitionBuilder.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.proxy.config;
+
+import java.util.Map;
+
+import io.kroxylicious.proxy.authentication.TransportSubjectBuilderService;
+
+public class TransportSubjectBuilderDefinitionBuilder extends AbstractDefinitionBuilder<TransportSubjectBuilderConfig> {
+    public TransportSubjectBuilderDefinitionBuilder(String type) {
+        super(type);
+    }
+
+    @Override
+    protected TransportSubjectBuilderConfig buildInternal(String type, Map<String, Object> config) {
+        ServiceBasedPluginFactoryRegistry registry = new ServiceBasedPluginFactoryRegistry();
+        PluginFactory<TransportSubjectBuilderService> factory = registry.pluginFactory(TransportSubjectBuilderService.class);
+        Class<?> configType = factory.configType(type);
+        return new TransportSubjectBuilderConfig(type, mapper.convertValue(config, configType));
+    }
+}

kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/BaseIT.java

@@ -6,25 +6,32 @@
 
 package io.kroxylicious.proxy;
 
+import java.time.Duration;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
 
 import org.apache.kafka.clients.admin.Admin;
 import org.apache.kafka.clients.admin.CreateTopicsResult;
 import org.apache.kafka.clients.admin.DeleteTopicsResult;
 import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.TopicCollection;
+import org.apache.kafka.common.serialization.Serdes;
 import org.junit.jupiter.api.extension.ExtendWith;
 
 import io.github.nettyplus.leakdetector.junit.NettyLeakDetectorExtension;
 
 import io.kroxylicious.test.tester.KroxyliciousTester;
 import io.kroxylicious.testing.kafka.junit5ext.KafkaClusterExtension;
+import io.kroxylicious.testing.kafka.junit5ext.Topic;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -78,4 +85,28 @@ protected final Producer<String, String> getProducerWithConfig(KroxyliciousTeste
         }
         return tester.producer(producerConfig);
     }
+
+    public static void sendReceiveBatches(KroxyliciousTester tester,
+                                          Topic topic,
+                                          Map<String, Object> producerConfig,
+                                          Map<String, Object> consumerConfig,
+                                          int numBatches,
+                                          BiConsumer<Integer, ConsumerRecords<String, byte[]>> recordConsumer) {
+        try (var producer = tester.producer(producerConfig);
+                var consumer = tester
+                        .consumer(Serdes.String(), Serdes.ByteArray(), consumerConfig)) {
+            int batchNumOneBased = 1;
+            while (batchNumOneBased <= numBatches) {
+                assertThat(producer.send(new ProducerRecord<>(topic.name(), "my-key", "my-value")))
+                        .succeedsWithin(Duration.ofSeconds(5));
+
+                consumer.subscribe(Set.of(topic.name()));
+                var records = consumer.poll(Duration.ofSeconds(10));
+
+                assertThat(records).hasSize(1);
+                recordConsumer.accept(batchNumOneBased, records);
+                batchNumOneBased += 1;
+            }
+        }
+    }
 }

kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/BaseOauthBearerIT.java

@@ -37,7 +37,7 @@
 @ExtendWith(NettyLeakDetectorExtension.class)
 @EnabledIf(value = "isDockerAvailable", disabledReason = "docker unavailable")
 @RestoreSystemProperties
-public class BaseOauthBearerIT {
+public class BaseOauthBearerIT extends BaseIT {
     private static final int OAUTH_SERVER_PORT = 28089;
 
     // This token is expired