ZOOKEEPER-3959: Add support for multiple SASL-authenticated super users

This implements a simple variant of the "multiple SASL-authenticated super users" functionality discussed on ticket ZOOKEEPER-3959.

With this patch, properties matching `zookeeper.superUser.*` are considered, in addition to `zookeeper.superUser`, to determine whether an authenticated ID should be given `super` privileges.

As before, `super` privileges remain a sharp tool and should be avoided where possible--but where necessary, allowing multiple IDs makes audit logs more useful.

(Many thanks to eolivelli and symat for their comments and suggestions.)

Author: Damien Diederen <dd@crosstwine.com>

Reviewers: Enrico Olivelli <eolivelli@apache.org>

Closes #1503 from ztzg/ZOOKEEPER-3959-multiple-superusers

zookeeper-docs/src/main/resources/markdown/zookeeperAdmin.md

@@ -1501,6 +1501,9 @@ and [SASL authentication for ZooKeeper](https://cwiki.apache.org/confluence/disp
     Similar to **zookeeper.X509AuthenticationProvider.superUser**
     but is generic for SASL based logins. It stores the name of
     a user that can access the znode hierarchy as a "super" user.
+    You can specify multiple SASL super users using the
+    **zookeeper.superUser.[suffix]** notation, e.g.:
+    `zookeeper.superUser.1=...`.
 
 * *ssl.authProvider* :
     (Java system property: **zookeeper.ssl.authProvider**)

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

@@ -32,6 +32,7 @@ import java.util.Deque;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Properties;
 import java.util.Random;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -106,6 +107,8 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
 
     static final boolean skipACL;
 
+    public static final String SASL_SUPER_USER = "zookeeper.superUser";
+
     public static final String ALLOW_SASL_FAILED_CLIENTS = "zookeeper.allowSaslFailedClients";
     public static final String ZOOKEEPER_DIGEST_ENABLED = "zookeeper.digest.enabled";
     private static boolean digestEnabled;
@@ -1644,6 +1647,28 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
         }
     }
 
+    private static boolean isSaslSuperUser(String id) {
+        if (id == null || id.isEmpty()) {
+            return false;
+        }
+
+        Properties properties = System.getProperties();
+        int prefixLen = SASL_SUPER_USER.length();
+
+        for (String k : properties.stringPropertyNames()) {
+            if (k.startsWith(SASL_SUPER_USER)
+                && (k.length() == prefixLen || k.charAt(prefixLen) == '.')) {
+                String value = properties.getProperty(k);
+
+                if (value != null && value.equals(id)) {
+                    return true;
+                }
+            }
+        }
+
+        return false;
+    }
+
     private static boolean shouldAllowSaslFailedClientsConnect() {
         return Boolean.getBoolean(ALLOW_SASL_FAILED_CLIENTS);
     }
@@ -1667,9 +1692,13 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
                     LOG.info("Session 0x{}: adding SASL authorization for authorizationID: {}",
                             Long.toHexString(cnxn.getSessionId()), authorizationID);
                     cnxn.addAuthInfo(new Id("sasl", authorizationID));
-                    if (System.getProperty("zookeeper.superUser") != null
-                        && authorizationID.equals(System.getProperty("zookeeper.superUser"))) {
+
+                    if (isSaslSuperUser(authorizationID)) {
                         cnxn.addAuthInfo(new Id("super", ""));
+                        LOG.info(
+                            "Session 0x{}: Authenticated Id '{}' as super user",
+                            Long.toHexString(cnxn.getSessionId()),
+                            authorizationID);
                     }
                 }
             } catch (SaslException e) {

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

@@ -30,8 +30,10 @@ import org.apache.zookeeper.WatchedEvent;
 import org.apache.zookeeper.Watcher.Event.KeeperState;
 import org.apache.zookeeper.ZooDefs.Perms;
 import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.client.ZKClientConfig;
 import org.apache.zookeeper.data.ACL;
 import org.apache.zookeeper.data.Id;
+import org.apache.zookeeper.server.ZooKeeperServer;
 import org.apache.zookeeper.server.auth.DigestAuthenticationProvider;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
@@ -42,12 +44,14 @@ public class SaslSuperUserTest extends ClientBase {
     private static Id otherSaslUser = new Id("sasl", "joe");
     private static Id otherDigestUser;
     private static String oldAuthProvider;
+    private static String oldClientConfigSection;
     private static String oldLoginConfig;
     private static String oldSuperUser;
 
     @BeforeAll
     public static void setupStatic() throws Exception {
         oldAuthProvider = System.setProperty("zookeeper.authProvider.1", "org.apache.zookeeper.server.auth.SASLAuthenticationProvider");
+        oldClientConfigSection = System.getProperty(ZKClientConfig.LOGIN_CONTEXT_NAME_KEY);
 
         File tmpDir = createTmpDir();
         File saslConfFile = new File(tmpDir, "jaas.conf");
@@ -56,42 +60,40 @@ public class SaslSuperUserTest extends ClientBase {
         fwriter.write(""
                               + "Server {\n"
                               + "          org.apache.zookeeper.server.auth.DigestLoginModule required\n"
-                              + "          user_super_duper=\"test\";\n"
+                              + "          user_super_duper=\"test\"\n"
+                              + "          user_other_super=\"test\";\n"
                               + "};\n"
                               + "Client {\n"
                               + "       org.apache.zookeeper.server.auth.DigestLoginModule required\n"
                               + "       username=\"super_duper\"\n"
                               + "       password=\"test\";\n"
                               + "};"
+                              + "OtherClient {\n"
+                              + "       org.apache.zookeeper.server.auth.DigestLoginModule required\n"
+                              + "       username=\"other_super\"\n"
+                              + "       password=\"test\";\n"
+                              + "};"
                               + "\n");
         fwriter.close();
         oldLoginConfig = System.setProperty("java.security.auth.login.config", saslConfFile.getAbsolutePath());
-        oldSuperUser = System.setProperty("zookeeper.superUser", "super_duper");
+        oldSuperUser = System.setProperty(ZooKeeperServer.SASL_SUPER_USER, "super_duper");
         otherDigestUser = new Id("digest", DigestAuthenticationProvider.generateDigest("jack:jack"));
     }
 
     @AfterAll
     public static void cleanupStatic() {
-        if (oldAuthProvider != null) {
-            System.setProperty("zookeeper.authProvider.1", oldAuthProvider);
-        } else {
-            System.clearProperty("zookeeper.authProvider.1");
-        }
-        oldAuthProvider = null;
-
-        if (oldLoginConfig != null) {
-            System.setProperty("java.security.auth.login.config", oldLoginConfig);
-        } else {
-            System.clearProperty("java.security.auth.login.config");
-        }
-        oldLoginConfig = null;
+        restoreProperty("zookeeper.authProvider.1", oldAuthProvider);
+        restoreProperty(ZKClientConfig.LOGIN_CONTEXT_NAME_KEY, oldClientConfigSection);
+        restoreProperty("java.security.auth.login.config", oldLoginConfig);
+        restoreProperty(ZooKeeperServer.SASL_SUPER_USER, oldSuperUser);
+    }
 
-        if (oldSuperUser != null) {
-            System.setProperty("zookeeper.superUser", oldSuperUser);
+    private static void restoreProperty(String property, String oldValue) {
+        if (oldValue != null) {
+            System.setProperty(property, oldValue);
         } else {
-            System.clearProperty("zookeeper.superUser");
+            System.clearProperty(property);
         }
-        oldSuperUser = null;
     }
 
     private AtomicInteger authFailed = new AtomicInteger(0);
@@ -115,8 +117,7 @@ public class SaslSuperUserTest extends ClientBase {
 
     }
 
-    @Test
-    public void testSuperIsSuper() throws Exception {
+    private void connectAndPerformSuperOps() throws Exception {
         ZooKeeper zk = createClient();
         try {
             zk.create("/digest_read", null, Arrays.asList(new ACL(Perms.READ, otherDigestUser)), CreateMode.PERSISTENT);
@@ -127,11 +128,35 @@ public class SaslSuperUserTest extends ClientBase {
             zk.delete("/digest_read", -1);
             zk.delete("/sasl_read/sub", -1);
             zk.delete("/sasl_read", -1);
-            //If the test failes it will most likely fail with a NoAuth exception before it ever gets to this assertion
-            assertEquals(authFailed.get(), 0);
         } finally {
             zk.close();
         }
     }
 
+    @Test
+    public void testSuperIsSuper() throws Exception {
+        connectAndPerformSuperOps();
+        //If the test fails it will most likely fail with a NoAuth exception before it ever gets to this assertion
+        assertEquals(authFailed.get(), 0);
+    }
+
+    @Test
+    public void testOtherSuperIsSuper() throws Exception {
+        String prevSection = System.setProperty(ZKClientConfig.LOGIN_CONTEXT_NAME_KEY, "OtherClient");
+
+        // KLUDGE: We do this quite late, as the server has been
+        // started at this point--but the implementation currently
+        // looks at the properties each time a SASL negotiation completes.
+        String superUser1Prop = ZooKeeperServer.SASL_SUPER_USER + ".1";
+        String prevSuperUser1 = System.setProperty(superUser1Prop, "other_super");
+
+        try {
+            connectAndPerformSuperOps();
+            //If the test fails it will most likely fail with a NoAuth exception before it ever gets to this assertion
+            assertEquals(authFailed.get(), 0);
+        } finally {
+            restoreProperty(superUser1Prop, prevSuperUser1);
+            restoreProperty(ZKClientConfig.LOGIN_CONTEXT_NAME_KEY, prevSection);
+        }
+    }
 }