Use SaslSubjectBuilder in the SASL inspector and PLAIN terminator (kroxylicious#2904)

* kroxylicious-filters: Use SubjectBuilder in SASL inspector (and IT's terminators)
* Add SubjectBuildingException
* Factor out PrincipalAdderConf and Map
* Log use of default subject builder

Signed-off-by: Tom Bentley <tbentley@redhat.com>

Author: tombentley

kroxylicious-api/src/main/java/io/kroxylicious/proxy/authentication/SaslSubjectBuilder.java

@@ -28,9 +28,10 @@ public interface SaslSubjectBuilder {
 
     /**
      * Returns an asynchronous result which completes with the {@code Subject} built
-     * from the
-     * @param context
-     * @return
+     * from the given {@code context}.
+     * @param context The context of a successful SASL authentication.
+     * @return The Subject. The returned stage should fail with an {@link SubjectBuildingException} if the builder was not able to
+     * build a subject.
      */
     CompletionStage<Subject> buildSaslSubject(SaslSubjectBuilder.Context context);
 

kroxylicious-api/src/main/java/io/kroxylicious/proxy/authentication/SubjectBuildingException.java

@@ -0,0 +1,20 @@
+/*
+ * 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.authentication;
+
+/**
+ * An exception to be thrown if a {@link Subject} cannot be built.
+ */
+public class SubjectBuildingException extends RuntimeException {
+    public SubjectBuildingException(String message) {
+        super(message);
+    }
+
+    public SubjectBuildingException(String message, Throwable cause) {
+        super(message, cause);
+    }
+}

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

@@ -71,6 +71,11 @@
             <artifactId>junit-jupiter-params</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>io.kroxylicious</groupId>
+            <artifactId>kroxylicious-runtime</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
 </project>

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

@@ -8,12 +8,18 @@
 
 import java.util.Set;
 
+import edu.umd.cs.findbugs.annotations.Nullable;
+
 /**
- * Config for the Sasl Initiation Filter.
+ * Configuration for the Sasl Initiation Filter.
  *
  * @param enabledMechanisms set of SASL mechanisms to enable. Refer to the mechanism by its
  * IANA registered name.  If enabledMechanisms is null, the system will automatically enable
  * the mechanism from all {@link SaslObserverFactory}s with
  * {@link SaslObserverFactory#transmitsCredentialInCleartext()} values of false.
+ * @param subjectBuilder The name of a plugin class implementing {@link io.kroxylicious.proxy.authentication.SaslSubjectBuilderService}
+ * @param subjectBuilderConfig The configuration for the SaslSubjectBuilderService.
  */
-public record Config(Set<String> enabledMechanisms) {}
+public record Config(@Nullable Set<String> enabledMechanisms,
+                     @Nullable String subjectBuilder,
+                     @Nullable Object subjectBuilderConfig) {}

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

@@ -10,10 +10,20 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
 import java.util.function.Function;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.kroxylicious.proxy.authentication.SaslSubjectBuilder;
+import io.kroxylicious.proxy.authentication.SaslSubjectBuilderService;
+import io.kroxylicious.proxy.authentication.Subject;
+import io.kroxylicious.proxy.authentication.User;
 import io.kroxylicious.proxy.filter.Filter;
 import io.kroxylicious.proxy.filter.FilterFactory;
 import io.kroxylicious.proxy.filter.FilterFactoryContext;
@@ -30,23 +40,49 @@
 @Plugin(configType = Config.class)
 public class SaslInspection implements FilterFactory<Config, Void> {
 
+    private static final Logger LOGGER = LoggerFactory.getLogger(SaslInspection.class);
+
+    public static final SaslSubjectBuilder DEFAULT_SUBJECT_BUILDER = new SaslSubjectBuilder() {
+        @Override
+        public CompletionStage<Subject> buildSaslSubject(Context context) {
+            return CompletableFuture.completedStage(new Subject(Set.of(new User(context.clientSaslContext().authorizationId()))));
+        }
+    };
     private @Nullable Map<String, SaslObserverFactory> observerFactoryMap;
+    private @Nullable SaslSubjectBuilder subjectBuilder;
 
     @Override
     public Void initialize(FilterFactoryContext context,
                            @Nullable Config config)
             throws PluginConfigurationException {
         observerFactoryMap = buildEnabledObserverFactoryMap(context, config);
+        subjectBuilder = buildSubjectBuilder(context, config);
         return null;
     }
 
+    @NonNull
+    private static SaslSubjectBuilder buildSubjectBuilder(FilterFactoryContext context,
+                                                          @Nullable Config config) {
+        if (config == null || config.subjectBuilder() == null) {
+            LOGGER.debug("No `subjectBuilder` configured. The default SaslSubjectBuilder will be used.");
+            return DEFAULT_SUBJECT_BUILDER;
+        }
+        else {
+            SaslSubjectBuilderService subjectBuilderFactory = context.pluginInstance(SaslSubjectBuilderService.class, config.subjectBuilder());
+            subjectBuilderFactory.initialize(config.subjectBuilderConfig());
+            return subjectBuilderFactory.build();
+        }
+    }
+
     @Override
-    public Filter createFilter(FilterFactoryContext context, Void unused) {
+    public Filter createFilter(FilterFactoryContext context, @Nullable Void unused) {
         Objects.requireNonNull(observerFactoryMap);
-        return new SaslInspectionFilter(observerFactoryMap);
+        Objects.requireNonNull(subjectBuilder);
+        return new SaslInspectionFilter(observerFactoryMap, subjectBuilder);
     }
 
-    private Map<String, SaslObserverFactory> buildEnabledObserverFactoryMap(FilterFactoryContext context, @Nullable Config config) {
+    private Map<String, SaslObserverFactory> buildEnabledObserverFactoryMap(FilterFactoryContext context,
+                                                                            @Nullable Config config) {
         var allNames = context.pluginImplementationNames(SaslObserverFactory.class);
         var allMap = allNames
                 .stream()

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

@@ -9,6 +9,7 @@
 import java.util.ArrayList;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.concurrent.CompletionStage;
 
 import javax.security.sasl.SaslException;
@@ -24,6 +25,10 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import io.kroxylicious.proxy.authentication.ClientSaslContext;
+import io.kroxylicious.proxy.authentication.SaslSubjectBuilder;
+import io.kroxylicious.proxy.authentication.Subject;
+import io.kroxylicious.proxy.authentication.SubjectBuildingException;
 import io.kroxylicious.proxy.filter.FilterContext;
 import io.kroxylicious.proxy.filter.RequestFilterResult;
 import io.kroxylicious.proxy.filter.ResponseFilterResult;
@@ -32,7 +37,9 @@
 import io.kroxylicious.proxy.filter.SaslHandshakeRequestFilter;
 import io.kroxylicious.proxy.filter.SaslHandshakeResponseFilter;
 import io.kroxylicious.proxy.tag.VisibleForTesting;
+import io.kroxylicious.proxy.tls.ClientTlsContext;
 
+import edu.umd.cs.findbugs.annotations.NonNull;
 import edu.umd.cs.findbugs.annotations.Nullable;
 
 /**
@@ -57,12 +64,14 @@ class SaslInspectionFilter
     static final String PROBE_UPSTREAM = PROBING_SASL_OBSERVER.mechanismName();
 
     private final Map<String, SaslObserverFactory> observerFactoryMap;
+    private final SaslSubjectBuilder subjectBuilder;
 
     private State currentState = State.start();
 
-    SaslInspectionFilter(Map<String, SaslObserverFactory> mechanismFactories) {
-        Objects.requireNonNull(mechanismFactories, "mechanismFactories");
-        this.observerFactoryMap = mechanismFactories;
+    SaslInspectionFilter(Map<String, SaslObserverFactory> mechanismFactories,
+                         SaslSubjectBuilder subjectBuilder) {
+        this.observerFactoryMap = Objects.requireNonNull(mechanismFactories, "mechanismFactories");
+        this.subjectBuilder = Objects.requireNonNull(subjectBuilder, "subjectBuilder");
     }
 
     @Override
@@ -253,7 +262,6 @@ private CompletionStage<ResponseFilterResult> processSuccessfulAuthenticateRespo
                 // Ordinarily, we never expect to be here. The sasl negotiation ought to be yielded a authorizationId before
                 // the negotiation is finished.
                 return closeConnectionWithResponse(header, response.setErrorCode(Errors.ILLEGAL_SASL_STATE.code()), context);
-
             }
 
             var expiredCredential = state.saslObserver().zeroLengthSessionImpliesAuthnFailure() && response.sessionLifetimeMs() == 0;
@@ -267,19 +275,66 @@ private CompletionStage<ResponseFilterResult> processSuccessfulAuthenticateRespo
                 context.clientSaslAuthenticationFailure(state.saslObserver().mechanismName(), authorizationIdFromClient, new SaslException("expired credential"));
             }
             else {
-                LOGGER.atInfo()
-                        .setMessage("Server accepts SASL credentials for client on channel {}, announcing that client has authorizationId {}")
-                        .addArgument(context::channelDescriptor)
-                        .addArgument(authorizationIdFromClient)
-                        .log();
-                context.clientSaslAuthenticationSuccess(saslObserver.mechanismName(), authorizationIdFromClient);
-
+                return buildSubjectAndAnnounce(header, response, context, state, saslObserver, authorizationIdFromClient);
             }
         }
         currentState = state.nextState(saslObserver.isFinished());
         return context.forwardResponse(header, response);
     }
 
+    @NonNull
+    private CompletionStage<ResponseFilterResult> buildSubjectAndAnnounce(ResponseHeaderData header,
+                                                                          SaslAuthenticateResponseData response,
+                                                                          FilterContext context,
+                                                                          State.AwaitingAuthenticateResponse state,
+                                                                          SaslObserver saslObserver,
+                                                                          String authorizationIdFromClient) {
+        CompletionStage<Subject> subjectCompletionStage = subjectBuilder.buildSaslSubject(new SaslSubjectBuilder.Context() {
+            @Override
+            public Optional<ClientTlsContext> clientTlsContext() {
+                return context.clientTlsContext();
+            }
+
+            @Override
+            public ClientSaslContext clientSaslContext() {
+                return new ClientSaslContext() {
+                    @Override
+                    public String mechanismName() {
+                        return saslObserver.mechanismName();
+                    }
+
+                    @Override
+                    public String authorizationId() {
+                        return authorizationIdFromClient;
+                    }
+                };
+            }
+        });
+
+        return subjectCompletionStage.thenCompose(subject -> {
+            LOGGER.atInfo()
+                    .setMessage("Server accepts SASL credentials for client on channel {}, announcing that client has authorizationId {}")
+                    .addArgument(context::channelDescriptor)
+                    .addArgument(authorizationIdFromClient)
+                    .log();
+            context.clientSaslAuthenticationSuccess(saslObserver.mechanismName(),
+                    subject);
+            currentState = state.nextState(saslObserver.isFinished());
+            return context.forwardResponse(header, response);
+        }).exceptionallyCompose(throwable -> {
+            Exception e;
+            if (throwable instanceof SubjectBuildingException exception) {
+                e = exception;
+            }
+            else {
+                e = new SubjectBuildingException("SaslSubjectBuilder " + subjectBuilder.getClass() + " threw an unexpected exception", throwable);
+            }
+            context.clientSaslAuthenticationFailure(saslObserver.mechanismName(),
+                    authorizationIdFromClient, e);
+            return closeConnectionWithResponse(header, response.setErrorCode(Errors.ILLEGAL_SASL_STATE.code()), context);
+        });
+    }
+
     private CompletionStage<ResponseFilterResult> processFailedAuthentication(ResponseHeaderData header,
                                                                               SaslAuthenticateResponseData response,
                                                                               FilterContext context,