KAFKA-18627:add allowed modules to JaasUtils

clients/src/main/java/org/apache/kafka/common/security/JaasContext.java

@@ -33,8 +33,9 @@
 import javax.security.auth.login.AppConfigurationEntry;
 import javax.security.auth.login.Configuration;
 
+import static org.apache.kafka.common.security.JaasUtils.ALLOWED_LOGIN_MODULES_CONFIG;
+import static org.apache.kafka.common.security.JaasUtils.ALLOWED_LOGIN_MODULES_DEFAULT;
 import static org.apache.kafka.common.security.JaasUtils.DISALLOWED_LOGIN_MODULES_CONFIG;
-import static org.apache.kafka.common.security.JaasUtils.DISALLOWED_LOGIN_MODULES_DEFAULT;
 
 public class JaasContext {
 
@@ -103,15 +104,41 @@ else if (contextModules.length != 1)
             return defaultContext(contextType, listenerContextName, globalContextName);
     }
 
+    @SuppressWarnings("deprecation")
     private static void throwIfLoginModuleIsNotAllowed(AppConfigurationEntry appConfigurationEntry) {
-        Set<String> disallowedLoginModuleList = Arrays.stream(
-                System.getProperty(DISALLOWED_LOGIN_MODULES_CONFIG, DISALLOWED_LOGIN_MODULES_DEFAULT).split(","))
+        String disallowedProperty = System.getProperty(DISALLOWED_LOGIN_MODULES_CONFIG);
+        if (disallowedProperty != null) {
+            LOG.warn("System property '{}' is deprecated and will be removed in a future release. Use '{}' instead.",
+                    DISALLOWED_LOGIN_MODULES_CONFIG, ALLOWED_LOGIN_MODULES_CONFIG);
+        }
+        String loginModuleName = appConfigurationEntry.getLoginModuleName().trim();
+        String allowedProperty = System.getProperty(ALLOWED_LOGIN_MODULES_CONFIG);
+        if (allowedProperty != null) {
+            Set<String> allowedLoginModuleList = Arrays.stream(allowedProperty.split(","))
+                    .map(String::trim)
+                    .collect(Collectors.toSet());
+            if (!allowedLoginModuleList.contains(loginModuleName)) {
+                throw new IllegalArgumentException(loginModuleName + " is not allowed. Update System property '"
+                        + ALLOWED_LOGIN_MODULES_CONFIG + "' to allow " + loginModuleName);
+            }
+            return;
+        }
+        if (disallowedProperty != null) {
+            Set<String> disallowedLoginModuleList = Arrays.stream(disallowedProperty.split(","))
+                    .map(String::trim)
+                    .collect(Collectors.toSet());
+            if (disallowedLoginModuleList.contains(loginModuleName)) {
+                throw new IllegalArgumentException(loginModuleName + " is not allowed. Update System property '"
+                        + DISALLOWED_LOGIN_MODULES_CONFIG + "' to allow " + loginModuleName);
+            }
+            return;
+        }
+        Set<String> defaultAllowedLoginModuleList = Arrays.stream(ALLOWED_LOGIN_MODULES_DEFAULT.split(","))
                 .map(String::trim)
                 .collect(Collectors.toSet());
-        String loginModuleName = appConfigurationEntry.getLoginModuleName().trim();
-        if (disallowedLoginModuleList.contains(loginModuleName)) {
+        if (!defaultAllowedLoginModuleList.contains(loginModuleName)) {
             throw new IllegalArgumentException(loginModuleName + " is not allowed. Update System property '"
-                    + DISALLOWED_LOGIN_MODULES_CONFIG + "' to allow " + loginModuleName);
+                    + ALLOWED_LOGIN_MODULES_CONFIG + "' to allow " + loginModuleName);
         }
     }
 

clients/src/main/java/org/apache/kafka/common/security/JaasUtils.java

@@ -16,13 +16,32 @@
  */
 package org.apache.kafka.common.security;
 
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
 public final class JaasUtils {
     public static final String JAVA_LOGIN_CONFIG_PARAM = "java.security.auth.login.config";
+    @Deprecated
     public static final String DISALLOWED_LOGIN_MODULES_CONFIG = "org.apache.kafka.disallowed.login.modules";
-    public static final String DISALLOWED_LOGIN_MODULES_DEFAULT =
-            "com.sun.security.auth.module.JndiLoginModule,com.sun.security.auth.module.LdapLoginModule";
+    public static final String ALLOWED_LOGIN_MODULES_CONFIG = "org.apache.kafka.allowed.login.modules";
+    public static final String ALLOWED_LOGIN_MODULES_DEFAULT = String.join(",", List.of(
+            "org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule",
+            "org.apache.kafka.common.security.plain.PlainLoginModule",
+            "org.apache.kafka.connect.rest.basic.auth.extension.PropertyFileLoginModule",
+            "org.apache.kafka.common.security.scram.ScramLoginModule",
+            "com.sun.security.auth.module.Krb5LoginModule"
+    ));
     public static final String SERVICE_NAME = "serviceName";
 
-    private JaasUtils() {}
+    private JaasUtils() {
+    }
+
+    public static void allowDefaultJaasAndCustomJass(String... customJaas) {
+        List<String> jaasModules = new ArrayList<>();
+        jaasModules.add(ALLOWED_LOGIN_MODULES_DEFAULT);
+        jaasModules.addAll(Arrays.asList(customJaas));
+        System.setProperty(org.apache.kafka.common.security.JaasUtils.ALLOWED_LOGIN_MODULES_CONFIG, String.join(",", jaasModules));
+    }
 
 }

clients/src/test/java/org/apache/kafka/common/security/JaasContextTest.java

@@ -39,7 +39,9 @@
 import javax.security.auth.login.AppConfigurationEntry.LoginModuleControlFlag;
 import javax.security.auth.login.Configuration;
 
+import static org.apache.kafka.common.security.JaasUtils.ALLOWED_LOGIN_MODULES_CONFIG;
 import static org.apache.kafka.common.security.JaasUtils.DISALLOWED_LOGIN_MODULES_CONFIG;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -68,11 +70,13 @@ public void tearDown() throws Exception {
 
     @Test
     public void testConfigNoOptions() throws Exception {
+        System.setProperty(ALLOWED_LOGIN_MODULES_CONFIG, "test.testConfigNoOptions");
         checkConfiguration("test.testConfigNoOptions", LoginModuleControlFlag.REQUIRED, new HashMap<>());
     }
 
     @Test
     public void testControlFlag() throws Exception {
+        System.setProperty(ALLOWED_LOGIN_MODULES_CONFIG, "test.testControlFlag");
         LoginModuleControlFlag[] controlFlags = new LoginModuleControlFlag[] {
             LoginModuleControlFlag.REQUIRED,
             LoginModuleControlFlag.REQUISITE,
@@ -88,13 +92,15 @@ public void testControlFlag() throws Exception {
 
     @Test
     public void testSingleOption() throws Exception {
+        System.setProperty(ALLOWED_LOGIN_MODULES_CONFIG, "test.testSingleOption");
         Map<String, Object> options = new HashMap<>();
         options.put("propName", "propValue");
         checkConfiguration("test.testSingleOption", LoginModuleControlFlag.REQUISITE, options);
     }
 
     @Test
     public void testMultipleOptions() throws Exception {
+        System.setProperty(ALLOWED_LOGIN_MODULES_CONFIG, "test.testMultipleOptions");
         Map<String, Object> options = new HashMap<>();
         for (int i = 0; i < 10; i++)
             options.put("propName" + i, "propValue" + i);
@@ -103,6 +109,7 @@ public void testMultipleOptions() throws Exception {
 
     @Test
     public void testQuotedOptionValue() throws Exception {
+        System.setProperty(ALLOWED_LOGIN_MODULES_CONFIG, "test.testQuotedOptionValue");
         Map<String, Object> options = new HashMap<>();
         options.put("propName", "prop value");
         options.put("propName2", "value1 = 1, value2 = 2");
@@ -112,6 +119,7 @@ public void testQuotedOptionValue() throws Exception {
 
     @Test
     public void testQuotedOptionName() throws Exception {
+        System.setProperty(ALLOWED_LOGIN_MODULES_CONFIG, "test.testQuotedOptionName");
         Map<String, Object> options = new HashMap<>();
         options.put("prop name", "propValue");
         String config = "test.testQuotedOptionName required \"prop name\"=propValue;";
@@ -224,8 +232,8 @@ public void testDisallowedLoginModulesSystemProperty() throws Exception {
                 "SOME-MECHANISM", Collections.emptyMap()));
 
 
-        //Remove default value for org.apache.kafka.disallowed.login.modules
-        System.setProperty(DISALLOWED_LOGIN_MODULES_CONFIG, "");
+        // add allowed login modules
+        System.setProperty(ALLOWED_LOGIN_MODULES_CONFIG, "com.sun.security.auth.module.JndiLoginModule, com.sun.security.auth.module.LdapLoginModule");
 
         checkConfiguration("com.sun.security.auth.module.JndiLoginModule", LoginModuleControlFlag.REQUIRED, new HashMap<>());
         checkConfiguration("com.sun.security.auth.module.LdapLoginModule", LoginModuleControlFlag.REQUIRED, new HashMap<>());
@@ -252,9 +260,34 @@ public void testDisallowedLoginModulesSystemProperty() throws Exception {
         checkEntry(context.configurationEntries().get(0), "com.sun.security.auth.module.LdapLoginModule",
                 LoginModuleControlFlag.REQUISITE, Collections.emptyMap());
     }
+
+    @Test
+     void testAllowedLoginModulesSystemProperty() {
+
+        //  default
+        String jaasConfigProp1 = "com.ibm.security.auth.module.LdapLoginModule required;";
+        assertThrows(IllegalArgumentException.class, () ->  configurationEntry(JaasContext.Type.CLIENT, jaasConfigProp1));
+
+        String jaasConfigProp2 = "com.sun.security.auth.module.JndiLoginModule required;";
+        //  set allow dont' set not allow
+        System.setProperty(JaasUtils.ALLOWED_LOGIN_MODULES_CONFIG, "com.ibm.security.auth.module.LdapLoginModule");
+        assertDoesNotThrow(() ->  configurationEntry(JaasContext.Type.CLIENT, jaasConfigProp1));
+        assertThrows(IllegalArgumentException.class, () ->  configurationEntry(JaasContext.Type.CLIENT, jaasConfigProp2));
+
+        //  set allow and set not allow
+        System.setProperty(JaasUtils.DISALLOWED_LOGIN_MODULES_CONFIG, "com.ibm.security.auth.module.LdapLoginModule");
+        assertDoesNotThrow(() ->  configurationEntry(JaasContext.Type.CLIENT, jaasConfigProp1));
+        assertThrows(IllegalArgumentException.class, () ->  configurationEntry(JaasContext.Type.CLIENT, jaasConfigProp2));
+
+        //  don't set allow and set not allow
+        System.clearProperty(JaasUtils.ALLOWED_LOGIN_MODULES_CONFIG);
+        assertThrows(IllegalArgumentException.class, () ->  configurationEntry(JaasContext.Type.CLIENT, jaasConfigProp1));
+        assertDoesNotThrow(() ->  configurationEntry(JaasContext.Type.CLIENT, jaasConfigProp2));
+    }
 
     @Test
     public void testNumericOptionWithQuotes() throws Exception {
+        System.setProperty(ALLOWED_LOGIN_MODULES_CONFIG, "test.testNumericOptionWithQuotes");
         Map<String, Object> options = new HashMap<>();
         options.put("option1", "3");
         String config = "test.testNumericOptionWithQuotes required option1=\"3\";";
@@ -263,6 +296,7 @@ public void testNumericOptionWithQuotes() throws Exception {
 
     @Test
     public void testLoadForServerWithListenerNameOverride() throws IOException {
+        System.setProperty(ALLOWED_LOGIN_MODULES_CONFIG, "test.LoginModuleDefault, test.LoginModuleOverride");
         writeConfiguration(Arrays.asList(
             "KafkaServer { test.LoginModuleDefault required; };",
             "plaintext.KafkaServer { test.LoginModuleOverride requisite; };"
@@ -278,6 +312,7 @@ public void testLoadForServerWithListenerNameOverride() throws IOException {
 
     @Test
     public void testLoadForServerWithListenerNameAndFallback() throws IOException {
+        System.setProperty(ALLOWED_LOGIN_MODULES_CONFIG, "test.LoginModule");
         writeConfiguration(Arrays.asList(
             "KafkaServer { test.LoginModule required; };",
             "other.KafkaServer { test.LoginModuleOther requisite; };"

clients/src/test/java/org/apache/kafka/common/security/authenticator/LoginManagerTest.java

@@ -20,6 +20,7 @@
 import org.apache.kafka.common.config.types.Password;
 import org.apache.kafka.common.network.ListenerName;
 import org.apache.kafka.common.security.JaasContext;
+import org.apache.kafka.common.security.JaasUtils;
 import org.apache.kafka.common.security.auth.AuthenticateCallbackHandler;
 import org.apache.kafka.common.security.auth.Login;
 import org.apache.kafka.common.security.plain.PlainLoginModule;
@@ -59,6 +60,7 @@ public void setUp() {
                 " required user=\"digestuser\" password=\"digest-secret\";");
         TestJaasConfig.createConfiguration("SCRAM-SHA-256",
                 Collections.singletonList("SCRAM-SHA-256"));
+        JaasUtils.allowDefaultJaasAndCustomJass("org.apache.kafka.common.security.authenticator.TestDigestLoginModule");
     }
 
     @AfterEach

clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java

@@ -66,6 +66,7 @@
 import org.apache.kafka.common.requests.SaslHandshakeRequest;
 import org.apache.kafka.common.requests.SaslHandshakeResponse;
 import org.apache.kafka.common.security.JaasContext;
+import org.apache.kafka.common.security.JaasUtils;
 import org.apache.kafka.common.security.TestSecurityConfig;
 import org.apache.kafka.common.security.auth.AuthenticateCallbackHandler;
 import org.apache.kafka.common.security.auth.AuthenticationContext;
@@ -174,6 +175,8 @@ public void setup() throws Exception {
         saslClientConfigs = clientCertStores.getTrustingConfig(serverCertStores);
         credentialCache = new CredentialCache();
         TestLogin.loginCount.set(0);
+        JaasUtils.allowDefaultJaasAndCustomJass("org.apache.kafka.common.security.authenticator.TestDigestLoginModule",
+                "org.apache.kafka.common.security.authenticator.SaslAuthenticatorTest$TestPlainLoginModule");
     }
 
     @AfterEach
@@ -1053,12 +1056,7 @@ public void testInvalidLoginModule() throws Exception {
 
         SecurityProtocol securityProtocol = SecurityProtocol.SASL_SSL;
         server = createEchoServer(securityProtocol);
-        try {
-            createSelector(securityProtocol, saslClientConfigs);
-            fail("SASL/PLAIN channel created without valid login module");
-        } catch (KafkaException e) {
-            // Expected exception
-        }
+        assertThrows(IllegalArgumentException.class, () -> createSelector(securityProtocol, saslClientConfigs));
     }
 
     /**

core/src/test/scala/integration/kafka/api/SaslEndToEndAuthorizationTest.scala

@@ -46,7 +46,6 @@ abstract class SaslEndToEndAuthorizationTest extends EndToEndAuthorizationTest {
     producerConfig.put(SaslConfigs.SASL_JAAS_CONFIG, clientLoginContext)
     consumerConfig.put(SaslConfigs.SASL_JAAS_CONFIG, clientLoginContext)
     adminClientConfig.put(SaslConfigs.SASL_JAAS_CONFIG, clientLoginContext)
-
     val superuserLoginContext = jaasAdminLoginModule(kafkaClientSaslMechanism)
     superuserClientConfig.put(SaslConfigs.SASL_JAAS_CONFIG, superuserLoginContext)
     super.setUp(testInfo)

core/src/test/scala/integration/kafka/api/SaslSslAdminIntegrationTest.scala

@@ -70,7 +70,6 @@ class SaslSslAdminIntegrationTest extends BaseAdminIntegrationTest with SaslSetu
     this.serverConfig.setProperty(DelegationTokenManagerConfigs.DELEGATION_TOKEN_SECRET_KEY_CONFIG, secretKey)
     this.serverConfig.setProperty(DelegationTokenManagerConfigs.DELEGATION_TOKEN_EXPIRY_TIME_MS_CONFIG, Long.MaxValue.toString)
     this.serverConfig.setProperty(DelegationTokenManagerConfigs.DELEGATION_TOKEN_MAX_LIFETIME_CONFIG, Long.MaxValue.toString)
-
     setUpSasl()
     super.setUp(testInfo)
     setInitialAcls()