SASL plain termination changes (ITs only) (kroxylicious#2776)

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

Author: tombentley

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

@@ -74,6 +74,7 @@ private static void doAThing(KafkaCluster cluster,
         NamedFilterDefinition saslTermination = new NamedFilterDefinitionBuilder(
                 SaslPlainTermination.class.getName(),
                 SaslPlainTermination.class.getName())
+                .withConfig("userNameToPassword", Map.of("alice", "alice-secret"))
                 .build();
         NamedFilterDefinition lawyer = new NamedFilterDefinitionBuilder(
                 ClientTlsAwareLawyer.class.getName(),

kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/testplugins/SaslPlainTermination.java

@@ -6,23 +6,93 @@
 
 package io.kroxylicious.proxy.testplugins;
 
+import java.util.List;
+import java.util.Map;
+
+import javax.security.auth.callback.Callback;
+import javax.security.auth.callback.NameCallback;
+import javax.security.auth.callback.UnsupportedCallbackException;
+import javax.security.auth.login.AppConfigurationEntry;
+
+import org.apache.kafka.common.security.auth.AuthenticateCallbackHandler;
+import org.apache.kafka.common.security.plain.PlainAuthenticateCallback;
+import org.apache.kafka.common.utils.Utils;
+
 import io.kroxylicious.proxy.filter.Filter;
 import io.kroxylicious.proxy.filter.FilterFactory;
 import io.kroxylicious.proxy.filter.FilterFactoryContext;
 import io.kroxylicious.proxy.plugin.Plugin;
 import io.kroxylicious.proxy.plugin.PluginConfigurationException;
+import io.kroxylicious.proxy.plugin.Plugins;
+
+import edu.umd.cs.findbugs.annotations.Nullable;
 
-@Plugin(configType = Void.class)
+@Plugin(configType = SaslPlainTerminationConfig.class)
 public class SaslPlainTermination
-        implements FilterFactory<Void, Void> {
+        implements FilterFactory<SaslPlainTerminationConfig, SaslPlainTermination.PasswordVerifier> {
+
+    interface PasswordVerifier extends AuthenticateCallbackHandler {
+    }
+
+    /**
+     * This password verifier is insecure in the sense that the user credentials are stored (and configured!) in plaintext.
+     */
+    static class InsecurePasswordVerifier implements PasswordVerifier {
+        private final Map<String, String> userNameToPassword;
+
+        InsecurePasswordVerifier(Map<String, String> userNameToPassword) {
+            this.userNameToPassword = userNameToPassword;
+        }
+
+        @Override
+        public void configure(Map<String, ?> configs, String saslMechanism, List<AppConfigurationEntry> jaasConfigEntries) {
+            if (!"PLAIN".equals(saslMechanism)) {
+                throw new IllegalStateException("This verifier only supports PLAIN authentication");
+            }
+        }
+
+        @Override
+        public void close() {
+
+        }
+
+        @Override
+        public void handle(Callback[] callbacks) throws UnsupportedCallbackException {
+            String username = null;
+            for (Callback callback : callbacks) {
+                if (callback instanceof NameCallback) {
+                    username = ((NameCallback) callback).getDefaultName();
+                }
+                else if (callback instanceof PlainAuthenticateCallback) {
+                    PlainAuthenticateCallback plainCallback = (PlainAuthenticateCallback) callback;
+                    boolean authenticated = authenticate(username, plainCallback.password());
+                    plainCallback.authenticated(authenticated);
+                }
+                else {
+                    throw new UnsupportedCallbackException(callback);
+                }
+            }
+        }
+
+        protected boolean authenticate(@Nullable String username, char[] password) {
+            if (username == null) {
+                return false;
+            }
+            else {
+                String expectedPassword = userNameToPassword.get(username);
+                return expectedPassword != null && Utils.isEqualConstantTime(password, expectedPassword.toCharArray());
+            }
+        }
+    }
 
     @Override
-    public Void initialize(FilterFactoryContext context, Void config) throws PluginConfigurationException {
-        return null;
+    public PasswordVerifier initialize(FilterFactoryContext context, SaslPlainTerminationConfig config) throws PluginConfigurationException {
+        Plugins.requireConfig(this, config);
+        return new InsecurePasswordVerifier(config.userNameToPassword());
     }
 
     @Override
-    public Filter createFilter(FilterFactoryContext context, Void initializationData) {
-        return new SaslPlainTerminationFilter();
+    public Filter createFilter(FilterFactoryContext context, PasswordVerifier passwordVerifier) {
+        return new SaslPlainTerminationFilter(passwordVerifier);
     }
 }

kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/testplugins/SaslPlainTerminationConfig.java

@@ -0,0 +1,11 @@
+/*
+ * 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.testplugins;
+
+import java.util.Map;
+
+public record SaslPlainTerminationConfig(Map<String, String> userNameToPassword) {}

kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/testplugins/SaslPlainTerminationFilter.java

@@ -7,10 +7,8 @@
 package io.kroxylicious.proxy.testplugins;
 
 import java.util.List;
-import java.util.Map;
 import java.util.concurrent.CompletionStage;
 
-import javax.security.auth.login.AppConfigurationEntry;
 import javax.security.sasl.Sasl;
 import javax.security.sasl.SaslException;
 import javax.security.sasl.SaslServer;
@@ -25,15 +23,12 @@
 import org.apache.kafka.common.protocol.ApiMessage;
 import org.apache.kafka.common.protocol.Errors;
 import org.apache.kafka.common.security.auth.AuthenticateCallbackHandler;
-import org.apache.kafka.common.security.plain.PlainLoginModule;
-import org.apache.kafka.common.security.plain.internals.PlainServerCallbackHandler;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import io.kroxylicious.proxy.filter.FilterContext;
 import io.kroxylicious.proxy.filter.RequestFilter;
 import io.kroxylicious.proxy.filter.RequestFilterResult;
-import io.kroxylicious.proxy.internal.KafkaAuthnHandler;
 
 /**
  * A minimal SASL termination filter supporting the {@code PLAIN} mechanism only.
@@ -47,17 +42,12 @@ public class SaslPlainTerminationFilter
 
     private static final Logger LOGGER = LoggerFactory.getLogger(SaslPlainTerminationFilter.class);
 
+    private final SaslPlainTermination.PasswordVerifier passwordVerifier;
+
     private SaslServer saslServer;
 
-    private AuthenticateCallbackHandler saslPlainCallbackHandler(String user,
-                                                                 String password) {
-        PlainServerCallbackHandler plainServerCallbackHandler = new PlainServerCallbackHandler();
-        plainServerCallbackHandler.configure(Map.of(),
-                KafkaAuthnHandler.SaslMechanism.PLAIN.mechanismName(),
-                List.of(new AppConfigurationEntry(PlainLoginModule.class.getName(),
-                        AppConfigurationEntry.LoginModuleControlFlag.REQUIRED,
-                        Map.of("user_" + user, password))));
-        return plainServerCallbackHandler;
+    public SaslPlainTerminationFilter(SaslPlainTermination.PasswordVerifier passwordVerifier) {
+        this.passwordVerifier = passwordVerifier;
     }
 
     CompletionStage<RequestFilterResult> onSaslHandshakeRequest(SaslHandshakeRequestData request,
@@ -66,7 +56,7 @@ CompletionStage<RequestFilterResult> onSaslHandshakeRequest(SaslHandshakeRequest
         AuthenticateCallbackHandler cbh = null;
         List<String> supportedMechanisms;
         if ("PLAIN".equals(request.mechanism())) {
-            cbh = saslPlainCallbackHandler("alice", "alice-secret");
+            cbh = this.passwordVerifier;
             errorCode = Errors.NONE;
             supportedMechanisms = List.of();
         }