ZOOKEEPER-1525: Plumb ZooKeeperServer object into auth plugins

Based on patch work from https://issues.apache.org/jira/browse/ZOOKEEPER-1525

Created ServerAuthenticationProvider which has a method to accept the ZooKeeper
server so that auth can be done using values in the ZK database. As this is a new
interface, existing implementations aren't affected helping backward compatibility

Author: randgalt <jordan@jordanzimmerman.com>

Reviewers: fpj <fpj@apache.org>, Allan Lyu <lvfangmin@gmail.com>

Closes #84 from Randgalt/ZOOKEEPER-1525

src/docs/src/documentation/content/xdocs/zookeeperProgrammers.xml

@@ -1193,6 +1193,54 @@ authProvider.2=com.f.MyAuth2
     only one will be used. Also all servers must have the same plugins defined, otherwise clients using
     the authentication schemes provided by the plugins will have problems connecting to some servers.
     </para>
+
+    <para> <emphasis role="bold">Added in 3.6.0</emphasis>: An alternate abstraction is available for pluggable
+    authentication. It provides additional arguments.
+    </para>
+
+    <programlisting>
+public abstract class ServerAuthenticationProvider implements AuthenticationProvider {
+    public abstract KeeperException.Code handleAuthentication(ServerObjs serverObjs, byte authData[]);
+    public abstract boolean matches(ServerObjs serverObjs, MatchValues matchValues);
+}
+    </programlisting>
+
+    <para>
+    Instead of implementing AuthenticationProvider you extend ServerAuthenticationProvider. Your handleAuthentication()
+    and matches() methods will then receive the additional parameters (via ServerObjs and MatchValues).
+    </para>
+
+    <itemizedlist>
+      <listitem>
+        <para><emphasis role="bold">ZooKeeperServer</emphasis></para>
+
+        <para>The ZooKeeperServer instance</para>
+      </listitem>
+
+      <listitem>
+        <para><emphasis role="bold">ServerCnxn</emphasis></para>
+
+        <para>The current connection</para>
+      </listitem>
+
+      <listitem>
+        <para><emphasis role="bold">path</emphasis></para>
+
+        <para>The ZNode path being operated on (or null if not used)</para>
+      </listitem>
+
+      <listitem>
+        <para><emphasis role="bold">perm</emphasis></para>
+
+        <para>The operation value or 0</para>
+      </listitem>
+
+      <listitem>
+        <para><emphasis role="bold">setAcls</emphasis></para>
+
+        <para>When the setAcl() method is being operated on, the list of ACLs that are being set</para>
+      </listitem>
+    </itemizedlist>
   </section>
       
   <section id="ch_zkGuarantees">

src/java/main/org/apache/zookeeper/server/FinalRequestProcessor.java

@@ -319,9 +319,9 @@ public class FinalRequestProcessor implements RequestProcessor {
                 if (n == null) {
                     throw new KeeperException.NoNodeException();
                 }
-                PrepRequestProcessor.checkACL(zks, zks.getZKDatabase().aclForNode(n),
+                PrepRequestProcessor.checkACL(zks, request.cnxn, zks.getZKDatabase().aclForNode(n),
                         ZooDefs.Perms.READ,
-                        request.authInfo);
+                        request.authInfo, getDataRequest.getPath(), null);
                 Stat stat = new Stat();
                 byte b[] = zks.getZKDatabase().getData(getDataRequest.getPath(), stat,
                         getDataRequest.getWatch() ? cnxn : null);
@@ -361,9 +361,9 @@ public class FinalRequestProcessor implements RequestProcessor {
                 if (n == null) {
                     throw new KeeperException.NoNodeException();
                 }
-                PrepRequestProcessor.checkACL(zks, zks.getZKDatabase().aclForNode(n),
+                PrepRequestProcessor.checkACL(zks, request.cnxn, zks.getZKDatabase().aclForNode(n),
                         ZooDefs.Perms.READ,
-                        request.authInfo);
+                        request.authInfo, getChildrenRequest.getPath(), null);
                 List<String> children = zks.getZKDatabase().getChildren(
                         getChildrenRequest.getPath(), null, getChildrenRequest
                                 .getWatch() ? cnxn : null);
@@ -380,9 +380,9 @@ public class FinalRequestProcessor implements RequestProcessor {
                 if (n == null) {
                     throw new KeeperException.NoNodeException();
                 }
-                PrepRequestProcessor.checkACL(zks, zks.getZKDatabase().aclForNode(n),
+                PrepRequestProcessor.checkACL(zks, request.cnxn, zks.getZKDatabase().aclForNode(n),
                         ZooDefs.Perms.READ,
-                        request.authInfo);
+                        request.authInfo, getChildren2Request.getPath(), null);
                 List<String> children = zks.getZKDatabase().getChildren(
                         getChildren2Request.getPath(), stat, getChildren2Request
                                 .getWatch() ? cnxn : null);

src/java/main/org/apache/zookeeper/server/NIOServerCnxn.java

@@ -112,7 +112,7 @@ public class NIOServerCnxn extends ServerCnxn {
         sock.socket().setSoLinger(false, -1);
         InetAddress addr = ((InetSocketAddress) sock.socket()
                 .getRemoteSocketAddress()).getAddress();
-        authInfo.add(new Id("ip", addr.getHostAddress()));
+        addAuthInfo(new Id("ip", addr.getHostAddress()));
         this.sessionTimeout = factory.sessionlessCnxnTimeout;
     }
 

src/java/main/org/apache/zookeeper/server/PrepRequestProcessor.java

@@ -44,6 +44,7 @@ import org.apache.zookeeper.proto.SetDataRequest;
 import org.apache.zookeeper.server.ZooKeeperServer.ChangeRecord;
 import org.apache.zookeeper.server.auth.AuthenticationProvider;
 import org.apache.zookeeper.server.auth.ProviderRegistry;
+import org.apache.zookeeper.server.auth.ServerAuthenticationProvider;
 import org.apache.zookeeper.server.quorum.Leader.XidRolloverException;
 import org.apache.zookeeper.server.quorum.LeaderZooKeeperServer;
 import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer;
@@ -285,14 +286,16 @@ public class PrepRequestProcessor extends ZooKeeperCriticalThread implements
 
     /**
      * Grant or deny authorization to an operation on a node as a function of:
-     *
-     * @param zks: not used.
-     * @param acl:  set of ACLs for the node
-     * @param perm: the permission that the client is requesting
-     * @param ids:  the credentials supplied by the client
+     * @param zks :     the ZooKeeper server
+     * @param cnxn :    the server connection
+     * @param acl :     set of ACLs for the node
+     * @param perm :    the permission that the client is requesting
+     * @param ids :     the credentials supplied by the client
+     * @param path :    the ZNode path
+     * @param setAcls : for set ACL operations, the list of ACLs being set. Otherwise null.
      */
-    static void checkACL(ZooKeeperServer zks, List<ACL> acl, int perm,
-            List<Id> ids) throws KeeperException.NoAuthException {
+    static void checkACL(ZooKeeperServer zks, ServerCnxn cnxn, List<ACL> acl, int perm, List<Id> ids,
+                         String path, List<ACL> setAcls) throws KeeperException.NoAuthException {
         if (skipACL) {
             return;
         }
@@ -316,12 +319,13 @@ public class PrepRequestProcessor extends ZooKeeperCriticalThread implements
                         && id.getId().equals("anyone")) {
                     return;
                 }
-                AuthenticationProvider ap = ProviderRegistry.getProvider(id
+                ServerAuthenticationProvider ap = ProviderRegistry.getServerProvider(id
                         .getScheme());
                 if (ap != null) {
                     for (Id authId : ids) {
                         if (authId.getScheme().equals(id.getScheme())
-                                && ap.matches(authId.getId(), id.getId())) {
+                                && ap.matches(new ServerAuthenticationProvider.ServerObjs(zks, cnxn),
+                                new ServerAuthenticationProvider.MatchValues(path, authId.getId(), id.getId(), perm, setAcls))) {
                             return;
                         }
                     }
@@ -398,7 +402,7 @@ public class PrepRequestProcessor extends ZooKeeperCriticalThread implements
                 String parentPath = getParentPathAndValidate(path);
                 ChangeRecord parentRecord = getRecordForPath(parentPath);
                 ChangeRecord nodeRecord = getRecordForPath(path);
-                checkACL(zks, parentRecord.acl, ZooDefs.Perms.DELETE, request.authInfo);
+                checkACL(zks, request.cnxn, parentRecord.acl, ZooDefs.Perms.DELETE, request.authInfo, path, null);
                 checkAndIncVersion(nodeRecord.stat.getVersion(), deleteRequest.getVersion(), path);
                 if (nodeRecord.childCount > 0) {
                     throw new KeeperException.NotEmptyException(path);
@@ -417,7 +421,7 @@ public class PrepRequestProcessor extends ZooKeeperCriticalThread implements
                 path = setDataRequest.getPath();
                 validatePath(path, request.sessionId);
                 nodeRecord = getRecordForPath(path);
-                checkACL(zks, nodeRecord.acl, ZooDefs.Perms.WRITE, request.authInfo);
+                checkACL(zks, request.cnxn, nodeRecord.acl, ZooDefs.Perms.WRITE, request.authInfo, path, null);
                 int newVersion = checkAndIncVersion(nodeRecord.stat.getVersion(), setDataRequest.getVersion(), path);
                 request.setTxn(new SetDataTxn(path, setDataRequest.getData(), newVersion));
                 nodeRecord = nodeRecord.duplicate(request.getHdr().getZxid());
@@ -552,7 +556,7 @@ public class PrepRequestProcessor extends ZooKeeperCriticalThread implements
                 }
                 
                 nodeRecord = getRecordForPath(ZooDefs.CONFIG_NODE);               
-                checkACL(zks, nodeRecord.acl, ZooDefs.Perms.WRITE, request.authInfo);                  
+                checkACL(zks, request.cnxn, nodeRecord.acl, ZooDefs.Perms.WRITE, request.authInfo, null, null);
                 request.setTxn(new SetDataTxn(ZooDefs.CONFIG_NODE, request.qv.toString().getBytes(), -1));    
                 nodeRecord = nodeRecord.duplicate(request.getHdr().getZxid());
                 nodeRecord.stat.setVersion(-1);                
@@ -567,7 +571,7 @@ public class PrepRequestProcessor extends ZooKeeperCriticalThread implements
                 validatePath(path, request.sessionId);
                 List<ACL> listACL = fixupACL(path, request.authInfo, setAclRequest.getAcl());
                 nodeRecord = getRecordForPath(path);
-                checkACL(zks, nodeRecord.acl, ZooDefs.Perms.ADMIN, request.authInfo);
+                checkACL(zks, request.cnxn, nodeRecord.acl, ZooDefs.Perms.ADMIN, request.authInfo, path, listACL);
                 newVersion = checkAndIncVersion(nodeRecord.stat.getAversion(), setAclRequest.getVersion(), path);
                 request.setTxn(new SetACLTxn(path, listACL, newVersion));
                 nodeRecord = nodeRecord.duplicate(request.getHdr().getZxid());
@@ -622,7 +626,7 @@ public class PrepRequestProcessor extends ZooKeeperCriticalThread implements
                 path = checkVersionRequest.getPath();
                 validatePath(path, request.sessionId);
                 nodeRecord = getRecordForPath(path);
-                checkACL(zks, nodeRecord.acl, ZooDefs.Perms.READ, request.authInfo);
+                checkACL(zks, request.cnxn, nodeRecord.acl, ZooDefs.Perms.READ, request.authInfo, path, null);
                 request.setTxn(new CheckVersionTxn(path, checkAndIncVersion(nodeRecord.stat.getVersion(),
                         checkVersionRequest.getVersion(), path)));
                 break;
@@ -664,7 +668,7 @@ public class PrepRequestProcessor extends ZooKeeperCriticalThread implements
         List<ACL> listACL = fixupACL(path, request.authInfo, acl);
         ChangeRecord parentRecord = getRecordForPath(parentPath);
 
-        checkACL(zks, parentRecord.acl, ZooDefs.Perms.CREATE, request.authInfo);
+        checkACL(zks, request.cnxn, parentRecord.acl, ZooDefs.Perms.CREATE, request.authInfo, path, listACL);
         int parentCVersion = parentRecord.stat.getCversion();
         if (createMode.isSequential()) {
             path = path + String.format(Locale.ENGLISH, "%010d", parentCVersion);
@@ -973,8 +977,8 @@ public class PrepRequestProcessor extends ZooKeeperCriticalThread implements
                 // authenticated ids of the requestor
                 boolean authIdValid = false;
                 for (Id cid : authInfo) {
-                    AuthenticationProvider ap =
-                        ProviderRegistry.getProvider(cid.getScheme());
+                    ServerAuthenticationProvider ap =
+                        ProviderRegistry.getServerProvider(cid.getScheme());
                     if (ap == null) {
                         LOG.error("Missing AuthenticationProvider for "
                             + cid.getScheme());
@@ -987,7 +991,7 @@ public class PrepRequestProcessor extends ZooKeeperCriticalThread implements
                     throw new KeeperException.InvalidACLException(path);
                 }
             } else {
-                AuthenticationProvider ap = ProviderRegistry.getProvider(id.getScheme());
+                ServerAuthenticationProvider ap = ProviderRegistry.getServerProvider(id.getScheme());
                 if (ap == null || !ap.isValid(id.getId())) {
                     throw new KeeperException.InvalidACLException(path);
                 }

src/java/main/org/apache/zookeeper/server/ServerCnxn.java

@@ -24,12 +24,8 @@ import java.io.StringWriter;
 import java.net.InetSocketAddress;
 import java.nio.ByteBuffer;
 import java.security.cert.Certificate;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.Date;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicLong;
 
 import org.apache.jute.Record;
@@ -51,7 +47,7 @@ public abstract class ServerCnxn implements Stats, Watcher {
     final public static Object me = new Object();
     private static final Logger LOG = LoggerFactory.getLogger(ServerCnxn.class);
     
-    protected ArrayList<Id> authInfo = new ArrayList<Id>();
+    private Set<Id> authInfo = Collections.newSetFromMap(new ConcurrentHashMap<Id, Boolean>());
 
     /**
      * If the client is of old version, we don't send r-o mode info to it.
@@ -78,13 +74,11 @@ public abstract class ServerCnxn implements Stats, Watcher {
 
     /** auth info for the cnxn, returns an unmodifyable list */
     public List<Id> getAuthInfo() {
-        return Collections.unmodifiableList(authInfo);
+        return Collections.unmodifiableList(new ArrayList<>(authInfo));
     }
 
     public void addAuthInfo(Id id) {
-        if (authInfo.contains(id) == false) {
-            authInfo.add(id);
-        }
+        authInfo.add(id);
     }
 
     public boolean removeAuthInfo(Id id) {

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

@@ -63,6 +63,7 @@ import org.apache.zookeeper.server.SessionTracker.Session;
 import org.apache.zookeeper.server.SessionTracker.SessionExpirer;
 import org.apache.zookeeper.server.auth.AuthenticationProvider;
 import org.apache.zookeeper.server.auth.ProviderRegistry;
+import org.apache.zookeeper.server.auth.ServerAuthenticationProvider;
 import org.apache.zookeeper.server.persistence.FileTxnSnapLog;
 import org.apache.zookeeper.server.quorum.ReadOnlyZooKeeperServer;
 import org.apache.zookeeper.txn.CreateSessionTxn;
@@ -1034,11 +1035,11 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
             AuthPacket authPacket = new AuthPacket();
             ByteBufferInputStream.byteBuffer2Record(incomingBuffer, authPacket);
             String scheme = authPacket.getScheme();
-            AuthenticationProvider ap = ProviderRegistry.getProvider(scheme);
+            ServerAuthenticationProvider ap = ProviderRegistry.getServerProvider(scheme);
             Code authReturn = KeeperException.Code.AUTHFAILED;
             if(ap != null) {
                 try {
-                    authReturn = ap.handleAuthentication(cnxn, authPacket.getAuth());
+                    authReturn = ap.handleAuthentication(new ServerAuthenticationProvider.ServerObjs(this, cnxn), authPacket.getAuth());
                 } catch(RuntimeException e) {
                     LOG.warn("Caught runtime exception from AuthenticationProvider: " + scheme + " due to " + e);
                     authReturn = KeeperException.Code.AUTHFAILED;

src/java/main/org/apache/zookeeper/server/auth/KeyAuthenticationProvider.java

@@ -0,0 +1,140 @@
+/**
+ * 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.server.auth;
+
+import java.io.UnsupportedEncodingException;
+
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.KeeperException.NoNodeException;
+import org.apache.zookeeper.data.Id;
+import org.apache.zookeeper.server.ZooKeeperServer;
+import org.apache.zookeeper.server.ZKDatabase;
+import org.apache.zookeeper.data.Stat;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/*
+ * This class is a sample implementation of being passed the ZooKeeperServer
+ * handle in the constructor, and reading data from zknodes to authenticate.
+ * At a minimum, a real Auth provider would need to override validate() to
+ * e.g. perform certificate validation of auth based a public key.
+ *
+ * See the "Pluggable ZooKeeper authentication" section of the 
+ * "Zookeeper Programmer's Guide" for general details of implementing an
+ * authentication plugin. e.g.
+ * http://zookeeper.apache.org/doc/trunk/zookeeperProgrammers.html#sc_ZooKeeperPluggableAuthentication
+ *
+ * This class looks for a numeric "key" under the /key node.
+ * Authorizaton is granted if the user passes in as authorization a number
+ * which is a multiple of the key value, i.e. 
+ *   (auth % key) == 0
+ * In a real implementation, you might do something like storing a public
+ * key in /key, and using it to verify that auth tokens passed in were signed
+ * by the corresponding private key.
+ *
+ * When the node /key does not exist, any auth token is accepted, so that 
+ * bootstrapping may occur.
+ *
+ */
+public class KeyAuthenticationProvider extends ServerAuthenticationProvider {
+    private static final Logger LOG = LoggerFactory.getLogger(KeyAuthenticationProvider.class);
+
+    public String getScheme() {
+        return "key";
+    }
+
+    private byte[] getKey(ZooKeeperServer zks) {
+        ZKDatabase db = zks.getZKDatabase();
+        if (db != null) {
+            try {
+                Stat stat = new Stat();
+                return db.getData("/key", stat, null);
+            } catch (NoNodeException e) {
+                LOG.error("getData failed", e);
+            }
+        }
+        return null;
+    }
+
+    private boolean validate(byte[] key, byte[] auth) {
+        // perform arbitrary function (auth is a multiple of key)
+        try {
+            String keyStr = new String(key, "UTF-8");
+            String authStr = new String(auth, "UTF-8");
+            int keyVal = Integer.parseInt(keyStr);
+            int authVal = Integer.parseInt(authStr);
+            if (keyVal!=0 && ((authVal % keyVal) != 0)) {
+              return false;
+            }
+        } catch (NumberFormatException | UnsupportedEncodingException nfe) {
+            LOG.error("bad formatting", nfe);
+          return false;
+        }
+        return true;
+    }
+
+    @Override
+    public KeeperException.Code handleAuthentication(ServerObjs serverObjs, byte[] authData) {
+        byte[] key = getKey(serverObjs.getZks());
+        String authStr = "";
+        String keyStr = "";
+        try {
+          authStr = new String(authData, "UTF-8");
+        } catch (Exception e) {
+            LOG.error("UTF-8", e);
+        }
+        if (key != null) {
+            if (!validate(key, authData)) {
+                try {
+                  keyStr = new String(key, "UTF-8");
+                } catch (Exception e) {
+                    LOG.error("UTF-8", e);
+                    // empty key
+                    keyStr = authStr;
+                }
+                LOG.debug("KeyAuthenticationProvider handleAuthentication ("+keyStr+", "+authStr+") -> FAIL.\n");
+                return KeeperException.Code.AUTHFAILED;
+            }
+        }
+        // default to allow, so the key can be initially written
+        LOG.debug("KeyAuthenticationProvider handleAuthentication -> OK.\n");
+        // NOTE: keyStr in addAuthInfo() sticks with the created node ACLs.
+        //   For transient keys or certificates, this presents a problem.
+        //   In that case, replace it with something non-ephemeral (or punt with null).
+        //
+        // BOTH addAuthInfo and an OK return-code are needed for authentication.
+        serverObjs.getCnxn().addAuthInfo(new Id(getScheme(), keyStr));
+        return KeeperException.Code.OK;
+    }
+
+    @Override
+    public boolean matches(ServerObjs serverObjs, MatchValues matchValues) {
+        return matchValues.getId().equals(matchValues.getAclExpr());
+    }
+
+    @Override
+    public boolean isAuthenticated() {
+        return true;
+    }
+
+    @Override
+    public boolean isValid(String id) {
+        return true;
+    }
+}

src/java/main/org/apache/zookeeper/server/auth/ProviderRegistry.java

@@ -31,7 +31,15 @@ public class ProviderRegistry {
 
     private static boolean initialized = false;
     private static HashMap<String, AuthenticationProvider> authenticationProviders =
-        new HashMap<String, AuthenticationProvider>();
+        new HashMap<>();
+
+    //VisibleForTesting
+    public static void reset() {
+        synchronized (ProviderRegistry.class) {
+            initialized = false;
+            authenticationProviders.clear();
+        }
+    }
 
     public static void initialize() {
         synchronized (ProviderRegistry.class) {
@@ -61,6 +69,10 @@ public class ProviderRegistry {
         }
     }
 
+    public static ServerAuthenticationProvider getServerProvider(String scheme) {
+        return WrappedAuthenticationProvider.wrap(getProvider(scheme));
+    }
+
     public static AuthenticationProvider getProvider(String scheme) {
         if(!initialized)
             initialize();
@@ -70,8 +82,8 @@ public class ProviderRegistry {
     public static String listProviders() {
         StringBuilder sb = new StringBuilder();
         for(String s: authenticationProviders.keySet()) {
-        sb.append(s + " ");
-}
+            sb.append(s).append(" ");
+        }
         return sb.toString();
     }
 }

src/java/main/org/apache/zookeeper/server/auth/ServerAuthenticationProvider.java

@@ -0,0 +1,140 @@
+/**
+ * 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.server.auth;
+
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.data.ACL;
+import org.apache.zookeeper.server.ServerCnxn;
+import org.apache.zookeeper.server.ZooKeeperServer;
+
+import java.util.List;
+
+/**
+ * A variation on {@link AuthenticationProvider} that provides additional
+ * parameters for more detailed authentication
+ */
+public abstract class ServerAuthenticationProvider implements AuthenticationProvider {
+    public static class ServerObjs {
+        private final ZooKeeperServer zks;
+        private final ServerCnxn cnxn;
+
+        /**
+         * @param zks
+         *                the ZooKeeper server instance
+         * @param cnxn
+         *                the cnxn that received the authentication information.
+         */
+        public ServerObjs(ZooKeeperServer zks, ServerCnxn cnxn) {
+            this.zks = zks;
+            this.cnxn = cnxn;
+        }
+
+        public ZooKeeperServer getZks() {
+            return zks;
+        }
+
+        public ServerCnxn getCnxn() {
+            return cnxn;
+        }
+    }
+
+    public static class MatchValues {
+        private final String path;
+        private final String id;
+        private final String aclExpr;
+        private final int perm;
+        private final List<ACL> setAcls;
+
+        /**
+         * @param path
+         *                the path of the operation being authenticated
+         * @param id
+         *                the id to check.
+         * @param aclExpr
+         *                the expression to match ids against.
+         * @param perm
+         *                the permission value being authenticated
+         * @param setAcls
+         *                for set ACL operations, the list of ACLs being set. Otherwise null.
+         */
+        public MatchValues(String path, String id, String aclExpr, int perm, List<ACL> setAcls) {
+            this.path = path;
+            this.id = id;
+            this.aclExpr = aclExpr;
+            this.perm = perm;
+            this.setAcls = setAcls;
+        }
+
+        public String getPath() {
+            return path;
+        }
+
+        public String getId() {
+            return id;
+        }
+
+        public String getAclExpr() {
+            return aclExpr;
+        }
+
+        public int getPerm() {
+            return perm;
+        }
+
+        public List<ACL> getSetAcls() {
+            return setAcls;
+        }
+    }
+
+    /**
+     * This method is called when a client passes authentication data for this
+     * scheme. The authData is directly from the authentication packet. The
+     * implementor may attach new ids to the authInfo field of cnxn or may use
+     * cnxn to send packets back to the client.
+     *
+     * @param serverObjs
+     *                cnxn/server/etc that received the authentication information.
+     * @param authData
+     *                the authentication data received.
+     * @return indication of success or failure
+     */
+    public abstract KeeperException.Code handleAuthentication(ServerObjs serverObjs, byte authData[]);
+
+    /**
+     * This method is called to see if the given id matches the given id
+     * expression in the ACL. This allows schemes to use application specific
+     * wild cards.
+     *
+     * @param serverObjs
+     *                cnxn/server/etc that received the authentication information.
+     * @param matchValues
+     *                values to be matched
+     */
+    public abstract boolean matches(ServerObjs serverObjs, MatchValues matchValues);
+
+    @Override
+    public final KeeperException.Code handleAuthentication(ServerCnxn cnxn, byte[] authData) {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public final boolean matches(String id, String aclExpr) {
+        throw new UnsupportedOperationException();
+    }
+}

src/java/main/org/apache/zookeeper/server/auth/WrappedAuthenticationProvider.java

@@ -0,0 +1,77 @@
+/**
+ * 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.server.auth;
+
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.server.ServerCnxn;
+
+/**
+ * Provides backwards compatibility between older {@link AuthenticationProvider}
+ * implementations and the new {@link ServerAuthenticationProvider} interface.
+ */
+class WrappedAuthenticationProvider extends ServerAuthenticationProvider {
+    private final AuthenticationProvider implementation;
+
+    static ServerAuthenticationProvider wrap(AuthenticationProvider provider) {
+        if (provider == null) {
+            return null;
+        }
+        return (provider instanceof ServerAuthenticationProvider) ? (ServerAuthenticationProvider)provider
+                : new WrappedAuthenticationProvider(provider);
+    }
+
+    private WrappedAuthenticationProvider(AuthenticationProvider implementation) {
+        this.implementation = implementation;
+    }
+
+    /**
+     * {@inheritDoc}
+     *
+     * forwards to older method {@link #handleAuthentication(ServerCnxn, byte[])}
+     */
+    @Override
+    public KeeperException.Code handleAuthentication(ServerObjs serverObjs, byte[] authData) {
+        return implementation.handleAuthentication(serverObjs.getCnxn(), authData);
+    }
+
+    /**
+     * {@inheritDoc}
+     *
+     * forwards to older method {@link #matches(String, String)}
+     */
+    @Override
+    public boolean matches(ServerObjs serverObjs, MatchValues matchValues) {
+        return implementation.matches(matchValues.getId(), matchValues.getAclExpr());
+    }
+
+    @Override
+    public String getScheme() {
+        return implementation.getScheme();
+    }
+
+    @Override
+    public boolean isAuthenticated() {
+        return implementation.isAuthenticated();
+    }
+
+    @Override
+    public boolean isValid(String id) {
+        return implementation.isValid(id);
+    }
+}

src/java/test/org/apache/zookeeper/test/KeyAuthClientTest.java

@@ -0,0 +1,131 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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 org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.ZooDefs.Ids;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.data.ACL;
+import org.junit.Assert;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+
+public class KeyAuthClientTest extends ClientBase {
+    private static final Logger LOG = LoggerFactory.getLogger(KeyAuthClientTest.class);
+
+    static {
+        System.setProperty("zookeeper.authProvider.1", "org.apache.zookeeper.server.auth.KeyAuthenticationProvider");
+    }
+
+    public void createNodePrintAcl(ZooKeeper zk, String path, String testName) {
+        try {
+            LOG.debug("KeyAuthenticationProvider Creating Test Node:" + path + ".\n");
+            zk.create(path, null, Ids.CREATOR_ALL_ACL, CreateMode.PERSISTENT);
+            List<ACL> acls = zk.getACL(path, null);
+            LOG.debug("Node: " + path + " Test:" + testName + " ACLs:");
+            for (ACL acl : acls) {
+                LOG.debug("  " + acl.toString());
+            }
+        } catch (Exception e) {
+            LOG.debug("  EXCEPTION THROWN", e);
+        }
+    }
+
+    public void preAuth() throws Exception {
+        ZooKeeper zk = createClient();
+        zk.addAuthInfo("key", "25".getBytes());
+        try {
+            createNodePrintAcl(zk, "/pre", "testPreAuth");
+            zk.setACL("/", Ids.CREATOR_ALL_ACL, -1);
+            zk.getChildren("/", false);
+            zk.create("/abc", null, Ids.CREATOR_ALL_ACL, CreateMode.PERSISTENT);
+            zk.setData("/abc", "testData1".getBytes(), -1);
+            zk.create("/key", null, Ids.CREATOR_ALL_ACL, CreateMode.PERSISTENT);
+            zk.setData("/key", "5".getBytes(), -1);
+            Thread.sleep(1000);
+        } catch (KeeperException e) {
+            Assert.fail("test failed :" + e);
+        } finally {
+            zk.close();
+        }
+    }
+
+    public void missingAuth() throws Exception {
+        ZooKeeper zk = createClient();
+        try {
+            zk.getData("/abc", false, null);
+            Assert.fail("Should not be able to get data");
+        } catch (KeeperException correct) {
+            // correct
+        }
+        try {
+            zk.setData("/abc", "testData2".getBytes(), -1);
+            Assert.fail("Should not be able to set data");
+        } catch (KeeperException correct) {
+            // correct
+        } finally {
+            zk.close();
+        }
+    }
+
+    public void validAuth() throws Exception {
+        ZooKeeper zk = createClient();
+        // any multiple of 5 will do...
+        zk.addAuthInfo("key", "25".getBytes());
+        try {
+            createNodePrintAcl(zk, "/valid", "testValidAuth");
+            zk.getData("/abc", false, null);
+            zk.setData("/abc", "testData3".getBytes(), -1);
+        } catch (KeeperException.AuthFailedException e) {
+            Assert.fail("test failed :" + e);
+        } finally {
+            zk.close();
+        }
+    }
+
+    public void validAuth2() throws Exception {
+        ZooKeeper zk = createClient();
+        // any multiple of 5 will do...
+        zk.addAuthInfo("key", "125".getBytes());
+        try {
+            createNodePrintAcl(zk, "/valid2", "testValidAuth2");
+            zk.getData("/abc", false, null);
+            zk.setData("/abc", "testData3".getBytes(), -1);
+        } catch (KeeperException.AuthFailedException e) {
+            Assert.fail("test failed :" + e);
+        } finally {
+            zk.close();
+        }
+    }
+
+    @Test
+    public void testAuth() throws Exception {
+        // NOTE: the tests need to run in-order, and older versions of
+        // junit don't provide any way to order tests
+        preAuth();
+        missingAuth();
+        validAuth();
+        validAuth2();
+    }
+
+}