From 41973325076d98fd16519c7f208e2f21e44e4abe Mon Sep 17 00:00:00 2001 From: litiliu <38579068+litiliu@users.noreply.github.com> Date: Tue, 4 Aug 2026 17:03:12 +0800 Subject: [PATCH 1/2] [security] Make SASL mechanisms extensible Add a stateless factory that creates one complete authenticator per connection while preserving the existing global SASL mechanism configuration and PLAIN/JAAS behavior. Shared mechanism/session lifecycle is deferred. --- .../auth/sasl/SaslAuthenticatorFactory.java | 80 +++++++++++ .../PlainSaslServerAuthenticator.java | 126 ++++++++++++++++++ .../SaslAuthenticationPlugin.java | 3 +- .../SaslServerAuthenticator.java | 110 +++++---------- .../sasl/SaslAuthenticatorFactoryTest.java | 72 ++++++++++ .../SaslAuthenticationITCase.java | 16 +-- 6 files changed, 322 insertions(+), 85 deletions(-) create mode 100644 fluss-common/src/main/java/org/apache/fluss/security/auth/sasl/SaslAuthenticatorFactory.java create mode 100644 fluss-common/src/main/java/org/apache/fluss/security/auth/sasl/authenticator/PlainSaslServerAuthenticator.java create mode 100644 fluss-common/src/test/java/org/apache/fluss/security/auth/sasl/SaslAuthenticatorFactoryTest.java diff --git a/fluss-common/src/main/java/org/apache/fluss/security/auth/sasl/SaslAuthenticatorFactory.java b/fluss-common/src/main/java/org/apache/fluss/security/auth/sasl/SaslAuthenticatorFactory.java new file mode 100644 index 0000000000..a14fbe0449 --- /dev/null +++ b/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)); + } +} diff --git a/fluss-common/src/main/java/org/apache/fluss/security/auth/sasl/authenticator/PlainSaslServerAuthenticator.java b/fluss-common/src/main/java/org/apache/fluss/security/auth/sasl/authenticator/PlainSaslServerAuthenticator.java new file mode 100644 index 0000000000..c630efd852 --- /dev/null +++ b/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 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; + } +} diff --git a/fluss-common/src/main/java/org/apache/fluss/security/auth/sasl/authenticator/SaslAuthenticationPlugin.java b/fluss-common/src/main/java/org/apache/fluss/security/auth/sasl/authenticator/SaslAuthenticationPlugin.java index 7417fb72ac..f3a2d663e8 100644 --- a/fluss-common/src/main/java/org/apache/fluss/security/auth/sasl/authenticator/SaslAuthenticationPlugin.java +++ b/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 diff --git a/fluss-common/src/main/java/org/apache/fluss/security/auth/sasl/authenticator/SaslServerAuthenticator.java b/fluss-common/src/main/java/org/apache/fluss/security/auth/sasl/authenticator/SaslServerAuthenticator.java index 48231932e9..11be35ac81 100644 --- a/fluss-common/src/main/java/org/apache/fluss/security/auth/sasl/authenticator/SaslServerAuthenticator.java +++ b/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 enabledMechanisms; - private SaslServer saslServer; - private final Map configs; + + private ServerAuthenticator delegate; public SaslServerAuthenticator(Configuration configuration) { - this.configs = configuration.toMap(); + this.configuration = configuration; List 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; + } } } diff --git a/fluss-common/src/test/java/org/apache/fluss/security/auth/sasl/SaslAuthenticatorFactoryTest.java b/fluss-common/src/test/java/org/apache/fluss/security/auth/sasl/SaslAuthenticatorFactoryTest.java new file mode 100644 index 0000000000..40566ae4d1 --- /dev/null +++ b/fluss-common/src/test/java/org/apache/fluss/security/auth/sasl/SaslAuthenticatorFactoryTest.java @@ -0,0 +1,72 @@ +/* + * 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.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.assertj.core.api.Assertions; +import org.junit.jupiter.api.Test; + +/** Tests for {@link SaslAuthenticatorFactory}. */ +class SaslAuthenticatorFactoryTest { + + @Test + void testCreatesIndependentAuthenticators() { + Configuration configuration = new Configuration(); + configuration.set(ConfigOptions.CLIENT_SASL_MECHANISM, "plain"); + + ClientAuthenticator firstClient = + SaslAuthenticatorFactory.createClientAuthenticator(configuration); + ClientAuthenticator secondClient = + SaslAuthenticatorFactory.createClientAuthenticator(configuration); + ServerAuthenticator firstServer = + SaslAuthenticatorFactory.createServerAuthenticator("plain", configuration); + ServerAuthenticator secondServer = + SaslAuthenticatorFactory.createServerAuthenticator("PLAIN", configuration); + + Assertions.assertThat(firstClient).isNotSameAs(secondClient); + Assertions.assertThat(firstServer).isNotSameAs(secondServer); + Assertions.assertThat(firstServer.protocol()).isEqualTo("PLAIN"); + Assertions.assertThatCode(() -> firstServer.matchProtocol("plain")) + .doesNotThrowAnyException(); + Assertions.assertThat(SaslAuthenticatorFactory.supportsServerMechanism("plain")).isTrue(); + } + + @Test + void testRejectsUnsupportedMechanisms() { + Configuration configuration = new Configuration(); + configuration.set(ConfigOptions.CLIENT_SASL_MECHANISM, "unknown"); + + Assertions.assertThatThrownBy( + () -> SaslAuthenticatorFactory.createClientAuthenticator(configuration)) + .isInstanceOf(AuthenticationException.class) + .hasMessage("Unable to find a matching SASL mechanism for UNKNOWN"); + Assertions.assertThatThrownBy( + () -> + SaslAuthenticatorFactory.createServerAuthenticator( + "unknown", configuration)) + .isInstanceOf(AuthenticationException.class) + .hasMessage("Unable to find a matching SASL mechanism for UNKNOWN"); + Assertions.assertThat(SaslAuthenticatorFactory.supportsServerMechanism("unknown")) + .isFalse(); + } +} diff --git a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/authenticate/SaslAuthenticationITCase.java b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/authenticate/SaslAuthenticationITCase.java index b8a2a155b0..7a8fa4ad62 100644 --- a/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/authenticate/SaslAuthenticationITCase.java +++ b/fluss-rpc/src/test/java/org/apache/fluss/rpc/netty/authenticate/SaslAuthenticationITCase.java @@ -94,7 +94,6 @@ void testClientLackMechanism() { clientConfig.setString("client.security.sasl.mechanism", "FAKE"); clientConfig.setString("client.security.sasl.jaas.config", jaasClientInfo); assertThatThrownBy(() -> testAuthentication(clientConfig)) - .cause() .isExactlyInstanceOf(AuthenticationException.class) .hasMessage("Unable to find a matching SASL mechanism for FAKE"); } @@ -105,7 +104,7 @@ void testClientLackLoginModule() { "org.apache.fluss.security.auth.sasl.jaas.FakeLoginModule required username=\"admin\" password=\"wrong-secret\";"; Configuration clientConfig = new Configuration(); clientConfig.setString("client.security.protocol", "sasl"); - clientConfig.setString("client.security.sasl.mechanism", "FAKE"); + clientConfig.setString("client.security.sasl.mechanism", "PLAIN"); clientConfig.setString("client.security.sasl.jaas.config", jaasClientInfo); assertThatThrownBy(() -> testAuthentication(clientConfig)) .isExactlyInstanceOf(AuthenticationException.class) @@ -115,16 +114,17 @@ void testClientLackLoginModule() { @Test void testClientMechanismNotMatchServer() { - String jaasClientInfo = - " org.apache.fluss.security.auth.sasl.jaas.DigestLoginModule required username=\"admin\" password=\"wrong-secret\";"; Configuration clientConfig = new Configuration(); clientConfig.setString("client.security.protocol", "sasl"); - clientConfig.setString("client.security.sasl.mechanism", "DIGEST-MD5"); - clientConfig.setString("client.security.sasl.jaas.config", jaasClientInfo); - assertThatThrownBy(() -> testAuthentication(clientConfig)) + clientConfig.setString("client.security.sasl.mechanism", "PLAIN"); + clientConfig.setString("client.security.sasl.jaas.config", CLIENT_JAAS_INFO); + Configuration serverConfig = getDefaultServerConfig(); + serverConfig.setString("security.sasl.enabled.mechanisms", "FAKE"); + assertThatThrownBy(() -> testAuthentication(clientConfig, serverConfig)) + .cause() .isExactlyInstanceOf(AuthenticationException.class) .hasMessageContaining( - "Only 'org.apache.fluss.security.auth.sasl.plain.PlainLoginModule' is supported in 'client.security.sasl.jaas.config'."); + "SASL server enables [FAKE] while protocol of client is 'PLAIN'"); } @Test From c2be8fbbeb414c5db06f0a01fb246476deff06de Mon Sep 17 00:00:00 2001 From: litiliu <38579068+litiliu@users.noreply.github.com> Date: Thu, 6 Aug 2026 10:24:57 +0800 Subject: [PATCH 2/2] [security] Retrigger CI