Repository: kroxylicious/kroxylicious (java)

Issue:
Filters cannot intercept SASL requests
**Describe the bug**
If a user implements `SaslAuthenticateRequestFilter` or `SaslHandshakeRequestFilter` their filter request code is not hit because the `SaslDecodePredicate` currently [always decides](https://github.com/kroxylicious/kroxylicious/blob/3894a79d68dc28020126ef91548d41fadfb3e9df/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/SaslDecodePredicate.java#L55) not to decode `SASL_HANDSHAKE` or `SASL_AUTHENTICATE`. (handleSasl is always `false` at the moment as KafkaProxy [configures the initializer with an empty map](https://github.com/kroxylicious/kroxylicious/blob/3894a79d68dc28020126ef91548d41fadfb3e9df/kroxylicious/src/main/java/io/kroxylicious/proxy/KafkaProxy.java#L161) of SASL mechanisms)

Maybe the logic should be an or. We decode if a filter targets these APIs or we have SASL mechanisms configured.

Pull request description:
Enable filters to intercept/mutate SASL requests when proxy is not ha…
### Type of change

- Bugfix

### Description

The proxy is only considering if the proxy is handling SASL offload when deciding if it should decode SASL requests. If no mechanisms are configured (currently hardcoded to no mechanisms), then SASL requests are never decoded even if you have a SaslHandshakeRequestFilter or SaslAuthenticateRequestFilter defined.

So we make two behaviour changes:
* On the first RPC from the client, before we have a delegate set on the `SaslDecodePredicate` we always decode all RPCs. This means the SASL requests will be available for filtering on the first hit.
* When the delegate is set, we decode if the delegate says we should decode _or_ if kroxylicious is handling SASL and it's a SASL request.

### Additional Context

Closes #261

Patch:
diff --git a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/SaslDecodePredicate.java b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/SaslDecodePredicate.java
index 88a753de..f4fe0629 100644
--- a/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/SaslDecodePredicate.java
+++ b/kroxylicious/src/main/java/io/kroxylicious/proxy/internal/SaslDecodePredicate.java
@@ -43,21 +43,22 @@ class SaslDecodePredicate implements DecodePredicate {
 
     @Override
     public boolean shouldDecodeRequest(ApiKeys apiKey, short apiVersion) {
-        boolean result;
         if (apiKey == ApiKeys.API_VERSIONS) {
             // TODO For now let's assume we need to always decode this, since the NetHandler
             // currently does this. At some point we'll need a way to figure out the mutual intersection
             // of api versions over all backend clusters plus the proxy itself.
-            result = true;
+            return true;
         }
-        else if (apiKey == ApiKeys.SASL_HANDSHAKE
-                || apiKey == ApiKeys.SASL_AUTHENTICATE) {
-            result = handleSasl;
+        if (delegate == null) {
+            // on the first request, before the delegate is set decode everything in case a filter wants
+            // to intercept it
+            return true;
         }
-        else {
-            result = delegate == null || delegate.shouldDecodeRequest(apiKey, apiVersion);
-        }
-        return result;
+        return isAuthenticationOffloaded(apiKey) || delegate.shouldDecodeRequest(apiKey, apiVersion);
+    }
+
+    private boolean isAuthenticationOffloaded(ApiKeys apiKey) {
+        return handleSasl && (apiKey == ApiKeys.SASL_HANDSHAKE || apiKey == ApiKeys.SASL_AUTHENTICATE);
     }
 
     @Override


Test patch:
diff --git a/integrationtests/src/test/java/io/kroxylicious/proxy/ProxyRpcTest.java b/integrationtests/src/test/java/io/kroxylicious/proxy/ProxyRpcTest.java
index 0c28cccf..a4ca03c0 100644
--- a/integrationtests/src/test/java/io/kroxylicious/proxy/ProxyRpcTest.java
+++ b/integrationtests/src/test/java/io/kroxylicious/proxy/ProxyRpcTest.java
@@ -34,8 +34,6 @@ import io.kroxylicious.test.server.MockServer;
 import static io.kroxylicious.proxy.Utils.startProxy;
 import static org.apache.kafka.common.protocol.ApiKeys.API_VERSIONS;
 import static org.apache.kafka.common.protocol.ApiKeys.CONTROLLED_SHUTDOWN;
-import static org.apache.kafka.common.protocol.ApiKeys.SASL_AUTHENTICATE;
-import static org.apache.kafka.common.protocol.ApiKeys.SASL_HANDSHAKE;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.fail;
 
@@ -102,10 +100,6 @@ public class ProxyRpcTest {
                 // controlled shutdown is the only usage of a version 0 header schema which doesn't have clientId
                 expected = "";
             }
-            else if (apiAndVersion.keys() == SASL_AUTHENTICATE || apiAndVersion.keys() == SASL_HANDSHAKE) {
-                // sasl requests have special logic in kroxy, and they do not get Filtered
-                expected = "mockClientId";
-            }
             else {
                 expected = "fixed";
             }
diff --git a/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/SaslDecodePredicateTest.java b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/SaslDecodePredicateTest.java
new file mode 100644
index 00000000..9f7ea7d6
--- /dev/null
+++ b/kroxylicious/src/test/java/io/kroxylicious/proxy/internal/SaslDecodePredicateTest.java
@@ -0,0 +1,146 @@
+/*
+ * 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.internal;
+
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+import io.kroxylicious.proxy.internal.codec.DecodePredicate;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class SaslDecodePredicateTest {
+
+    private static final DecodePredicate TARGET_NOTHING = new DecodePredicate() {
+        @Override
+        public boolean shouldDecodeRequest(ApiKeys apiKey, short apiVersion) {
+            return false;
+        }
+
+        @Override
+        public boolean shouldDecodeResponse(ApiKeys apiKey, short apiVersion) {
+            return false;
+        }
+    };
+
+    private static final DecodePredicate TARGET_ALL = new DecodePredicate() {
+        @Override
+        public boolean shouldDecodeRequest(ApiKeys apiKey, short apiVersion) {
+            return true;
+        }
+
+        @Override
+        public boolean shouldDecodeResponse(ApiKeys apiKey, short apiVersion) {
+            return true;
+        }
+    };
+
+    private SaslDecodePredicate predicate;
+
+    @Test
+    public void testApiVersionAlwaysDecoded_SaslNotHandledAndBeforeDelegateSet() {
+        givenSaslHandlingDisabled();
+        assertPredicateTargetsRequestKey(ApiKeys.API_VERSIONS);
+    }
+
+    @Test
+    public void testApiVersionAlwaysDecoded_SaslHandledAndBeforeDelegateSet() {
+        givenSaslHandlingEnabled();
+        assertPredicateTargetsRequestKey(ApiKeys.API_VERSIONS);
+    }
+
+    @Test
+    public void testApiVersionAlwaysDecoded_SaslNotHandledAndDelegateSet() {
+        givenSaslHandlingDisabled();
+        givenDelegateTargetsNothing();
+        assertPredicateTargetsRequestKey(ApiKeys.API_VERSIONS);
+    }
+
+    @Test
+    public void testApiVersionAlwaysDecoded_SaslHandledAndDelegateSet() {
+        givenSaslHandlingEnabled();
+        givenDelegateTargetsNothing();
+        assertPredicateTargetsRequestKey(ApiKeys.API_VERSIONS);
+    }
+
+    @EnumSource(ApiKeys.class)
+    @ParameterizedTest
+    public void testAllKeysDecodedBeforeDelegateSet(ApiKeys keys) {
+        givenSaslHandlingDisabled();
+        assertPredicateTargetsRequestKey(keys);
+        givenSaslHandlingEnabled();
+        assertPredicateTargetsRequestKey(keys);
+    }
+
+    @Test
+    public void testSaslKeysTargetedForDecodeWithOffloadingAuth() {
+        givenSaslHandlingEnabled();
+        givenDelegateTargetsNothing();
+        assertPredicateTargetsRequestKey(ApiKeys.SASL_AUTHENTICATE);
+        assertPredicateTargetsRequestKey(ApiKeys.SASL_HANDSHAKE);
+    }
+
+    @EnumSource(value = ApiKeys.class, mode = EnumSource.Mode.EXCLUDE, names = { "SASL_AUTHENTICATE", "SASL_HANDSHAKE", "API_VERSIONS" })
+    @ParameterizedTest
+    public void testNonSaslKeysNotTargetedForDecodeWithOffloadingAuthWhenPredicateDeniesThem(ApiKeys apiKeys) {
+        givenSaslHandlingEnabled();
+        givenDelegateTargetsNothing();
+        assertPredicateDoesNotTargetRequestKey(apiKeys);
+    }
+
+    @EnumSource(value = ApiKeys.class, mode = EnumSource.Mode.EXCLUDE, names = { "API_VERSIONS" })
+    @ParameterizedTest
+    public void testAllKeysCanBeDeniedByDelegateWhenNotOffloadingAuth(ApiKeys apiKeys) {
+        givenSaslHandlingDisabled();
+        givenDelegateTargetsNothing();
+        assertPredicateDoesNotTargetRequestKey(apiKeys);
+    }
+
+    @EnumSource(value = ApiKeys.class, mode = EnumSource.Mode.EXCLUDE, names = { "API_VERSIONS" })
+    @ParameterizedTest
+    public void testAllKeysCanBeTargetedByDelegateWhenNotOffloadingAuth(ApiKeys apiKeys) {
+        givenSaslHandlingDisabled();
+        givenDelegateTargetsAll();
+        assertPredicateTargetsRequestKey(apiKeys);
+    }
+
+    private void givenDelegateTargetsAll() {
+        givenPredicateDelegateSet(TARGET_ALL);
+    }
+
+    private void givenDelegateTargetsNothing() {
+        givenPredicateDelegateSet(TARGET_NOTHING);
+    }
+
+    private void givenPredicateDelegateSet(DecodePredicate predicate) {
+        this.predicate.setDelegate(predicate);
+    }
+
+    private void givenSaslHandlingDisabled() {
+        givenPredicateWithHandleSasl(false);
+    }
+
+    private void givenSaslHandlingEnabled() {
+        givenPredicateWithHandleSasl(true);
+    }
+
+    private void givenPredicateWithHandleSasl(boolean handleSasl) {
+        predicate = new SaslDecodePredicate(handleSasl);
+    }
+
+    private void assertPredicateTargetsRequestKey(ApiKeys key) {
+        assertTrue(predicate.shouldDecodeRequest(key, key.latestVersion()), "predicate did not target key " + key);
+    }
+
+    private void assertPredicateDoesNotTargetRequestKey(ApiKeys key) {
+        assertFalse(predicate.shouldDecodeRequest(key, key.latestVersion()), "predicate unexpectedly targeted key " + key);
+    }
+
+}
\ No newline at end of file