ZOOKEEPER-4839: Fix SASL DIGEST-MD5 authenticated with last successfully logined username

Reviewers: kezhuw, kezhuw, kezhuw, anmolnar
Author: luoxiner
Closes #2176 from luoxiner/master

zookeeper-server/src/main/java/org/apache/zookeeper/Login.java

@@ -28,6 +28,7 @@ package org.apache.zookeeper;
 import java.util.Date;
 import java.util.Set;
 import java.util.concurrent.ThreadLocalRandom;
+import java.util.function.Supplier;
 import javax.security.auth.Subject;
 import javax.security.auth.callback.CallbackHandler;
 import javax.security.auth.kerberos.KerberosPrincipal;
@@ -48,7 +49,7 @@ public class Login {
     private static final String KINIT_COMMAND_DEFAULT = "/usr/bin/kinit";
     private static final Logger LOG = LoggerFactory.getLogger(Login.class);
     public static final String SYSTEM_USER = System.getProperty("user.name", "<NA>");
-    public CallbackHandler callbackHandler;
+    private final Supplier<CallbackHandler> callbackHandlerSupplier;
 
     // LoginThread will sleep until 80% of time from last refresh to
     // ticket's expiry has been reached, at which time it will wake
@@ -89,17 +90,17 @@ public class Login {
      *            name of section in JAAS file that will be use to login. Passed
      *            as first param to javax.security.auth.login.LoginContext().
      *
-     * @param callbackHandler
-     *            Passed as second param to
-     *            javax.security.auth.login.LoginContext().
+     * @param callbackHandlerSupplier
+     *            Per connection callbackhandler supplier.
+     *
      * @param zkConfig
      *            client or server configurations
      * @throws javax.security.auth.login.LoginException
      *             Thrown if authentication fails.
      */
-    public Login(final String loginContextName, CallbackHandler callbackHandler, final ZKConfig zkConfig) throws LoginException {
+    public Login(final String loginContextName, Supplier<CallbackHandler> callbackHandlerSupplier, final ZKConfig zkConfig) throws LoginException {
         this.zkConfig = zkConfig;
-        this.callbackHandler = callbackHandler;
+        this.callbackHandlerSupplier = callbackHandlerSupplier;
         login = login(loginContextName);
         this.loginContextName = loginContextName;
         subject = login.getSubject();
@@ -274,6 +275,17 @@ public class Login {
         t.setDaemon(true);
     }
 
+    /**
+     * Return a new CallbackHandler for connections
+     * to avoid race conditions and state sharing in
+     * connection login processing.
+     *
+     * @return connection dependent CallbackHandler
+     */
+    public CallbackHandler newCallbackHandler() {
+        return callbackHandlerSupplier.get();
+    }
+
     public void startThreadIfNeeded() {
         // thread object 't' will be null if a refresh thread is not needed.
         if (t != null) {
@@ -315,7 +327,7 @@ public class Login {
                                      + ") and your "
                                      + getLoginContextMessage());
         }
-        LoginContext loginContext = new LoginContext(loginContextName, callbackHandler);
+        LoginContext loginContext = new LoginContext(loginContextName, newCallbackHandler());
         loginContext.login();
         LOG.info("{} successfully logged in.", loginContextName);
         return loginContext;

zookeeper-server/src/main/java/org/apache/zookeeper/client/ZooKeeperSaslClient.java

@@ -22,7 +22,9 @@ import java.io.IOException;
 import java.security.PrivilegedActionException;
 import java.security.PrivilegedExceptionAction;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
 import javax.security.auth.Subject;
+import javax.security.auth.callback.CallbackHandler;
 import javax.security.auth.login.AppConfigurationEntry;
 import javax.security.auth.login.Configuration;
 import javax.security.auth.login.LoginException;
@@ -240,9 +242,10 @@ public class ZooKeeperSaslClient {
         try {
             if (loginRef.get() == null) {
                 LOG.debug("JAAS loginContext is: {}", loginContext);
-                // note that the login object is static: it's shared amongst all zookeeper-related connections.
-                // in order to ensure the login is initialized only once, it must be synchronized the code snippet.
-                Login l = new Login(loginContext, new SaslClientCallbackHandler(null, "Client"), clientConfig);
+                Supplier<CallbackHandler> callbackHandlerSupplier = () -> {
+                    return new SaslClientCallbackHandler(null, "Client");
+                };
+                Login l = new Login(loginContext, callbackHandlerSupplier, clientConfig);
                 if (loginRef.compareAndSet(null, l)) {
                     l.startThreadIfNeeded();
                 }

zookeeper-server/src/main/java/org/apache/zookeeper/server/ServerCnxnFactory.java

@@ -22,10 +22,13 @@ import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.nio.ByteBuffer;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Supplier;
 import javax.management.JMException;
+import javax.security.auth.callback.CallbackHandler;
 import javax.security.auth.login.AppConfigurationEntry;
 import javax.security.auth.login.Configuration;
 import javax.security.auth.login.LoginException;
@@ -41,6 +44,7 @@ public abstract class ServerCnxnFactory {
 
     public static final String ZOOKEEPER_SERVER_CNXN_FACTORY = "zookeeper.serverCnxnFactory";
     private static final String ZOOKEEPER_MAX_CONNECTION = "zookeeper.maxCnxns";
+    private static final String DIGEST_MD5_USER_PREFIX = "user_";
     public static final int ZOOKEEPER_MAX_CONNECTION_DEFAULT = 0;
 
     private static final Logger LOG = LoggerFactory.getLogger(ServerCnxnFactory.class);
@@ -113,7 +117,6 @@ public abstract class ServerCnxnFactory {
 
     public abstract void reconfigure(InetSocketAddress addr);
 
-    protected SaslServerCallbackHandler saslServerCallbackHandler;
     public Login login;
 
     /** Maximum number of connections allowed from particular host (ip) */
@@ -269,8 +272,11 @@ public abstract class ServerCnxnFactory {
 
         // jaas.conf entry available
         try {
-            saslServerCallbackHandler = new SaslServerCallbackHandler(Configuration.getConfiguration());
-            login = new Login(serverSection, saslServerCallbackHandler, new ZKConfig());
+            Map<String, String> credentials = getDigestMd5Credentials(entries);
+            Supplier<CallbackHandler> callbackHandlerSupplier = () -> {
+                return new SaslServerCallbackHandler(credentials);
+            };
+            login = new Login(serverSection, callbackHandlerSupplier, new ZKConfig());
             setLoginUser(login.getUserName());
             login.startThreadIfNeeded();
         } catch (LoginException e) {
@@ -280,6 +286,28 @@ public abstract class ServerCnxnFactory {
         }
     }
 
+    /**
+     * make server credentials map from configuration's server section.
+     * @param appConfigurationEntries AppConfigurationEntry List
+     * @return Server credentials map
+     */
+    private static Map<String, String> getDigestMd5Credentials(final  AppConfigurationEntry[] appConfigurationEntries) {
+        Map<String, String> credentials = new HashMap<>();
+        for (AppConfigurationEntry entry : appConfigurationEntries) {
+            Map<String, ?> options = entry.getOptions();
+            // Populate DIGEST-MD5 user -> password map with JAAS configuration entries from the "Server" section.
+            // Usernames are distinguished from other options by prefixing the username with a "user_" prefix.
+            for (Map.Entry<String, ?> pair : options.entrySet()) {
+                String key = pair.getKey();
+                if (key.startsWith(DIGEST_MD5_USER_PREFIX)) {
+                    String userName = key.substring(DIGEST_MD5_USER_PREFIX.length());
+                    credentials.put(userName, (String) pair.getValue());
+                }
+            }
+        }
+        return credentials;
+    }
+
     private static void setLoginUser(String name) {
         //Created this method to avoid ST_WRITE_TO_STATIC_FROM_INSTANCE_METHOD find bug issue
         loginUser = name;

zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperSaslServer.java

@@ -41,7 +41,7 @@ public class ZooKeeperSaslServer {
     private SaslServer createSaslServer(final Login login) {
         synchronized (login) {
             Subject subject = login.getSubject();
-            return SecurityUtils.createSaslServer(subject, "zookeeper", "zk-sasl-md5", login.callbackHandler, LOG);
+            return SecurityUtils.createSaslServer(subject, "zookeeper", "zk-sasl-md5", login.newCallbackHandler(), LOG);
         }
     }
 

zookeeper-server/src/main/java/org/apache/zookeeper/server/auth/SaslServerCallbackHandler.java

@@ -19,57 +19,29 @@
 package org.apache.zookeeper.server.auth;
 
 import java.io.IOException;
-import java.util.HashMap;
 import java.util.Map;
 import javax.security.auth.callback.Callback;
 import javax.security.auth.callback.CallbackHandler;
 import javax.security.auth.callback.NameCallback;
 import javax.security.auth.callback.PasswordCallback;
 import javax.security.auth.callback.UnsupportedCallbackException;
-import javax.security.auth.login.AppConfigurationEntry;
-import javax.security.auth.login.Configuration;
 import javax.security.sasl.AuthorizeCallback;
 import javax.security.sasl.RealmCallback;
-import org.apache.zookeeper.server.ZooKeeperSaslServer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 public class SaslServerCallbackHandler implements CallbackHandler {
 
-    private static final String USER_PREFIX = "user_";
     private static final Logger LOG = LoggerFactory.getLogger(SaslServerCallbackHandler.class);
     private static final String SYSPROP_SUPER_PASSWORD = "zookeeper.SASLAuthenticationProvider.superPassword";
     private static final String SYSPROP_REMOVE_HOST = "zookeeper.kerberos.removeHostFromPrincipal";
     private static final String SYSPROP_REMOVE_REALM = "zookeeper.kerberos.removeRealmFromPrincipal";
 
     private String userName;
-    private final Map<String, String> credentials = new HashMap<>();
+    private final Map<String, String> credentials;
 
-    public SaslServerCallbackHandler(Configuration configuration) throws IOException {
-        String serverSection = System.getProperty(
-            ZooKeeperSaslServer.LOGIN_CONTEXT_NAME_KEY,
-            ZooKeeperSaslServer.DEFAULT_LOGIN_CONTEXT_NAME);
-
-        AppConfigurationEntry[] configurationEntries = configuration.getAppConfigurationEntry(serverSection);
-
-        if (configurationEntries == null) {
-            String errorMessage = "Could not find a '" + serverSection + "' entry in this configuration: Server cannot start.";
-            LOG.error(errorMessage);
-            throw new IOException(errorMessage);
-        }
-        credentials.clear();
-        for (AppConfigurationEntry entry : configurationEntries) {
-            Map<String, ?> options = entry.getOptions();
-            // Populate DIGEST-MD5 user -> password map with JAAS configuration entries from the "Server" section.
-            // Usernames are distinguished from other options by prefixing the username with a "user_" prefix.
-            for (Map.Entry<String, ?> pair : options.entrySet()) {
-                String key = pair.getKey();
-                if (key.startsWith(USER_PREFIX)) {
-                    String userName = key.substring(USER_PREFIX.length());
-                    credentials.put(userName, (String) pair.getValue());
-                }
-            }
-        }
+    public SaslServerCallbackHandler(Map<String, String> credentials) {
+        this.credentials = credentials;
     }
 
     public void handle(Callback[] callbacks) throws UnsupportedCallbackException {

zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/auth/SaslQuorumAuthLearner.java

@@ -25,7 +25,9 @@ import java.io.IOException;
 import java.net.Socket;
 import java.security.PrivilegedActionException;
 import java.security.PrivilegedExceptionAction;
+import java.util.function.Supplier;
 import javax.security.auth.Subject;
+import javax.security.auth.callback.CallbackHandler;
 import javax.security.auth.login.AppConfigurationEntry;
 import javax.security.auth.login.Configuration;
 import javax.security.auth.login.LoginException;
@@ -62,9 +64,13 @@ public class SaslQuorumAuthLearner implements QuorumAuthLearner {
                     "SASL-authentication failed because the specified JAAS configuration section '%s' could not be found.",
                     loginContext));
             }
+
+            Supplier<CallbackHandler> callbackSupplier = () -> {
+                return new SaslClientCallbackHandler(null, "QuorumLearner");
+            };
             this.learnerLogin = new Login(
                 loginContext,
-                new SaslClientCallbackHandler(null, "QuorumLearner"),
+                callbackSupplier,
                 new ZKConfig());
             this.learnerLogin.startThreadIfNeeded();
         } catch (LoginException e) {

zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/auth/SaslQuorumAuthServer.java

@@ -24,6 +24,8 @@ import java.io.DataOutputStream;
 import java.io.IOException;
 import java.net.Socket;
 import java.util.Set;
+import java.util.function.Supplier;
+import javax.security.auth.callback.CallbackHandler;
 import javax.security.auth.login.AppConfigurationEntry;
 import javax.security.auth.login.Configuration;
 import javax.security.auth.login.LoginException;
@@ -55,9 +57,10 @@ public class SaslQuorumAuthServer implements QuorumAuthServer {
                     "SASL-authentication failed because the specified JAAS configuration section '%s' could not be found.",
                     loginContext));
             }
-            SaslQuorumServerCallbackHandler saslServerCallbackHandler = new SaslQuorumServerCallbackHandler(
-                Configuration.getConfiguration(), loginContext, authzHosts);
-            serverLogin = new Login(loginContext, saslServerCallbackHandler, new ZKConfig());
+            Supplier<CallbackHandler> callbackSupplier = () -> {
+                return new SaslQuorumServerCallbackHandler(entries, authzHosts);
+            };
+            serverLogin = new Login(loginContext, callbackSupplier, new ZKConfig());
             serverLogin.startThreadIfNeeded();
         } catch (Throwable e) {
             throw new SaslException("Failed to initialize authentication mechanism using SASL", e);
@@ -86,7 +89,7 @@ public class SaslQuorumAuthServer implements QuorumAuthServer {
                 serverLogin.getSubject(),
                 QuorumAuth.QUORUM_SERVER_PROTOCOL_NAME,
                 QuorumAuth.QUORUM_SERVER_SASL_DIGEST,
-                serverLogin.callbackHandler,
+                serverLogin.newCallbackHandler(),
                 LOG);
             while (!ss.isComplete()) {
                 challenge = ss.evaluateResponse(token);

zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/auth/SaslQuorumServerCallbackHandler.java

@@ -18,7 +18,6 @@
 
 package org.apache.zookeeper.server.quorum.auth;
 
-import java.io.IOException;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
@@ -29,7 +28,6 @@ import javax.security.auth.callback.NameCallback;
 import javax.security.auth.callback.PasswordCallback;
 import javax.security.auth.callback.UnsupportedCallbackException;
 import javax.security.auth.login.AppConfigurationEntry;
-import javax.security.auth.login.Configuration;
 import javax.security.sasl.AuthorizeCallback;
 import javax.security.sasl.RealmCallback;
 import org.apache.zookeeper.server.auth.DigestLoginModule;
@@ -53,16 +51,8 @@ public class SaslQuorumServerCallbackHandler implements CallbackHandler {
     private final Set<String> authzHosts;
 
     public SaslQuorumServerCallbackHandler(
-        Configuration configuration,
-        String serverSection,
-        Set<String> authzHosts) throws IOException {
-        AppConfigurationEntry[] configurationEntries = configuration.getAppConfigurationEntry(serverSection);
-
-        if (configurationEntries == null) {
-            String errorMessage = "Could not find a '" + serverSection + "' entry in this configuration: Server cannot start.";
-            LOG.error(errorMessage);
-            throw new IOException(errorMessage);
-        }
+        AppConfigurationEntry[] configurationEntries,
+        Set<String> authzHosts) {
 
         Map<String, String> credentials = new HashMap<>();
         boolean isDigestAuthn = true;

zookeeper-server/src/test/java/org/apache/zookeeper/KerberosTicketRenewalTest.java

@@ -127,7 +127,9 @@ public class KerberosTicketRenewalTest {
     private CountDownLatch continueRefreshThread = new CountDownLatch(1);
 
     public TestableKerberosLogin() throws LoginException {
-      super(JAAS_CONFIG_SECTION, (callbacks) -> {}, new ZKConfig());
+      super(JAAS_CONFIG_SECTION, () -> {
+        return (callbacks) -> {};
+      }, new ZKConfig());
     }
 
     @Override

zookeeper-server/src/test/java/org/apache/zookeeper/test/SaslAuthRequiredMultiClientTest.java

@@ -0,0 +1,88 @@
+/*
+ * 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.zookeeper.test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.fail;
+import javax.security.auth.login.Configuration;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.ZooDefs.Ids;
+import org.apache.zookeeper.ZooKeeper;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+public class SaslAuthRequiredMultiClientTest extends ClientBase {
+
+    @BeforeAll
+    public static void setUpBeforeClass() {
+        System.setProperty(SaslTestUtil.requireSASLAuthProperty, "true");
+        System.setProperty(SaslTestUtil.authProviderProperty, SaslTestUtil.authProvider);
+        System.setProperty(SaslTestUtil.jaasConfig, SaslTestUtil.createJAASConfigFile("jaas.conf", "test"));
+    }
+
+    @AfterAll
+    public static void tearDownAfterClass() {
+        System.clearProperty(SaslTestUtil.requireSASLAuthProperty);
+        System.clearProperty(SaslTestUtil.authProviderProperty);
+        System.clearProperty(SaslTestUtil.jaasConfig);
+    }
+
+    @Test
+    public void testClientOpWithInvalidSASLUserAuthAfterSuccessLogin() throws Exception {
+        resetJaasConfiguration("jaas.conf", "super", "test");
+        try  (ZooKeeper zk = createClient()) {
+            zk.create("/foobar", null, Ids.CREATOR_ALL_ACL, CreateMode.PERSISTENT);
+        } catch (KeeperException e) {
+            fail("Client operation should succeed with valid SASL configuration.");
+        }
+
+        resetJaasConfiguration("jaas.conf", "super_wrong", "test");
+        try  (ZooKeeper wrongUserZk = createClient()) {
+            wrongUserZk.create("/bar", null, Ids.CREATOR_ALL_ACL, CreateMode.PERSISTENT);
+            fail("Client with wrong SASL config should not pass SASL authentication.");
+        } catch (KeeperException e) {
+            assertEquals(KeeperException.Code.AUTHFAILED, e.code());
+        }
+    }
+
+    @Test
+    public void testClientOpWithInvalidSASLPasswordAuthAfterSuccessLogin() throws Exception {
+        resetJaasConfiguration("jaas.conf", "super", "test");
+        try (ZooKeeper zk = createClient()) {
+            zk.create("/foobar", null, Ids.CREATOR_ALL_ACL, CreateMode.PERSISTENT);
+        } catch (KeeperException e) {
+            fail("Client operation should succeed with valid SASL configuration.");
+        }
+
+        resetJaasConfiguration("jaas.conf", "super", "test_wrongong");
+        try (ZooKeeper wrongPasswordZk = createClient()) {
+            wrongPasswordZk.create("/bar", null, Ids.CREATOR_ALL_ACL, CreateMode.PERSISTENT);
+            fail("Client with wrong SASL config should not pass SASL authentication.");
+        } catch (KeeperException e) {
+            assertEquals(KeeperException.Code.AUTHFAILED, e.code());
+        }
+    }
+
+    protected static void resetJaasConfiguration(String fileName, String userName, String password) {
+        Configuration.setConfiguration(null);
+        System.setProperty(SaslTestUtil.jaasConfig, SaslTestUtil.createJAASConfigFile(fileName, userName, password));
+    }
+}

zookeeper-server/src/test/java/org/apache/zookeeper/test/SaslTestUtil.java

@@ -28,6 +28,7 @@ public class SaslTestUtil extends ClientBase {
     // The maximum time (in milliseconds) a client should take to observe
     // a disconnect event of the same client from server.
     static Integer CLIENT_DISCONNECT_TIMEOUT = 3000;
+    static String SUPER_USER_NAME = "super";
     static String requireSASLAuthProperty = "zookeeper.sessionRequireClientSASLAuth";
     static String authProviderProperty = "zookeeper.authProvider.1";
     static String authProvider = "org.apache.zookeeper.server.auth.SASLAuthenticationProvider";
@@ -35,6 +36,10 @@ public class SaslTestUtil extends ClientBase {
     static String jaasConfig = "java.security.auth.login.config";
 
     static String createJAASConfigFile(String fileName, String password) {
+        return createJAASConfigFile(fileName, SUPER_USER_NAME, password);
+    }
+
+    static String createJAASConfigFile(String fileName, String userName, String password) {
         String ret = null;
         try {
             File tmpDir = createTmpDir();
@@ -47,7 +52,7 @@ public class SaslTestUtil extends ClientBase {
                     + "};\n"
                     + "Client {\n"
                     + "       " + digestLoginModule + " required\n"
-                    + "       username=\"super\"\n"
+                    + "       username=\"" + userName + "\"\n"
                     + "       password=\"" + password + "\";\n"
                     + "};" + "\n");
             fwriter.close();