Allow SaslInspection to function as a barrier (kroxylicious#2951)

The `SaslInspection` filter was purely about inspecting the SASL exchange. This meant that any client that was not configured do SASL authentication would be able to make any other requests it likes which the proxy would forward to the broker. A broker configured with a `required` JAAS `LoginModule` would then reject the request. So in a sense no harm done. However:

* It's very confusing when a client not configured to use SASL behaves like this: nothing in the proxy (in particular neither the `SaslInspection` nor `Authentication` filters) will log either the requests or the responses (which will be error responses with `SASL_AUTHENTICATION_FAILED` error code). But this seems like an important security event which the proxy should explicitly know about.
* It also means that those requests will transit the filter chain before being ultimately rejected by the broker. Any side-effects caused by those filters would be applied. It seems likely that this could lead to subtle errors because of the expectation by a person configuring the proxy that the inspection filter prevents this kind of thing.

For these reasons it seems sensible to allow the SASL inspection filter to function as a security barrier (based on configuration), rejecting on its own account any requests except ApiVersions, SaslHandshake and SaslAuthenticate, which are made prior to a successful authentication.

Because SaslInspection has been public API since 0.17.0 we have to disable this behaviour by default.

Signed-off-by: Tom Bentley <[email protected]>

Author: tombentley

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

@@ -19,7 +19,12 @@
  * {@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.
+ * @param requireAuthentication If true then successful authentication will be required before the filter
+ * will forward any requests other than those strictly required to perform SASL authentication.
+ * If false then the filter will forward all requests regardless of whether
+ * 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 Object subjectBuilderConfig,
+                     @Nullable Boolean requireAuthentication) {}

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

@@ -50,13 +50,15 @@ public CompletionStage<Subject> buildSaslSubject(Context context) {
     };
     private @Nullable Map<String, SaslObserverFactory> observerFactoryMap;
     private @Nullable SaslSubjectBuilder subjectBuilder;
+    private boolean authenticationRequired = false;
 
     @Override
     public Void initialize(FilterFactoryContext context,
                            @Nullable Config config)
             throws PluginConfigurationException {
         observerFactoryMap = buildEnabledObserverFactoryMap(context, config);
         subjectBuilder = buildSubjectBuilder(context, config);
+        authenticationRequired = config != null && config.requireAuthentication() != null && config.requireAuthentication();
         return null;
     }
 
@@ -78,7 +80,7 @@ private static SaslSubjectBuilder buildSubjectBuilder(FilterFactoryContext conte
     public Filter createFilter(FilterFactoryContext context, @Nullable Void unused) {
         Objects.requireNonNull(observerFactoryMap);
         Objects.requireNonNull(subjectBuilder);
-        return new SaslInspectionFilter(observerFactoryMap, subjectBuilder);
+        return new SaslInspectionFilter(observerFactoryMap, subjectBuilder, authenticationRequired);
     }
 
     private Map<String, SaslObserverFactory> buildEnabledObserverFactoryMap(FilterFactoryContext context,

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

@@ -20,6 +20,7 @@
 import org.apache.kafka.common.message.SaslAuthenticateResponseData;
 import org.apache.kafka.common.message.SaslHandshakeRequestData;
 import org.apache.kafka.common.message.SaslHandshakeResponseData;
+import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.protocol.ApiMessage;
 import org.apache.kafka.common.protocol.Errors;
 import org.slf4j.Logger;
@@ -30,11 +31,10 @@
 import io.kroxylicious.proxy.authentication.Subject;
 import io.kroxylicious.proxy.authentication.SubjectBuildingException;
 import io.kroxylicious.proxy.filter.FilterContext;
+import io.kroxylicious.proxy.filter.RequestFilter;
 import io.kroxylicious.proxy.filter.RequestFilterResult;
 import io.kroxylicious.proxy.filter.ResponseFilterResult;
-import io.kroxylicious.proxy.filter.SaslAuthenticateRequestFilter;
 import io.kroxylicious.proxy.filter.SaslAuthenticateResponseFilter;
-import io.kroxylicious.proxy.filter.SaslHandshakeRequestFilter;
 import io.kroxylicious.proxy.filter.SaslHandshakeResponseFilter;
 import io.kroxylicious.proxy.tag.VisibleForTesting;
 import io.kroxylicious.proxy.tls.ClientTlsContext;
@@ -53,9 +53,8 @@
  */
 class SaslInspectionFilter
         implements
-        SaslHandshakeRequestFilter,
+        RequestFilter,
         SaslHandshakeResponseFilter,
-        SaslAuthenticateRequestFilter,
         SaslAuthenticateResponseFilter {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(SaslInspectionFilter.class);
@@ -65,20 +64,50 @@ class SaslInspectionFilter
 
     private final Map<String, SaslObserverFactory> observerFactoryMap;
     private final SaslSubjectBuilder subjectBuilder;
+    private final boolean authenticationRequired;
 
     private State currentState = State.start();
 
     SaslInspectionFilter(Map<String, SaslObserverFactory> mechanismFactories,
-                         SaslSubjectBuilder subjectBuilder) {
+                         SaslSubjectBuilder subjectBuilder,
+                         boolean authenticationRequired) {
         this.observerFactoryMap = Objects.requireNonNull(mechanismFactories, "mechanismFactories");
         this.subjectBuilder = Objects.requireNonNull(subjectBuilder, "subjectBuilder");
+        this.authenticationRequired = authenticationRequired;
     }
 
     @Override
-    public CompletionStage<RequestFilterResult> onSaslHandshakeRequest(short apiVersion,
-                                                                       RequestHeaderData header,
-                                                                       SaslHandshakeRequestData request,
-                                                                       FilterContext context) {
+    public CompletionStage<RequestFilterResult> onRequest(ApiKeys apiKey, RequestHeaderData header, ApiMessage request, FilterContext context) {
+        return switch (apiKey) {
+            case API_VERSIONS -> context.forwardRequest(header, request);
+            case SASL_AUTHENTICATE -> onSaslAuthenticateRequest(header.requestApiVersion(), header, (SaslAuthenticateRequestData) request, context);
+            case SASL_HANDSHAKE -> onSaslHandshakeRequest(header, (SaslHandshakeRequestData) request, context);
+            default -> {
+                if (!authenticationRequired || currentState.clientIsAuthenticated()) {
+                    yield context.forwardRequest(header, request);
+                }
+                else {
+                    String disposition;
+                    if (currentState instanceof State.RequiringAuthenticateRequest) {
+                        disposition = "attempted";
+                    }
+                    else {
+                        disposition = "completed";
+                    }
+                    LOGGER.info("{}: Client attempted {} request without having {} SASL authentication", context.sessionId(), apiKey, disposition);
+
+                    yield context.requestFilterResultBuilder()
+                            .errorResponse(header, request, Errors.SASL_AUTHENTICATION_FAILED.exception())
+                            .withCloseConnection()
+                            .completed();
+                }
+            }
+        };
+    }
+
+    CompletionStage<RequestFilterResult> onSaslHandshakeRequest(RequestHeaderData header,
+                                                                SaslHandshakeRequestData request,
+                                                                FilterContext context) {
         if (currentState instanceof State.ExpectingHandshakeRequestState handshakeRequestState) {
             var saslObserverFactory = observerFactoryMap.get(request.mechanism());
             SaslObserver saslObserver;
@@ -177,11 +206,10 @@ private CompletionStage<ResponseFilterResult> processFailedHandshakeResponse(Res
                 .completed();
     }
 
-    @Override
-    public CompletionStage<RequestFilterResult> onSaslAuthenticateRequest(short apiVersion,
-                                                                          RequestHeaderData header,
-                                                                          SaslAuthenticateRequestData request,
-                                                                          FilterContext context) {
+    CompletionStage<RequestFilterResult> onSaslAuthenticateRequest(short apiVersion,
+                                                                   RequestHeaderData header,
+                                                                   SaslAuthenticateRequestData request,
+                                                                   FilterContext context) {
         if (this.currentState instanceof State.RequiringAuthenticateRequest state) {
             try {
                 var acquiredAuthorizationId = state.saslObserver().clientResponse(request.authBytes());
@@ -229,8 +257,23 @@ public CompletionStage<RequestFilterResult> onSaslAuthenticateRequest(short apiV
     @Override
     public CompletionStage<ResponseFilterResult> onSaslAuthenticateResponse(short apiVersion,
                                                                             ResponseHeaderData header,
-                                                                            SaslAuthenticateResponseData response,
+                                                                            SaslAuthenticateResponseData brokerResponse,
                                                                             FilterContext context) {
+        return this.handleSaslAuthenticateResponse(header, brokerResponse, context)
+                .thenApply(clientFacingResponse -> {
+                    if (currentState.clientIsAuthenticated()) {
+                        LOGGER.info("Authentication successful");
+                    }
+                    else {
+                        LOGGER.info("Authentication NOT successful (yet)");
+                    }
+                    return clientFacingResponse;
+                });
+    }
+
+    CompletionStage<ResponseFilterResult> handleSaslAuthenticateResponse(ResponseHeaderData header,
+                                                                         SaslAuthenticateResponseData response,
+                                                                         FilterContext context) {
         if (this.currentState instanceof State.AwaitingAuthenticateResponse state) {
             try {
                 state.saslObserver().serverChallenge(response.authBytes());

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

@@ -38,6 +38,10 @@ static RequiringHandshakeRequest start() {
         return new RequiringHandshakeRequest();
     }
 
+    default boolean clientIsAuthenticated() {
+        return false;
+    }
+
     sealed interface ExpectingHandshakeRequestState extends State permits RequiringHandshakeRequest, AllowingHandshakeRequest {
         /**
          * Transition to the next state.
@@ -151,6 +155,11 @@ final class AllowingHandshakeRequest implements ExpectingHandshakeRequestState {
         private AllowingHandshakeRequest() {
         }
 
+        @Override
+        public boolean clientIsAuthenticated() {
+            return true;
+        }
+
         /**
          * Transition to the next state.
          *
@@ -171,6 +180,11 @@ public AwaitingHandshakeResponse nextState(SaslObserver saslObserver) {
     final class DisallowingAuthenticateRequest implements State {
         private DisallowingAuthenticateRequest() {
         }
+
+        @Override
+        public boolean clientIsAuthenticated() {
+            return true;
+        }
     }
 
     enum NegotiationType {