[security] Make SASL mechanisms extensible

fluss-common/src/main/java/org/apache/fluss/security/auth/sasl/SaslAuthenticatorFactory.java

@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.security.auth.sasl;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.AuthenticationException;
+import org.apache.fluss.security.auth.ClientAuthenticator;
+import org.apache.fluss.security.auth.ServerAuthenticator;
+import org.apache.fluss.security.auth.sasl.authenticator.PlainSaslServerAuthenticator;
+import org.apache.fluss.security.auth.sasl.authenticator.SaslClientAuthenticator;
+import org.apache.fluss.security.auth.sasl.plain.PlainSaslServer;
+
+import java.util.Locale;
+
+/** Factory for creating connection-local SASL authenticators. */
+@Internal
+public final class SaslAuthenticatorFactory {
+    private SaslAuthenticatorFactory() {}
+
+    /** Creates a connection-local client SASL authenticator. */
+    public static ClientAuthenticator createClientAuthenticator(Configuration configuration) {
+        String mechanism = configuration.get(ConfigOptions.CLIENT_SASL_MECHANISM);
+        switch (normalize(mechanism)) {
+            case PlainSaslServer.PLAIN_MECHANISM:
+                return new SaslClientAuthenticator(configuration);
+            default:
+                // TODO: Add OAUTHBEARER client authenticator in a follow-up change.
+                throw unsupportedMechanism(mechanism);
+        }
+    }
+
+    /** Creates a server authenticator for the requested mechanism. */
+    public static ServerAuthenticator createServerAuthenticator(
+            String mechanism, Configuration configuration) {
+        switch (normalize(mechanism)) {
+            case PlainSaslServer.PLAIN_MECHANISM:
+                return new PlainSaslServerAuthenticator(configuration);
+            default:
+                // TODO: Add OAUTHBEARER server authenticator in a follow-up change.
+                throw unsupportedMechanism(mechanism);
+        }
+    }
+
+    /** Returns whether the server supports the requested mechanism. */
+    public static boolean supportsServerMechanism(String mechanism) {
+        // TODO: Add OAUTHBEARER server authenticator in a follow-up change.
+        switch (normalize(mechanism)) {
+            case PlainSaslServer.PLAIN_MECHANISM:
+                return true;
+            default:
+                return false;
+        }
+    }
+
+    private static String normalize(String mechanism) {
+        return mechanism.toUpperCase(Locale.ROOT);
+    }
+
+    private static AuthenticationException unsupportedMechanism(String mechanism) {
+        return new AuthenticationException(
+                "Unable to find a matching SASL mechanism for " + normalize(mechanism));
+    }
+}

fluss-common/src/main/java/org/apache/fluss/security/auth/sasl/authenticator/PlainSaslServerAuthenticator.java

@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.security.auth.sasl.authenticator;
+
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.AuthenticationException;
+import org.apache.fluss.security.acl.FlussPrincipal;
+import org.apache.fluss.security.auth.ServerAuthenticator;
+import org.apache.fluss.security.auth.sasl.jaas.JaasContext;
+import org.apache.fluss.security.auth.sasl.jaas.LoginManager;
+import org.apache.fluss.security.auth.sasl.jaas.SaslServerFactory;
+import org.apache.fluss.security.auth.sasl.plain.PlainSaslServer;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.security.sasl.SaslException;
+import javax.security.sasl.SaslServer;
+
+import java.util.Locale;
+import java.util.Map;
+
+/** A connection-local SASL/PLAIN server authenticator. */
+public final class PlainSaslServerAuthenticator implements ServerAuthenticator {
+    private static final Logger LOG = LoggerFactory.getLogger(PlainSaslServerAuthenticator.class);
+    private static final String SERVER_AUTHENTICATOR_PREFIX = "security.sasl.";
+
+    private final Map<String, String> configs;
+    private SaslServer saslServer;
+
+    public PlainSaslServerAuthenticator(Configuration configuration) {
+        this.configs = configuration.toMap();
+    }
+
+    @Override
+    public String protocol() {
+        return PlainSaslServer.PLAIN_MECHANISM;
+    }
+
+    @Override
+    public void initialize(AuthenticateContext context) {
+        String dynamicJaasConfig = findJaasConfig(context.listenerName());
+        JaasContext contextConfig =
+                JaasContext.loadServerContext(context.listenerName(), dynamicJaasConfig);
+        try {
+            LoginManager loginManager = LoginManager.acquireLoginManager(contextConfig);
+            saslServer =
+                    SaslServerFactory.createSaslServer(
+                            PlainSaslServer.PLAIN_MECHANISM,
+                            context.ipAddress(),
+                            configs,
+                            loginManager,
+                            contextConfig.configurationEntries());
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    @Override
+    public byte[] evaluateResponse(byte[] token) throws AuthenticationException {
+        try {
+            return saslServer.evaluateResponse(token);
+        } catch (SaslException e) {
+            throw new AuthenticationException(
+                    String.format(
+                            "Failed to evaluate SASL response, reason is %s", e.getMessage()));
+        }
+    }
+
+    @Override
+    public boolean isCompleted() {
+        return saslServer != null && saslServer.isComplete();
+    }
+
+    @Override
+    public FlussPrincipal createPrincipal() {
+        return new FlussPrincipal(saslServer.getAuthorizationID(), "User");
+    }
+
+    private String findJaasConfig(String listenerName) {
+        String listenerMechanismKey =
+                String.format(
+                        SERVER_AUTHENTICATOR_PREFIX
+                                + "listener.name.%s.%s."
+                                + JaasContext.SASL_JAAS_CONFIG,
+                        listenerName.toLowerCase(Locale.ROOT),
+                        PlainSaslServer.PLAIN_MECHANISM.toLowerCase(Locale.ROOT));
+        String dynamicJaasConfig = configs.get(listenerMechanismKey);
+        if (dynamicJaasConfig != null && !dynamicJaasConfig.isEmpty()) {
+            return dynamicJaasConfig;
+        }
+
+        String globalMechanismKey =
+                SERVER_AUTHENTICATOR_PREFIX
+                        + PlainSaslServer.PLAIN_MECHANISM.toLowerCase(Locale.ROOT)
+                        + "."
+                        + JaasContext.SASL_JAAS_CONFIG;
+        LOG.debug(
+                "No listener-mechanism JAAS config found for key: '{}'. Falling back to mechanism-level config: '{}'",
+                listenerMechanismKey,
+                globalMechanismKey);
+        dynamicJaasConfig = configs.get(globalMechanismKey);
+        if (dynamicJaasConfig == null || dynamicJaasConfig.isEmpty()) {
+            LOG.warn(
+                    "No mechanism-level JAAS config found for key: '{}'. Falling back to JVM option: -D{}",
+                    globalMechanismKey,
+                    JaasContext.JAVA_LOGIN_CONFIG_PARAM);
+        }
+        return dynamicJaasConfig;
+    }
+}

fluss-common/src/main/java/org/apache/fluss/security/auth/sasl/authenticator/SaslAuthenticationPlugin.java

@@ -22,6 +22,7 @@
 import org.apache.fluss.security.auth.ClientAuthenticator;
 import org.apache.fluss.security.auth.ServerAuthenticationPlugin;
 import org.apache.fluss.security.auth.ServerAuthenticator;
+import org.apache.fluss.security.auth.sasl.SaslAuthenticatorFactory;
 
 /** Authentication plugin for SASL. */
 public class SaslAuthenticationPlugin
@@ -30,7 +31,7 @@ public class SaslAuthenticationPlugin
 
     @Override
     public ClientAuthenticator createClientAuthenticator(Configuration configuration) {
-        return new SaslClientAuthenticator(configuration);
+        return SaslAuthenticatorFactory.createClientAuthenticator(configuration);
     }
 
     @Override

fluss-common/src/main/java/org/apache/fluss/security/auth/sasl/authenticator/SaslServerAuthenticator.java

@@ -21,99 +21,45 @@
 import org.apache.fluss.exception.AuthenticationException;
 import org.apache.fluss.security.acl.FlussPrincipal;
 import org.apache.fluss.security.auth.ServerAuthenticator;
-import org.apache.fluss.security.auth.sasl.jaas.JaasContext;
-import org.apache.fluss.security.auth.sasl.jaas.LoginManager;
+import org.apache.fluss.security.auth.sasl.SaslAuthenticatorFactory;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import javax.security.sasl.SaslException;
-import javax.security.sasl.SaslServer;
-
 import java.util.List;
 import java.util.Locale;
-import java.util.Map;
 import java.util.stream.Collectors;
 
 import static org.apache.fluss.config.ConfigOptions.SERVER_SASL_ENABLED_MECHANISMS_CONFIG;
 import static org.apache.fluss.security.auth.sasl.authenticator.SaslAuthenticationPlugin.SASL_AUTH_PROTOCOL;
-import static org.apache.fluss.security.auth.sasl.jaas.JaasContext.SASL_JAAS_CONFIG;
-import static org.apache.fluss.security.auth.sasl.jaas.SaslServerFactory.createSaslServer;
 
-/** An authenticator that uses SASL to authenticate clients. */
+/** A connection-local authenticator that selects one SASL mechanism. */
 public class SaslServerAuthenticator implements ServerAuthenticator {
     private static final Logger LOG = LoggerFactory.getLogger(SaslServerAuthenticator.class);
-    private static final String SERVER_AUTHENTICATOR_PREFIX = "security.sasl.";
+
+    private final Configuration configuration;
     private final List<String> enabledMechanisms;
-    private SaslServer saslServer;
-    private final Map<String, String> configs;
+
+    private ServerAuthenticator delegate;
 
     public SaslServerAuthenticator(Configuration configuration) {
-        this.configs = configuration.toMap();
+        this.configuration = configuration;
         List<String> enabledMechanisms = configuration.get(SERVER_SASL_ENABLED_MECHANISMS_CONFIG);
         if (enabledMechanisms == null || enabledMechanisms.isEmpty()) {
             throw new IllegalArgumentException("No SASL mechanisms are enabled");
         }
         this.enabledMechanisms =
-                enabledMechanisms.stream().map(String::toUpperCase).collect(Collectors.toList());
+                enabledMechanisms.stream()
+                        .map(mechanism -> mechanism.toUpperCase(Locale.ROOT))
+                        .collect(Collectors.toList());
     }
 
     @Override
     public void initialize(AuthenticateContext context) {
-        String mechanism = context.protocol();
-        String listenerName = context.listenerName();
-        String address = context.ipAddress();
+        String mechanism = context.protocol().toUpperCase(Locale.ROOT);
         matchProtocol(mechanism);
-        // Try to load JAAS config in the following order:
-        // 1. security.sasl.listener.name.{listenerName}.{mechanism}.jaas.config (fine-grained per
-        // listener and mechanism)
-        // 2. security.sasl.{mechanism}.jaas.config (fallback global config for mechanism)
-        // 3. JVM option -Djava.security.auth.login.config (system-level fallback)
-
-        String dynamicJaasConfig;
-
-        // 1. Check listener-specific and mechanism-specific config
-        String listenerMechanismKey =
-                String.format(
-                        SERVER_AUTHENTICATOR_PREFIX + "listener.name.%s.%s." + SASL_JAAS_CONFIG,
-                        listenerName.toLowerCase(Locale.ROOT),
-                        mechanism.toLowerCase(Locale.ROOT));
-        dynamicJaasConfig = configs.get(listenerMechanismKey);
-
-        if (dynamicJaasConfig == null || dynamicJaasConfig.isEmpty()) {
-            String globalMechanismKey =
-                    SERVER_AUTHENTICATOR_PREFIX
-                            + mechanism.toLowerCase(Locale.ROOT)
-                            + "."
-                            + SASL_JAAS_CONFIG;
-            LOG.debug(
-                    "No listener-mechanism JAAS config found for key: '{}'. Falling back to mechanism-level config: '{}'",
-                    listenerMechanismKey,
-                    globalMechanismKey);
-            // 2. Fallback to global mechanism-level config
-            dynamicJaasConfig = configs.get(globalMechanismKey);
-            if (dynamicJaasConfig == null || dynamicJaasConfig.isEmpty()) {
-                LOG.warn(
-                        "No mechanism-level JAAS config found for key: '{}'. Falling back to JVM option: -D{}",
-                        globalMechanismKey,
-                        JaasContext.JAVA_LOGIN_CONFIG_PARAM);
-            }
-        }
-
-        JaasContext jaasContext = JaasContext.loadServerContext(listenerName, dynamicJaasConfig);
-
-        try {
-            LoginManager loginManager = LoginManager.acquireLoginManager(jaasContext);
-            saslServer =
-                    createSaslServer(
-                            mechanism,
-                            address,
-                            configs,
-                            loginManager,
-                            jaasContext.configurationEntries());
-        } catch (Exception e) {
-            throw new RuntimeException(e);
-        }
+        delegate = SaslAuthenticatorFactory.createServerAuthenticator(mechanism, configuration);
+        delegate.initialize(context);
     }
 
     @Override
@@ -123,31 +69,43 @@ public String protocol() {
 
     @Override
     public void matchProtocol(String protocol) {
-        if (!enabledMechanisms.contains(protocol.toUpperCase())) {
+        if (!enabledMechanisms.contains(protocol.toUpperCase(Locale.ROOT))) {
             throw new AuthenticationException(
                     String.format(
                             "SASL server enables %s while protocol of client is '%s'",
                             enabledMechanisms, protocol));
         }
+        if (!SaslAuthenticatorFactory.supportsServerMechanism(protocol)) {
+            throw new AuthenticationException(
+                    "Unable to find a matching SASL mechanism for "
+                            + protocol.toUpperCase(Locale.ROOT));
+        }
     }
 
     @Override
     public byte[] evaluateResponse(byte[] token) throws AuthenticationException {
-        try {
-            return saslServer.evaluateResponse(token);
-        } catch (SaslException e) {
-            throw new AuthenticationException(
-                    String.format("Failed to evaluate SASL response，reason is %s", e.getMessage()));
-        }
+        return delegate.evaluateResponse(token);
     }
 
     @Override
     public boolean isCompleted() {
-        return saslServer != null && saslServer.isComplete();
+        return delegate != null && delegate.isCompleted();
     }
 
     @Override
     public FlussPrincipal createPrincipal() {
-        return new FlussPrincipal(saslServer.getAuthorizationID(), "User");
+        return delegate.createPrincipal();
+    }
+
+    @Override
+    public void close() {
+        if (delegate != null) {
+            try {
+                delegate.close();
+            } catch (Exception e) {
+                LOG.warn("Failed to close SASL server authenticator.", e);
+            }
+            delegate = null;
+        }
     }
 }