ZOOKEEPER-1634: hardening security by teaching server to enforce client authentication.

## Motivation
Previously ZooKeeper server is open to the world as the server does not enforce client authentication - anonymous clients are allowed to establish session with server. This behavior raises a couple of issues from the perspective of performance and security for example:
* It is easy to launch a DDoS attack to server, by having a fleet of anonymous clients connect to the ensemble, as each session would consume valuable resources (socket, memory, etc) from server.
* It is cumbersome to enforce certain security models with the presence of anonymous clients login - for example as clients are not trusted the root ACL has to be disabled for writes to world, among other configurations an admin has to do to secure a cluster in a multi-tenant environment.

So the goal here is to address such issue by hardening ZooKeeper security to provide a more confined access option that user could opt-in, which in addition to the existing ACLs together could lead to more secured / resource optimal ensemble.

## Design Abstract
* Introduce a new server side Java property that if set, ZooKeeper server will only accept connections and requests from clients that have authenticated with server via SASL.
* Clients that are not configured with SASL authentication, or configured with SASL but fail authentication (i.e. with invalid credential) will not be able to establish a session with server. A typed error code (-124) will be delivered in such case, both Java and C client will close the session with server thereafter, without further attempts on retrying to reconnect.
* This feature overrules the server property "zookeeper.allowSaslFailedClients". So even if server is configured to allow clients that fail SASL authentication to login, client will not be able to establish a session with server if this feature is enabled.
* Only support SASL because only SASL authentication has the property that no operations will happen until SASL authentication process finished. Thus, the decision of whether to close the session or not can be quickly made on server side upon receiving a client connection request. We could later add support for other auth scheme via add_auth_info if that's desired (if we do, then a session has to be maintained until add_auth_info is invoked.).
* As a side benefit, this PR fixes an issue mentioned in ZOOKEEPER-2346 by correctly propagate events from server to client side so a SASL auth failure will manifest as an auth / config failure rather than generic ConnectionLoss event.

JIRA: https://issues.apache.org/jira/browse/ZOOKEEPER-1634
The PR also covers (or part of):
https://issues.apache.org/jira/browse/ZOOKEEPER-2462
https://issues.apache.org/jira/browse/ZOOKEEPER-2526
https://issues.apache.org/jira/browse/ZOOKEEPER-2346

Author: Michael Han <lhan@twitter.com>

Reviewers: Enrico Olivelli <eolivelli@apache.org>, Andor Molnar <andor@apache.org>

Closes #118 from hanm/ZOOKEEPER-1634

4 files changed, 127 insertions, 2 deletions

diff --git a/zookeeper-client/zookeeper-client-c/Makefile.am b/zookeeper-client/zookeeper-client-c/Makefile.am
index 059b22eef..a5312a8e9 100644
--- a/zookeeper-client/zookeeper-client-c/Makefile.am
+++ b/zookeeper-client/zookeeper-client-c/Makefile.am
@@ -107,6 +107,7 @@ TEST_SOURCES = \
 	tests/ZooKeeperQuorumServer.h \
 	tests/TestReadOnlyClient.cc \
 	tests/TestLogClientEnv.cc \
+        tests/TestServerRequireClientSASLAuth.cc \
 	$(NULL)
 
 if SOLARIS
diff --git a/zookeeper-client/zookeeper-client-c/include/zookeeper.h b/zookeeper-client/zookeeper-client-c/include/zookeeper.h
index 801027d82..304dafe2f 100644
--- a/zookeeper-client/zookeeper-client-c/include/zookeeper.h
+++ b/zookeeper-client/zookeeper-client-c/include/zookeeper.h
@@ -130,7 +130,8 @@ enum ZOO_ERRORS {
   ZNOTREADONLY = -119, /*!< state-changing request is passed to read-only server */
   ZEPHEMERALONLOCALSESSION = -120, /*!< Attempt to create ephemeral node on a local session */
   ZNOWATCHER = -121, /*!< The watcher couldn't be found */
-  ZRECONFIGDISABLED = -123 /*!< Attempts to perform a reconfiguration operation when reconfiguration feature is disabled */
+  ZRECONFIGDISABLED = -123, /*!< Attempts to perform a reconfiguration operation when reconfiguration feature is disabled */
+  ZSESSIONCLOSEDREQUIRESASLAUTH = -124 /*!< The session has been closed by server because server requires client to do SASL authentication, but client is not configured with SASL authentication or configuted with SASL but failed (i.e. wrong credential used.). */
 };
 
 #ifdef __cplusplus
diff --git a/zookeeper-client/zookeeper-client-c/tests/TestServerRequireClientSASLAuth.cc b/zookeeper-client/zookeeper-client-c/tests/TestServerRequireClientSASLAuth.cc
new file mode 100644
index 000000000..2c5290ded
--- /dev/null
+++ b/zookeeper-client/zookeeper-client-c/tests/TestServerRequireClientSASLAuth.cc
@@ -0,0 +1,109 @@
+/**
+ * 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.
+ */
+
+#include <cppunit/extensions/HelperMacros.h>
+#include "CppAssertHelper.h"
+
+#include <sys/socket.h>
+#include <unistd.h>
+
+#include <zookeeper.h>
+
+#include "Util.h"
+#include "WatchUtil.h"
+
+ZOOAPI int zoo_create2(zhandle_t *zh, const char *path, const char *value,
+        int valuelen, const struct ACL_vector *acl, int mode,
+        char *path_buffer, int path_buffer_len, struct Stat *stat);
+
+class Zookeeper_serverRequireClientSASL : public CPPUNIT_NS::TestFixture {
+    CPPUNIT_TEST_SUITE(Zookeeper_serverRequireClientSASL);
+#ifdef THREADED
+    CPPUNIT_TEST(testServerRequireClientSASL);
+#endif
+    CPPUNIT_TEST_SUITE_END();
+    FILE *logfile;
+    static const char hostPorts[];
+    static void watcher(zhandle_t *, int type, int state, const char *path,void*v){
+        watchctx_t *ctx = (watchctx_t*)v;
+
+        if (state == ZOO_CONNECTED_STATE) {
+            ctx->connected = true;
+        } else {
+            ctx->connected = false;
+        }
+        if (type != ZOO_SESSION_EVENT) {
+            evt_t evt;
+            evt.path = path;
+            evt.type = type;
+            ctx->putEvent(evt);
+        }
+    }
+
+public:
+    Zookeeper_serverRequireClientSASL() {
+      logfile = openlogfile("Zookeeper_serverRequireClientSASL");
+    }
+
+    ~Zookeeper_serverRequireClientSASL() {
+      if (logfile) {
+        fflush(logfile);
+        fclose(logfile);
+        logfile = 0;
+      }
+    }
+
+    void setUp() {
+        zoo_set_log_stream(logfile);
+    }
+
+    void startServer() {
+        char cmd[1024];
+        sprintf(cmd, "%s startRequireSASLAuth", ZKSERVER_CMD);
+        CPPUNIT_ASSERT(system(cmd) == 0);
+    }
+
+    void stopServer() {
+        char cmd[1024];
+        sprintf(cmd, "%s stop", ZKSERVER_CMD);
+        CPPUNIT_ASSERT(system(cmd) == 0);
+    }
+
+    void testServerRequireClientSASL() {
+        startServer();
+
+        watchctx_t ctx;
+        int rc = 0;
+        zhandle_t *zk = zookeeper_init(hostPorts, watcher, 10000, 0, &ctx, 0);
+        ctx.zh = zk;
+        CPPUNIT_ASSERT(zk);
+
+        char pathbuf[80];
+        struct Stat stat_a = {0};
+
+        rc = zoo_create2(zk, "/serverRequireClientSASL", "", 0,
+                         &ZOO_OPEN_ACL_UNSAFE, 0, pathbuf, sizeof(pathbuf), &stat_a);
+        CPPUNIT_ASSERT_EQUAL((int)ZSESSIONCLOSEDREQUIRESASLAUTH, rc);
+
+        stopServer();
+    }
+};
+
+const char Zookeeper_serverRequireClientSASL::hostPorts[] = "127.0.0.1:23456";
+
+CPPUNIT_TEST_SUITE_REGISTRATION(Zookeeper_serverRequireClientSASL);
diff --git a/zookeeper-client/zookeeper-client-c/tests/zkServer.sh b/zookeeper-client/zookeeper-client-c/tests/zkServer.sh
index af558eab3..ebc3df42b 100755
--- a/zookeeper-client/zookeeper-client-c/tests/zkServer.sh
+++ b/zookeeper-client/zookeeper-client-c/tests/zkServer.sh
@@ -21,7 +21,7 @@ ZOOPORT=22181
 
 if [ "x$1" == "x" ]
 then
-    echo "USAGE: $0 startClean|start|startReadOnly|stop hostPorts"
+    echo "USAGE: $0 startClean|start|startReadOnly|startRequireSASLAuth|stop hostPorts"
     exit 2
 fi
 
@@ -180,6 +180,20 @@ startReadOnly)
     fi
 
     ;;
+startRequireSASLAuth)
+    if [ "x${base_dir}" == "x" ]
+    then
+        echo "this target is for unit tests only"
+        exit 2
+    else
+        mkdir -p "${base_dir}/build/tmp/zkdata"
+        java -cp "$CLASSPATH" -Dzookeeper.sessionRequireClientSASLAuth=true org.apache.zookeeper.server.ZooKeeperServerMain 23456 "${base_dir}/build/tmp/zkdata" 3000 $ZKMAXCNXNS &> "${base_dir}/build/tmp/zk.log" &
+        pid=$!
+        echo -n $pid > "${base_dir}/build/tmp/zk.pid"
+        sleep 3 # wait until server is up.
+    fi
+
+    ;;
 stop)
     # Already killed above
     ;;