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

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

@@ -28,6 +28,7 @@
 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 void run() {
         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 @@ private synchronized LoginContext login(final String loginContextName) throws Lo
                                      + ") 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.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 @@ private SaslClient createSaslClient(
         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.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 void configure(InetSocketAddress addr, int maxcc, int backlog) throws IOE
 
     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 @@ protected void configureSaslLogin() throws IOException {
 
         // 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 @@ protected void configureSaslLogin() throws IOException {
         }
     }
 
+    /**
+     * 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.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 SaslQuorumAuthLearner(
                     "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.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 SaslQuorumAuthServer(boolean quorumRequireSasl, String loginContext, Set<
                     "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 void authenticate(Socket sock, DataInputStream din) throws SaslException
                 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.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 @@ private static class TestableKerberosLogin extends Login {
     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