ZOOKEEPER-4753: zookeeper-server: Improvement: Explicit handling of DIGEST-MD5 vs GSSAPI in quorum auth

Before this, the SASL-based quorum authorizer did not explicitly
distinguish between the DIGEST-MD5 and GSSAPI mechanisms: it was
simply relying on NameCallback and PasswordCallback for authentication
with the former and examining Kerberos principals in AuthorizeCallback
for the latter.

It turns out that some SASL/DIGEST-MD5 configurations cause
authentication and authorization IDs not to match the expected format,
and the DIGEST-MD5-based portions of the quorum test suite to fail
with obscure errors.  (They can be traced to failures to join the
quorum, but only by looking into detailed logs.)

This patch uses the login module name to determine whether DIGEST-MD5
or GSSAPI is used, and relaxes the authentication ID check for the
former.  As a cleanup, it keeps the password-based credential map
empty when Kerberos principals are expected.  It finally adapts a
test, and adds a new one, ensuring "weirdly-shaped" credentials only
cause authentication failures in the GSSAPI case.

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

@@ -19,6 +19,7 @@
 package org.apache.zookeeper.server.quorum.auth;
 
 import java.io.IOException;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
@@ -31,6 +32,7 @@ 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;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -46,7 +48,8 @@ public class SaslQuorumServerCallbackHandler implements CallbackHandler {
     private static final Logger LOG = LoggerFactory.getLogger(SaslQuorumServerCallbackHandler.class);
 
     private String userName;
-    private final Map<String, String> credentials = new HashMap<>();
+    private final boolean isDigestAuthn;
+    private final Map<String, String> credentials;
     private final Set<String> authzHosts;
 
     public SaslQuorumServerCallbackHandler(
@@ -60,20 +63,35 @@ public class SaslQuorumServerCallbackHandler implements CallbackHandler {
             LOG.error(errorMessage);
             throw new IOException(errorMessage);
         }
-        credentials.clear();
+
+        Map<String, String> credentials = new HashMap<>();
+        boolean isDigestAuthn = true;
+
         for (AppConfigurationEntry entry : configurationEntries) {
-            Map<String, ?> options = entry.getOptions();
-            // Populate DIGEST-MD5 user -> password map with JAAS configuration entries from the "QuorumServer" 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());
+            if (entry.getLoginModuleName().equals(DigestLoginModule.class.getName())) {
+                Map<String, ?> options = entry.getOptions();
+                // Populate DIGEST-MD5 user -> password map with JAAS configuration entries from the "QuorumServer" 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());
+                    }
                 }
+            } else {
+                isDigestAuthn = false;
             }
         }
 
+        this.isDigestAuthn = isDigestAuthn;
+        if (isDigestAuthn) {
+            this.credentials = Collections.unmodifiableMap(credentials);
+            LOG.warn("Using DIGEST-MD5 for quorum authorization");
+        } else {
+            this.credentials = Collections.emptyMap();
+        }
+
         // authorized host lists
         this.authzHosts = authzHosts;
     }
@@ -126,13 +144,15 @@ public class SaslQuorumServerCallbackHandler implements CallbackHandler {
         // 2. Verify whether the connecting host is present in authorized hosts.
         // If not exists, then connecting peer is not authorized to join the
         // ensemble and will reject it.
-        if (authzFlag) {
+        if (!isDigestAuthn && authzFlag) {
             String[] components = authorizationID.split("[/@]");
             if (components.length == 3) {
                 authzFlag = authzHosts.contains(components[1]);
+            } else {
+                authzFlag = false;
             }
             if (!authzFlag) {
-                LOG.error("SASL authorization completed, {} is not authorized to connect", components[1]);
+                LOG.error("SASL authorization completed, {} is not authorized to connect", authorizationID);
             }
         }
 

zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/auth/QuorumKerberosAuthTest.java

@@ -57,7 +57,7 @@ public class QuorumKerberosAuthTest extends KerberosSecurityTestcase {
                                      + "       debug=false\n"
                                      + "       refreshKrb5Config=true\n"
                                      + "       principal=\""
-                                     + KerberosTestUtils.getServerPrincipal()
+                                     + KerberosTestUtils.replaceHostPattern(KerberosTestUtils.getHostServerPrincipal())
                                      + "\";\n"
                                      + "};\n"
                                      + "QuorumLearner {\n"
@@ -71,7 +71,7 @@ public class QuorumKerberosAuthTest extends KerberosSecurityTestcase {
                                      + "       debug=false\n"
                                      + "       refreshKrb5Config=true\n"
                                      + "       principal=\""
-                                     + KerberosTestUtils.getLearnerPrincipal()
+                                     + KerberosTestUtils.replaceHostPattern(KerberosTestUtils.getHostLearnerPrincipal())
                                      + "\";\n"
                                      + "};\n";
         setupJaasConfig(jaasEntries);
@@ -81,10 +81,10 @@ public class QuorumKerberosAuthTest extends KerberosSecurityTestcase {
     public static void setUp() throws Exception {
         // create keytab
         keytabFile = new File(KerberosTestUtils.getKeytabFile());
-        String learnerPrincipal = KerberosTestUtils.getLearnerPrincipal();
-        String serverPrincipal = KerberosTestUtils.getServerPrincipal();
-        learnerPrincipal = learnerPrincipal.substring(0, learnerPrincipal.lastIndexOf("@"));
-        serverPrincipal = serverPrincipal.substring(0, serverPrincipal.lastIndexOf("@"));
+        String learnerPrincipal = KerberosTestUtils.getHostLearnerPrincipal();
+        String serverPrincipal = KerberosTestUtils.getHostServerPrincipal();
+        learnerPrincipal = KerberosTestUtils.replaceHostPattern(learnerPrincipal.substring(0, learnerPrincipal.lastIndexOf("@")));
+        serverPrincipal = KerberosTestUtils.replaceHostPattern(serverPrincipal.substring(0, serverPrincipal.lastIndexOf("@")));
         getKdc().createPrincipal(keytabFile, learnerPrincipal, serverPrincipal);
     }
 

zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/auth/QuorumKerberosHostBasedAuthTest.java

@@ -44,15 +44,17 @@ public class QuorumKerberosHostBasedAuthTest extends KerberosSecurityTestcase {
     private static String hostServerPrincipal = KerberosTestUtils.getHostServerPrincipal();
     private static String hostLearnerPrincipal = KerberosTestUtils.getHostLearnerPrincipal();
     private static String hostNamedLearnerPrincipal = KerberosTestUtils.getHostNamedLearnerPrincipal("myHost");
+    private static String hostlessLearnerPrincipal = KerberosTestUtils.getLearnerPrincipal();
 
     static {
-        setupJaasConfigEntries(hostServerPrincipal, hostLearnerPrincipal, hostNamedLearnerPrincipal);
+        setupJaasConfigEntries(hostServerPrincipal, hostLearnerPrincipal, hostNamedLearnerPrincipal, hostlessLearnerPrincipal);
     }
 
     private static void setupJaasConfigEntries(
         String hostServerPrincipal,
         String hostLearnerPrincipal,
-        String hostNamedLearnerPrincipal) {
+        String hostNamedLearnerPrincipal,
+        String hostlessLearnerPrincipal) {
         String keytabFilePath = FilenameUtils.normalize(KerberosTestUtils.getKeytabFile(), true);
 
         // note: we use "refreshKrb5Config=true" to refresh the kerberos config in the JVM,
@@ -93,6 +95,18 @@ public class QuorumKerberosHostBasedAuthTest extends KerberosSecurityTestcase {
                              + "       refreshKrb5Config=true\n"
                              + "       principal=\"" + hostNamedLearnerPrincipal
                              + "\";\n"
+                             + "};\n"
+                             + "QuorumLearnerMissingHost {\n"
+                             + "       com.sun.security.auth.module.Krb5LoginModule required\n"
+                             + "       useKeyTab=true\n"
+                             + "       keyTab=\"" + keytabFilePath
+                             + "\"\n"
+                             + "       storeKey=true\n"
+                             + "       useTicketCache=false\n"
+                             + "       debug=false\n"
+                             + "       refreshKrb5Config=true\n"
+                             + "       principal=\"" + hostlessLearnerPrincipal
+                             + "\";\n"
                              + "};\n";
         setupJaasConfig(jaasEntries);
     }
@@ -110,7 +124,11 @@ public class QuorumKerberosHostBasedAuthTest extends KerberosSecurityTestcase {
 
         // learner with ipaddress in principal
         String learnerPrincipal2 = hostNamedLearnerPrincipal.substring(0, hostNamedLearnerPrincipal.lastIndexOf("@"));
-        getKdc().createPrincipal(keytabFile, learnerPrincipal, learnerPrincipal2, serverPrincipal);
+
+        // learner without host in principal
+        String learnerPrincipal3 = hostlessLearnerPrincipal.substring(0, hostlessLearnerPrincipal.lastIndexOf("@"));
+
+        getKdc().createPrincipal(keytabFile, learnerPrincipal, learnerPrincipal2, learnerPrincipal3, serverPrincipal);
     }
 
     @AfterEach
@@ -224,4 +242,52 @@ public class QuorumKerberosHostBasedAuthTest extends KerberosSecurityTestcase {
         }
     }
 
+    /**
+     * Test to verify that the bad server connection to the quorum should be rejected.
+     */
+    @Test
+    @Timeout(value = 120)
+    public void testConnectHostlessPrincipalBadServer() throws Exception {
+        String serverPrincipal = hostServerPrincipal.substring(0, hostServerPrincipal.lastIndexOf("@"));
+        Map<String, String> authConfigs = new HashMap<>();
+        authConfigs.put(QuorumAuth.QUORUM_SASL_AUTH_ENABLED, "true");
+        authConfigs.put(QuorumAuth.QUORUM_SERVER_SASL_AUTH_REQUIRED, "true");
+        authConfigs.put(QuorumAuth.QUORUM_LEARNER_SASL_AUTH_REQUIRED, "true");
+        authConfigs.put(QuorumAuth.QUORUM_KERBEROS_SERVICE_PRINCIPAL, serverPrincipal);
+        String connectStr = startQuorum(3, authConfigs, 3);
+        CountdownWatcher watcher = new CountdownWatcher();
+        ZooKeeper zk = new ZooKeeper(connectStr, ClientBase.CONNECTION_TIMEOUT, watcher);
+        watcher.waitForConnected(ClientBase.CONNECTION_TIMEOUT);
+        for (int i = 0; i < 10; i++) {
+            zk.create("/" + i, new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+        }
+        zk.close();
+
+        String quorumCfgSection = mt.get(0).getQuorumCfgSection();
+        StringBuilder sb = new StringBuilder();
+        sb.append(quorumCfgSection);
+
+        int myid = mt.size() + 1;
+        final int clientPort = PortAssignment.unique();
+        String server = String.format("server.%d=localhost:%d:%d:participant", myid, PortAssignment.unique(), PortAssignment.unique());
+        sb.append(server + "\n");
+        quorumCfgSection = sb.toString();
+        authConfigs.put(QuorumAuth.QUORUM_LEARNER_SASL_LOGIN_CONTEXT, "QuorumLearnerMissingHost");
+        MainThread badServer = new MainThread(myid, clientPort, quorumCfgSection, authConfigs);
+        badServer.start();
+        watcher = new CountdownWatcher();
+        connectStr = "127.0.0.1:" + clientPort;
+        zk = new ZooKeeper(connectStr, ClientBase.CONNECTION_TIMEOUT, watcher);
+        try {
+            watcher.waitForConnected(ClientBase.CONNECTION_TIMEOUT / 3);
+            fail("Must throw exception as the principal does not include an authorized host!");
+        } catch (TimeoutException e) {
+            // expected
+        } finally {
+            zk.close();
+            badServer.shutdown();
+            badServer.deleteBaseDir();
+        }
+    }
+
 }