You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@zookeeper.apache.org by ph...@apache.org on 2017/11/27 23:11:32 UTC
[3/3] zookeeper git commit: ZOOKEEPER-2935: [QP MutualAuth]: Port
ZOOKEEPER-1045 implementation from branch-3.5 to trunk
ZOOKEEPER-2935: [QP MutualAuth]: Port ZOOKEEPER-1045 implementation from branch-3.5 to trunk
Author: Abraham Fine <af...@apache.org>
Reviewers: phunt@apache.org
Closes #417 from afine/ZOOKEEPER-2935 and squashes the following commits:
99bfbe94 [Abraham Fine] Add debugging line and improve ivy.xml by removing unnecessary excludes
1d6c7de5 [Abraham Fine] Fix missing test.data.kerberos.dir in build.xml
06d0b6fa [Abraham Fine] ZOOKEEPER-2935: [QP MutualAuth]: Port ZOOKEEPER-1045 implementation from branch-3.5 to trunk.
Change-Id: I2b88221c0006e4336a39f74fd5a87d1aded68c90
Project: http://git-wip-us.apache.org/repos/asf/zookeeper/repo
Commit: http://git-wip-us.apache.org/repos/asf/zookeeper/commit/75411ab3
Tree: http://git-wip-us.apache.org/repos/asf/zookeeper/tree/75411ab3
Diff: http://git-wip-us.apache.org/repos/asf/zookeeper/diff/75411ab3
Branch: refs/heads/master
Commit: 75411ab34a3d53c43c2d508b12314a9788aa417d
Parents: 748585d
Author: Abraham Fine <af...@apache.org>
Authored: Mon Nov 27 15:10:13 2017 -0800
Committer: Patrick Hunt <ph...@apache.org>
Committed: Mon Nov 27 15:10:13 2017 -0800
----------------------------------------------------------------------
build.xml | 9 +
ivy.xml | 33 ++
src/java/main/org/apache/zookeeper/Login.java | 2 +-
.../zookeeper/SaslClientCallbackHandler.java | 104 +++++
.../zookeeper/client/ZooKeeperSaslClient.java | 164 +-------
.../org/apache/zookeeper/server/ServerCnxn.java | 8 +-
.../zookeeper/server/ZooKeeperSaslServer.java | 121 +-----
.../zookeeper/server/ZooKeeperServer.java | 1 +
.../server/auth/SaslServerCallbackHandler.java | 10 +-
.../server/quorum/FastLeaderElection.java | 2 +
.../zookeeper/server/quorum/Follower.java | 5 +-
.../apache/zookeeper/server/quorum/Leader.java | 10 +-
.../apache/zookeeper/server/quorum/Learner.java | 18 +-
.../zookeeper/server/quorum/LearnerHandler.java | 27 +-
.../zookeeper/server/quorum/Observer.java | 8 +-
.../server/quorum/QuorumCnxManager.java | 300 +++++++++++--
.../zookeeper/server/quorum/QuorumPeer.java | 198 ++++++++-
.../server/quorum/QuorumPeerConfig.java | 48 +++
.../zookeeper/server/quorum/QuorumPeerMain.java | 15 +-
.../quorum/auth/NullQuorumAuthLearner.java | 33 ++
.../quorum/auth/NullQuorumAuthServer.java | 34 ++
.../server/quorum/auth/QuorumAuth.java | 96 +++++
.../server/quorum/auth/QuorumAuthLearner.java | 40 ++
.../server/quorum/auth/QuorumAuthServer.java | 41 ++
.../quorum/auth/SaslQuorumAuthLearner.java | 223 ++++++++++
.../quorum/auth/SaslQuorumAuthServer.java | 179 ++++++++
.../auth/SaslQuorumServerCallbackHandler.java | 148 +++++++
.../apache/zookeeper/util/SecurityUtils.java | 298 +++++++++++++
src/java/test/config/findbugsExcludeFile.xml | 5 +
src/java/test/data/kerberos/minikdc-krb5.conf | 30 ++
src/java/test/data/kerberos/minikdc.ldiff | 52 +++
.../quorum/EphemeralNodeDeletionTest.java | 8 +-
.../quorum/FLEBackwardElectionRoundTest.java | 4 +-
.../server/quorum/FLELostMessageTest.java | 2 +-
.../server/quorum/LearnerHandlerTest.java | 3 +-
.../zookeeper/server/quorum/LearnerTest.java | 4 +-
.../server/quorum/QuorumPeerTestBase.java | 97 ++++-
.../server/quorum/RaceConditionTest.java | 7 +-
.../zookeeper/server/quorum/Zab1_0Test.java | 45 +-
.../quorum/auth/KerberosSecurityTestcase.java | 120 ++++++
.../server/quorum/auth/KerberosTestUtils.java | 76 ++++
.../zookeeper/server/quorum/auth/MiniKdc.java | 418 +++++++++++++++++++
.../server/quorum/auth/MiniKdcTest.java | 185 ++++++++
.../server/quorum/auth/QuorumAuthTestBase.java | 146 +++++++
.../quorum/auth/QuorumAuthUpgradeTest.java | 239 +++++++++++
.../quorum/auth/QuorumDigestAuthTest.java | 222 ++++++++++
.../quorum/auth/QuorumKerberosAuthTest.java | 110 +++++
.../auth/QuorumKerberosHostBasedAuthTest.java | 184 ++++++++
.../apache/zookeeper/test/CnxManagerTest.java | 12 +-
.../apache/zookeeper/test/FLEPredicateTest.java | 2 +-
src/zookeeper.jute | 5 +
51 files changed, 3755 insertions(+), 396 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/75411ab3/build.xml
----------------------------------------------------------------------
diff --git a/build.xml b/build.xml
index 7528487..e61f7d7 100644
--- a/build.xml
+++ b/build.xml
@@ -79,6 +79,7 @@ xmlns:cs="antlib:com.puppycrawl.tools.checkstyle.ant">
<property name="test.data.dir" value="${test.java.build.dir}/data" />
<property name="test.data.invalid.dir" value="${test.data.dir}/invalidsnap" />
<property name="test.data.buffersize.dir" value="${test.data.dir}/buffersize" />
+ <property name="test.data.kerberos.dir" value="${test.data.dir}/kerberos" />
<property name="test.data.ssl.dir" value="${test.data.dir}/ssl" />
<property name="test.cppunit.dir" value="${test.java.build.dir}/test-cppunit"/>
<property name="test.tmp.dir" value="${test.java.build.dir}/tmp" />
@@ -226,6 +227,9 @@ xmlns:cs="antlib:com.puppycrawl.tools.checkstyle.ant">
<property name="jackson-mapper-asl.version" value="1.9.11"/>
<property name="dependency-check-ant.version" value="2.1.0"/>
+ <property name="commons-io.version" value="2.4"/>
+ <property name="kerby.version" value="1.0.0-RC2"/>
+
<!-- ====================================================== -->
<!-- Macro definitions -->
<!-- ====================================================== -->
@@ -1165,6 +1169,7 @@ xmlns:cs="antlib:com.puppycrawl.tools.checkstyle.ant">
<delete dir="${test.tmp.dir}" />
<delete dir="${test.data.invalid.dir}" />
<delete dir="${test.data.buffersize.dir}" />
+ <delete dir="${test.data.kerberos.dir}" />
<delete dir="${test.data.ssl.dir}" />
<delete dir="${test.data.dir}" />
<mkdir dir="${test.log.dir}" />
@@ -1182,6 +1187,10 @@ xmlns:cs="antlib:com.puppycrawl.tools.checkstyle.ant">
<copy todir="${test.data.ssl.dir}">
<fileset dir="${basedir}/src/java/test/data/ssl"/>
</copy>
+ <mkdir dir="${test.data.kerberos.dir}" />
+ <copy todir="${test.data.kerberos.dir}">
+ <fileset dir="${basedir}/src/java/test/data/kerberos"/>
+ </copy>
</target>
<target name="junit-init" depends="test-category">
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/75411ab3/ivy.xml
----------------------------------------------------------------------
diff --git a/ivy.xml b/ivy.xml
index 8a9c603..4298aa9 100644
--- a/ivy.xml
+++ b/ivy.xml
@@ -90,6 +90,39 @@
<dependency org="org.owasp" name="dependency-check-ant"
rev="${dependency-check-ant.version}" conf="owasp->default"/>
+ <dependency org="commons-io" name="commons-io"
+ rev="${commons-io.version}" conf="test->default"/>
+
+ <dependency org="org.apache.kerby" name="kerby-config" rev="${kerby.version}" conf="test->default">
+ <exclude org="org.slf4j" module="slf4j-api"/>
+ <exclude org="org.slf4j" module="slf4j-log4j12"/>
+ </dependency>
+ <dependency org="org.apache.kerby" name="kerb-simplekdc" rev="${kerby.version}" conf="test->default">
+ <exclude org="org.slf4j" module="slf4j-api"/>
+ </dependency>
+ <dependency org="org.apache.kerby" name="kerb-core"
+ rev="${kerby.version}" conf="test->default">
+ <exclude org="org.slf4j" module="slf4j-api"/>
+ </dependency>
+ <dependency org="org.apache.kerby" name="kerb-server"
+ rev="${kerby.version}" conf="test->default"/>
+ <dependency org="org.apache.kerby" name="kerb-common"
+ rev="${kerby.version}" conf="test->default"/>
+ <dependency org="org.apache.kerby" name="kerb-admin"
+ rev="${kerby.version}" conf="test->default"/>
+ <dependency org="org.apache.kerby" name="kerb-identity"
+ rev="${kerby.version}" conf="test->default"/>
+ <dependency org="org.apache.kerby" name="kerb-client"
+ rev="${kerby.version}" conf="test->default"/>
+ <dependency org="org.apache.kerby" name="kerb-util"
+ rev="${kerby.version}" conf="test->default"/>
+ <dependency org="org.apache.kerby" name="kerb-crypto"
+ rev="${kerby.version}" conf="test->default"/>
+ <dependency org="org.apache.kerby" name="kerby-util"
+ rev="${kerby.version}" conf="test->default"/>
+ <dependency org="org.apache.kerby" name="kerby-asn1"
+ rev="${kerby.version}" conf="test->default"/>
+
<dependency org="net.java.dev.javacc" name="javacc" rev="${javacc.version}"
conf="javacc->default" />
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/75411ab3/src/java/main/org/apache/zookeeper/Login.java
----------------------------------------------------------------------
diff --git a/src/java/main/org/apache/zookeeper/Login.java b/src/java/main/org/apache/zookeeper/Login.java
index fd9a4c2..d97d6c1 100644
--- a/src/java/main/org/apache/zookeeper/Login.java
+++ b/src/java/main/org/apache/zookeeper/Login.java
@@ -299,7 +299,7 @@ public class Login {
}
LoginContext loginContext = new LoginContext(loginContextName,callbackHandler);
loginContext.login();
- LOG.info("successfully logged in.");
+ LOG.info("{} successfully logged in.", loginContextName);
return loginContext;
}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/75411ab3/src/java/main/org/apache/zookeeper/SaslClientCallbackHandler.java
----------------------------------------------------------------------
diff --git a/src/java/main/org/apache/zookeeper/SaslClientCallbackHandler.java b/src/java/main/org/apache/zookeeper/SaslClientCallbackHandler.java
new file mode 100644
index 0000000..d6f5549
--- /dev/null
+++ b/src/java/main/org/apache/zookeeper/SaslClientCallbackHandler.java
@@ -0,0 +1,104 @@
+/**
+ * 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;
+
+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.sasl.AuthorizeCallback;
+import javax.security.sasl.RealmCallback;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This is used by the SASL mechanisms to get further information to complete
+ * the authentication. For example, a SASL mechanism might use this callback
+ * handler to do verification operation. The CallbackHandler interface here
+ * refers to javax.security.auth.callback.CallbackHandler. It should not be
+ * confused with ZooKeeper packet callbacks like
+ * org.apache.zookeeper.server.auth.SaslServerCallbackHandler.
+ */
+public class SaslClientCallbackHandler implements CallbackHandler {
+ private String password = null;
+ private static final Logger LOG = LoggerFactory.getLogger(SaslClientCallbackHandler.class);
+ private final String entity;
+ public SaslClientCallbackHandler(String password, String client) {
+ this.password = password;
+ this.entity = client;
+ }
+
+ public void handle(Callback[] callbacks) throws UnsupportedCallbackException {
+ for (Callback callback : callbacks) {
+ if (callback instanceof NameCallback) {
+ NameCallback nc = (NameCallback) callback;
+ nc.setName(nc.getDefaultName());
+ }
+ else {
+ if (callback instanceof PasswordCallback) {
+ PasswordCallback pc = (PasswordCallback)callback;
+ if (password != null) {
+ pc.setPassword(this.password.toCharArray());
+ } else {
+ LOG.warn("Could not login: the {} is being asked for a password, but the ZooKeeper {}" +
+ " code does not currently support obtaining a password from the user." +
+ " Make sure that the {} is configured to use a ticket cache (using" +
+ " the JAAS configuration setting 'useTicketCache=true)' and restart the {}. If" +
+ " you still get this message after that, the TGT in the ticket cache has expired and must" +
+ " be manually refreshed. To do so, first determine if you are using a password or a" +
+ " keytab. If the former, run kinit in a Unix shell in the environment of the user who" +
+ " is running this Zookeeper {} using the command" +
+ " 'kinit <princ>' (where <princ> is the name of the {}'s Kerberos principal)." +
+ " If the latter, do" +
+ " 'kinit -k -t <keytab> <princ>' (where <princ> is the name of the Kerberos principal, and" +
+ " <keytab> is the location of the keytab file). After manually refreshing your cache," +
+ " restart this {}. If you continue to see this message after manually refreshing" +
+ " your cache, ensure that your KDC host's clock is in sync with this host's clock.",
+ new Object[]{entity, entity, entity, entity, entity, entity, entity});
+ }
+ }
+ else {
+ if (callback instanceof RealmCallback) {
+ RealmCallback rc = (RealmCallback) callback;
+ rc.setText(rc.getDefaultText());
+ }
+ else {
+ if (callback instanceof AuthorizeCallback) {
+ AuthorizeCallback ac = (AuthorizeCallback) callback;
+ String authid = ac.getAuthenticationID();
+ String authzid = ac.getAuthorizationID();
+ if (authid.equals(authzid)) {
+ ac.setAuthorized(true);
+ } else {
+ ac.setAuthorized(false);
+ }
+ if (ac.isAuthorized()) {
+ ac.setAuthorizedID(authzid);
+ }
+ }
+ else {
+ throw new UnsupportedCallbackException(callback, "Unrecognized SASL " + entity + "Callback");
+ }
+ }
+ }
+ }
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/75411ab3/src/java/main/org/apache/zookeeper/client/ZooKeeperSaslClient.java
----------------------------------------------------------------------
diff --git a/src/java/main/org/apache/zookeeper/client/ZooKeeperSaslClient.java b/src/java/main/org/apache/zookeeper/client/ZooKeeperSaslClient.java
index f921ca3..8550626 100644
--- a/src/java/main/org/apache/zookeeper/client/ZooKeeperSaslClient.java
+++ b/src/java/main/org/apache/zookeeper/client/ZooKeeperSaslClient.java
@@ -19,40 +19,27 @@
package org.apache.zookeeper.client;
import java.io.IOException;
-import java.security.Principal;
import java.security.PrivilegedActionException;
import java.security.PrivilegedExceptionAction;
import javax.security.auth.Subject;
-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.auth.login.LoginException;
-import javax.security.sasl.AuthorizeCallback;
-import javax.security.sasl.RealmCallback;
-import javax.security.sasl.Sasl;
import javax.security.sasl.SaslClient;
import javax.security.sasl.SaslException;
import org.apache.zookeeper.AsyncCallback;
import org.apache.zookeeper.ClientCnxn;
import org.apache.zookeeper.Login;
+import org.apache.zookeeper.SaslClientCallbackHandler;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs;
-import org.apache.zookeeper.common.ZKConfig;
import org.apache.zookeeper.data.Stat;
import org.apache.zookeeper.proto.GetSASLRequest;
import org.apache.zookeeper.proto.SetSASLResponse;
-import org.apache.zookeeper.server.auth.KerberosName;
-import org.ietf.jgss.GSSContext;
-import org.ietf.jgss.GSSCredential;
-import org.ietf.jgss.GSSException;
-import org.ietf.jgss.GSSManager;
-import org.ietf.jgss.Oid;
+
+import org.apache.zookeeper.util.SecurityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -259,84 +246,14 @@ public class ZooKeeperSaslClient {
}
// 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 = new Login(loginContext, new ClientCallbackHandler(null), clientConfig);
+ login = new Login(loginContext, new SaslClientCallbackHandler(null, "Client"), clientConfig);
login.startThreadIfNeeded();
initializedLogin = true;
}
}
}
- Subject subject = login.getSubject();
- SaslClient saslClient;
- // Use subject.getPrincipals().isEmpty() as an indication of which SASL mechanism to use:
- // if empty, use DIGEST-MD5; otherwise, use GSSAPI.
- if (subject.getPrincipals().isEmpty()) {
- // no principals: must not be GSSAPI: use DIGEST-MD5 mechanism instead.
- LOG.info("Client will use DIGEST-MD5 as SASL mechanism.");
- String[] mechs = {"DIGEST-MD5"};
- String username = (String)(subject.getPublicCredentials().toArray()[0]);
- String password = (String)(subject.getPrivateCredentials().toArray()[0]);
- // "zk-sasl-md5" is a hard-wired 'domain' parameter shared with zookeeper server code (see ServerCnxnFactory.java)
- saslClient = Sasl.createSaslClient(mechs, username, "zookeeper", "zk-sasl-md5", null, new ClientCallbackHandler(password));
- return saslClient;
- }
- else { // GSSAPI.
- boolean usingNativeJgss = clientConfig.getBoolean(ZKConfig.JGSS_NATIVE);
- if (usingNativeJgss) {
- // http://docs.oracle.com/javase/6/docs/technotes/guides/security/jgss/jgss-features.html
- // """
- // In addition, when performing operations as a particular
- // Subject, e.g. Subject.doAs(...) or Subject.doAsPrivileged(...),
- // the to-be-used GSSCredential should be added to Subject's
- // private credential set. Otherwise, the GSS operations will
- // fail since no credential is found.
- // """
- try {
- GSSManager manager = GSSManager.getInstance();
- Oid krb5Mechanism = new Oid("1.2.840.113554.1.2.2");
- GSSCredential cred = manager.createCredential(null,
- GSSContext.DEFAULT_LIFETIME,
- krb5Mechanism,
- GSSCredential.INITIATE_ONLY);
- subject.getPrivateCredentials().add(cred);
- if (LOG.isDebugEnabled()) {
- LOG.debug("Added private credential to subject: " + cred);
- }
- } catch (GSSException ex) {
- LOG.warn("Cannot add private credential to subject; " +
- "authentication at the server may fail", ex);
- }
- }
- final Object[] principals = subject.getPrincipals().toArray();
- // determine client principal from subject.
- final Principal clientPrincipal = (Principal)principals[0];
- final KerberosName clientKerberosName = new KerberosName(clientPrincipal.getName());
- // assume that server and client are in the same realm (by default; unless the system property
- // "zookeeper.server.realm" is set).
- String serverRealm = clientConfig.getProperty(
- ZKClientConfig.ZOOKEEPER_SERVER_REALM,
- clientKerberosName.getRealm());
- KerberosName serviceKerberosName = new KerberosName(servicePrincipal+"@"+serverRealm);
- final String serviceName = serviceKerberosName.getServiceName();
- final String serviceHostname = serviceKerberosName.getHostName();
- final String clientPrincipalName = clientKerberosName.toString();
- try {
- saslClient = Subject.doAs(subject,new PrivilegedExceptionAction<SaslClient>() {
- public SaslClient run() throws SaslException {
- LOG.info("Client will use GSSAPI as SASL mechanism.");
- String[] mechs = {"GSSAPI"};
- LOG.debug("creating sasl client: client="+clientPrincipalName+";service="+serviceName+";serviceHostname="+serviceHostname);
- SaslClient saslClient = Sasl.createSaslClient(mechs,clientPrincipalName,serviceName,serviceHostname,null,new ClientCallbackHandler(null));
- return saslClient;
- }
- });
- return saslClient;
- }
- catch (Exception e) {
- LOG.error("Exception while trying to create SASL client", e);
- e.printStackTrace();
- return null;
- }
- }
+ return SecurityUtils.createSaslClient(login.getSubject(),
+ servicePrincipal, "zookeeper", "zk-sasl-md5", LOG, "Client");
} catch (LoginException e) {
// We throw LoginExceptions...
throw e;
@@ -505,75 +422,6 @@ public class ZooKeeperSaslClient {
}
}
- // The CallbackHandler interface here refers to
- // javax.security.auth.callback.CallbackHandler.
- // It should not be confused with Zookeeper packet callbacks like
- // org.apache.zookeeper.server.auth.SaslServerCallbackHandler.
- public static class ClientCallbackHandler implements CallbackHandler {
- private String password = null;
-
- public ClientCallbackHandler(String password) {
- this.password = password;
- }
-
- public void handle(Callback[] callbacks) throws
- UnsupportedCallbackException {
- for (Callback callback : callbacks) {
- if (callback instanceof NameCallback) {
- NameCallback nc = (NameCallback) callback;
- nc.setName(nc.getDefaultName());
- }
- else {
- if (callback instanceof PasswordCallback) {
- PasswordCallback pc = (PasswordCallback)callback;
- if (password != null) {
- pc.setPassword(this.password.toCharArray());
- } else {
- LOG.warn("Could not login: the client is being asked for a password, but the Zookeeper" +
- " client code does not currently support obtaining a password from the user." +
- " Make sure that the client is configured to use a ticket cache (using" +
- " the JAAS configuration setting 'useTicketCache=true)' and restart the client. If" +
- " you still get this message after that, the TGT in the ticket cache has expired and must" +
- " be manually refreshed. To do so, first determine if you are using a password or a" +
- " keytab. If the former, run kinit in a Unix shell in the environment of the user who" +
- " is running this Zookeeper client using the command" +
- " 'kinit <princ>' (where <princ> is the name of the client's Kerberos principal)." +
- " If the latter, do" +
- " 'kinit -k -t <keytab> <princ>' (where <princ> is the name of the Kerberos principal, and" +
- " <keytab> is the location of the keytab file). After manually refreshing your cache," +
- " restart this client. If you continue to see this message after manually refreshing" +
- " your cache, ensure that your KDC host's clock is in sync with this host's clock.");
- }
- }
- else {
- if (callback instanceof RealmCallback) {
- RealmCallback rc = (RealmCallback) callback;
- rc.setText(rc.getDefaultText());
- }
- else {
- if (callback instanceof AuthorizeCallback) {
- AuthorizeCallback ac = (AuthorizeCallback) callback;
- String authid = ac.getAuthenticationID();
- String authzid = ac.getAuthorizationID();
- if (authid.equals(authzid)) {
- ac.setAuthorized(true);
- } else {
- ac.setAuthorized(false);
- }
- if (ac.isAuthorized()) {
- ac.setAuthorizedID(authzid);
- }
- }
- else {
- throw new UnsupportedCallbackException(callback,"Unrecognized SASL ClientCallback");
- }
- }
- }
- }
- }
- }
- }
-
public boolean clientTunneledAuthenticationInProgress() {
if (!isSASLConfigured) {
return false;
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/75411ab3/src/java/main/org/apache/zookeeper/server/ServerCnxn.java
----------------------------------------------------------------------
diff --git a/src/java/main/org/apache/zookeeper/server/ServerCnxn.java b/src/java/main/org/apache/zookeeper/server/ServerCnxn.java
index e469cda..f8a90cf 100644
--- a/src/java/main/org/apache/zookeeper/server/ServerCnxn.java
+++ b/src/java/main/org/apache/zookeeper/server/ServerCnxn.java
@@ -24,7 +24,13 @@ import java.io.StringWriter;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.security.cert.Certificate;
-import java.util.*;
+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.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/75411ab3/src/java/main/org/apache/zookeeper/server/ZooKeeperSaslServer.java
----------------------------------------------------------------------
diff --git a/src/java/main/org/apache/zookeeper/server/ZooKeeperSaslServer.java b/src/java/main/org/apache/zookeeper/server/ZooKeeperSaslServer.java
index c792378..af19e5d 100644
--- a/src/java/main/org/apache/zookeeper/server/ZooKeeperSaslServer.java
+++ b/src/java/main/org/apache/zookeeper/server/ZooKeeperSaslServer.java
@@ -18,22 +18,12 @@
package org.apache.zookeeper.server;
-import java.security.Principal;
-import java.security.PrivilegedActionException;
-import java.security.PrivilegedExceptionAction;
-
import javax.security.auth.Subject;
-import javax.security.sasl.Sasl;
import javax.security.sasl.SaslException;
import javax.security.sasl.SaslServer;
import org.apache.zookeeper.Login;
-import org.ietf.jgss.GSSContext;
-import org.ietf.jgss.GSSCredential;
-import org.ietf.jgss.GSSException;
-import org.ietf.jgss.GSSManager;
-import org.ietf.jgss.GSSName;
-import org.ietf.jgss.Oid;
+import org.apache.zookeeper.util.SecurityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -51,114 +41,9 @@ public class ZooKeeperSaslServer {
private SaslServer createSaslServer(final Login login) {
synchronized (login) {
Subject subject = login.getSubject();
- if (subject != null) {
- // server is using a JAAS-authenticated subject: determine service principal name and hostname from zk server's subject.
- if (subject.getPrincipals().size() > 0) {
- try {
- final Object[] principals = subject.getPrincipals().toArray();
- final Principal servicePrincipal = (Principal)principals[0];
-
- // e.g. servicePrincipalNameAndHostname := "zookeeper/myhost.foo.com@FOO.COM"
- final String servicePrincipalNameAndHostname = servicePrincipal.getName();
-
- int indexOf = servicePrincipalNameAndHostname.indexOf("/");
-
- // e.g. serviceHostnameAndKerbDomain := "myhost.foo.com@FOO.COM"
- final String serviceHostnameAndKerbDomain = servicePrincipalNameAndHostname.substring(indexOf+1,servicePrincipalNameAndHostname.length());
-
- int indexOfAt = serviceHostnameAndKerbDomain.indexOf("@");
-
- // Handle Kerberos Service as well as User Principal Names
- final String servicePrincipalName, serviceHostname;
- if (indexOf > 0){
- // e.g. servicePrincipalName := "zookeeper"
- servicePrincipalName = servicePrincipalNameAndHostname.substring(0, indexOf);
- // e.g. serviceHostname := "myhost.foo.com"
- serviceHostname = serviceHostnameAndKerbDomain.substring(0, indexOfAt);
- } else {
- servicePrincipalName = servicePrincipalNameAndHostname.substring(0, indexOfAt);
- serviceHostname = null;
- }
-
- final String mech = "GSSAPI"; // TODO: should depend on zoo.cfg specified mechs, but if subject is non-null, it can be assumed to be GSSAPI.
-
- LOG.debug("serviceHostname is '"+ serviceHostname + "'");
- LOG.debug("servicePrincipalName is '"+ servicePrincipalName + "'");
- LOG.debug("SASL mechanism(mech) is '"+ mech +"'");
-
- boolean usingNativeJgss =
- Boolean.getBoolean("sun.security.jgss.native");
- if (usingNativeJgss) {
- // http://docs.oracle.com/javase/6/docs/technotes/guides/security/jgss/jgss-features.html
- // """
- // In addition, when performing operations as a particular
- // Subject, e.g. Subject.doAs(...) or
- // Subject.doAsPrivileged(...), the to-be-used
- // GSSCredential should be added to Subject's
- // private credential set. Otherwise, the GSS operations
- // will fail since no credential is found.
- // """
- try {
- GSSManager manager = GSSManager.getInstance();
- Oid krb5Mechanism = new Oid("1.2.840.113554.1.2.2");
- GSSName gssName = manager.createName(
- servicePrincipalName + "@" + serviceHostname,
- GSSName.NT_HOSTBASED_SERVICE);
- GSSCredential cred = manager.createCredential(gssName,
- GSSContext.DEFAULT_LIFETIME,
- krb5Mechanism,
- GSSCredential.ACCEPT_ONLY);
- subject.getPrivateCredentials().add(cred);
- if (LOG.isDebugEnabled()) {
- LOG.debug("Added private credential to subject: " + cred);
- }
- } catch (GSSException ex) {
- LOG.warn("Cannot add private credential to subject; " +
- "clients authentication may fail", ex);
- }
- }
- try {
- return Subject.doAs(subject,new PrivilegedExceptionAction<SaslServer>() {
- public SaslServer run() {
- try {
- SaslServer saslServer;
- saslServer = Sasl.createSaslServer(mech, servicePrincipalName, serviceHostname, null, login.callbackHandler);
- return saslServer;
- }
- catch (SaslException e) {
- LOG.error("Zookeeper Server failed to create a SaslServer to interact with a client during session initiation: " + e);
- e.printStackTrace();
- return null;
- }
- }
- }
- );
- }
- catch (PrivilegedActionException e) {
- // TODO: exit server at this point(?)
- LOG.error("Zookeeper Quorum member experienced a PrivilegedActionException exception while creating a SaslServer using a JAAS principal context:" + e);
- e.printStackTrace();
- }
- }
- catch (IndexOutOfBoundsException e) {
- LOG.error("server principal name/hostname determination error: ", e);
- }
- }
- else {
- // JAAS non-GSSAPI authentication: assuming and supporting only DIGEST-MD5 mechanism for now.
- // TODO: use 'authMech=' value in zoo.cfg.
- try {
- SaslServer saslServer = Sasl.createSaslServer("DIGEST-MD5","zookeeper","zk-sasl-md5",null, login.callbackHandler);
- return saslServer;
- }
- catch (SaslException e) {
- LOG.error("Zookeeper Quorum member failed to create a SaslServer to interact with a client during session initiation", e);
- }
- }
- }
+ return SecurityUtils.createSaslServer(subject, "zookeeper",
+ "zk-sasl-md5", login.callbackHandler, LOG);
}
- LOG.error("failed to create saslServer object.");
- return null;
}
public byte[] evaluateResponse(byte[] response) throws SaslException {
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/75411ab3/src/java/main/org/apache/zookeeper/server/ZooKeeperServer.java
----------------------------------------------------------------------
diff --git a/src/java/main/org/apache/zookeeper/server/ZooKeeperServer.java b/src/java/main/org/apache/zookeeper/server/ZooKeeperServer.java
index 11b6981..3ac1ddd 100644
--- a/src/java/main/org/apache/zookeeper/server/ZooKeeperServer.java
+++ b/src/java/main/org/apache/zookeeper/server/ZooKeeperServer.java
@@ -61,6 +61,7 @@ import org.apache.zookeeper.server.RequestProcessor.RequestProcessorException;
import org.apache.zookeeper.server.ServerCnxn.CloseRequestException;
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;
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/75411ab3/src/java/main/org/apache/zookeeper/server/auth/SaslServerCallbackHandler.java
----------------------------------------------------------------------
diff --git a/src/java/main/org/apache/zookeeper/server/auth/SaslServerCallbackHandler.java b/src/java/main/org/apache/zookeeper/server/auth/SaslServerCallbackHandler.java
index 7fdffde..9f53a4d 100644
--- a/src/java/main/org/apache/zookeeper/server/auth/SaslServerCallbackHandler.java
+++ b/src/java/main/org/apache/zookeeper/server/auth/SaslServerCallbackHandler.java
@@ -46,13 +46,15 @@ public class SaslServerCallbackHandler implements CallbackHandler {
private String userName;
private final Map<String,String> credentials = new HashMap<String,String>();
- public SaslServerCallbackHandler(Configuration configuration) throws IOException {
- String serverSection = System.getProperty(ZooKeeperSaslServer.LOGIN_CONTEXT_NAME_KEY,
- ZooKeeperSaslServer.DEFAULT_LOGIN_CONTEXT_NAME);
+ 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 'Server' entry in this configuration: Server cannot start.";
+ String errorMessage = "Could not find a '" + serverSection + "' entry in this configuration: Server cannot start.";
LOG.error(errorMessage);
throw new IOException(errorMessage);
}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/75411ab3/src/java/main/org/apache/zookeeper/server/quorum/FastLeaderElection.java
----------------------------------------------------------------------
diff --git a/src/java/main/org/apache/zookeeper/server/quorum/FastLeaderElection.java b/src/java/main/org/apache/zookeeper/server/quorum/FastLeaderElection.java
index 4e85ce0..f1112b7 100644
--- a/src/java/main/org/apache/zookeeper/server/quorum/FastLeaderElection.java
+++ b/src/java/main/org/apache/zookeeper/server/quorum/FastLeaderElection.java
@@ -1078,6 +1078,8 @@ public class FastLeaderElection implements Election {
LOG.warn("Failed to unregister with JMX", e);
}
self.jmxLeaderElectionBean = null;
+ LOG.debug("Number of connection processing threads: {}",
+ manager.getConnectionThreadCount());
}
}
}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/75411ab3/src/java/main/org/apache/zookeeper/server/quorum/Follower.java
----------------------------------------------------------------------
diff --git a/src/java/main/org/apache/zookeeper/server/quorum/Follower.java b/src/java/main/org/apache/zookeeper/server/quorum/Follower.java
index 1cdc202..84d29c8 100644
--- a/src/java/main/org/apache/zookeeper/server/quorum/Follower.java
+++ b/src/java/main/org/apache/zookeeper/server/quorum/Follower.java
@@ -27,6 +27,7 @@ import org.apache.zookeeper.ZooDefs.OpCode;
import org.apache.zookeeper.common.Time;
import org.apache.zookeeper.server.Request;
import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;
+import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer;
import org.apache.zookeeper.server.util.SerializeUtils;
import org.apache.zookeeper.server.util.ZxidUtils;
import org.apache.zookeeper.txn.SetDataTxn;
@@ -71,9 +72,9 @@ public class Follower extends Learner{
self.end_fle = 0;
fzk.registerJMX(new FollowerBean(this, zk), self.jmxLocalPeerBean);
try {
- InetSocketAddress addr = findLeader();
+ QuorumServer leaderServer = findLeader();
try {
- connectToLeader(addr);
+ connectToLeader(leaderServer.addr, leaderServer.hostname);
long newEpochZxid = registerWithLeader(Leader.FOLLOWERINFO);
if (self.isReconfigStateChange())
throw new Exception("learned about role change");
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/75411ab3/src/java/main/org/apache/zookeeper/server/quorum/Leader.java
----------------------------------------------------------------------
diff --git a/src/java/main/org/apache/zookeeper/server/quorum/Leader.java b/src/java/main/org/apache/zookeeper/server/quorum/Leader.java
index cf2eb37..cebf2e3 100644
--- a/src/java/main/org/apache/zookeeper/server/quorum/Leader.java
+++ b/src/java/main/org/apache/zookeeper/server/quorum/Leader.java
@@ -19,6 +19,7 @@
package org.apache.zookeeper.server.quorum;
import java.io.ByteArrayOutputStream;
+import java.io.BufferedInputStream;
import java.io.IOException;
import java.net.BindException;
import java.net.ServerSocket;
@@ -39,6 +40,8 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;
+import javax.security.sasl.SaslException;
+
import org.apache.jute.BinaryOutputArchive;
import org.apache.zookeeper.ZooDefs.OpCode;
import org.apache.zookeeper.common.Time;
@@ -367,7 +370,10 @@ public class Leader {
// in LearnerHandler switch to the syncLimit
s.setSoTimeout(self.tickTime * self.initLimit);
s.setTcpNoDelay(nodelay);
- LearnerHandler fh = new LearnerHandler(s, Leader.this);
+
+ BufferedInputStream is = new BufferedInputStream(
+ s.getInputStream());
+ LearnerHandler fh = new LearnerHandler(s, is, Leader.this);
fh.start();
} catch (SocketException e) {
if (stop) {
@@ -381,6 +387,8 @@ public class Leader {
} else {
throw e;
}
+ } catch (SaslException e){
+ LOG.error("Exception while connecting to quorum learner", e);
}
}
} catch (Exception e) {
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/75411ab3/src/java/main/org/apache/zookeeper/server/quorum/Learner.java
----------------------------------------------------------------------
diff --git a/src/java/main/org/apache/zookeeper/server/quorum/Learner.java b/src/java/main/org/apache/zookeeper/server/quorum/Learner.java
index 726e688..a37cc93 100644
--- a/src/java/main/org/apache/zookeeper/server/quorum/Learner.java
+++ b/src/java/main/org/apache/zookeeper/server/quorum/Learner.java
@@ -50,6 +50,8 @@ import org.apache.zookeeper.server.util.SerializeUtils;
import org.apache.zookeeper.server.util.ZxidUtils;
import org.apache.zookeeper.txn.SetDataTxn;
import org.apache.zookeeper.txn.TxnHeader;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* This class is the superclass of two of the three main actors in a ZK
@@ -193,21 +195,21 @@ public class Learner {
/**
* Returns the address of the node we think is the leader.
*/
- protected InetSocketAddress findLeader() {
- InetSocketAddress addr = null;
+ protected QuorumServer findLeader() {
+ QuorumServer leaderServer = null;
// Find the leader by id
Vote current = self.getCurrentVote();
for (QuorumServer s : self.getView().values()) {
if (s.id == current.getId()) {
- addr = s.addr;
+ leaderServer = s;
break;
}
}
- if (addr == null) {
+ if (leaderServer == null) {
LOG.warn("Couldn't find the leader with id = "
+ current.getId());
}
- return addr;
+ return leaderServer;
}
/**
@@ -232,10 +234,11 @@ public class Learner {
* until either initLimit time has elapsed or 5 tries have happened.
* @param addr - the address of the Leader to connect to.
* @throws IOException - if the socket connection fails on the 5th attempt
+ * <li>if there is an authentication failure while connecting to leader</li>
* @throws ConnectException
* @throws InterruptedException
*/
- protected void connectToLeader(InetSocketAddress addr)
+ protected void connectToLeader(InetSocketAddress addr, String hostname)
throws IOException, ConnectException, InterruptedException {
sock = new Socket();
sock.setSoTimeout(self.tickTime * self.initLimit);
@@ -279,6 +282,9 @@ public class Learner {
}
Thread.sleep(1000);
}
+
+ self.authLearner.authenticate(sock, hostname);
+
leaderIs = BinaryInputArchive.getArchive(new BufferedInputStream(
sock.getInputStream()));
bufferedOutput = new BufferedOutputStream(sock.getOutputStream());
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/75411ab3/src/java/main/org/apache/zookeeper/server/quorum/LearnerHandler.java
----------------------------------------------------------------------
diff --git a/src/java/main/org/apache/zookeeper/server/quorum/LearnerHandler.java b/src/java/main/org/apache/zookeeper/server/quorum/LearnerHandler.java
index 1fbe708..7247f5c 100644
--- a/src/java/main/org/apache/zookeeper/server/quorum/LearnerHandler.java
+++ b/src/java/main/org/apache/zookeeper/server/quorum/LearnerHandler.java
@@ -33,6 +33,8 @@ import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock;
+import javax.security.sasl.SaslException;
+
import org.apache.jute.BinaryInputArchive;
import org.apache.jute.BinaryOutputArchive;
import org.apache.jute.Record;
@@ -155,6 +157,7 @@ public class LearnerHandler extends ZooKeeperThread {
private BinaryOutputArchive oa;
+ private final BufferedInputStream bufferedInput;
private BufferedOutputStream bufferedOutput;
/**
@@ -179,11 +182,27 @@ public class LearnerHandler extends ZooKeeperThread {
*/
private long leaderLastZxid;
- LearnerHandler(Socket sock, Leader leader) throws IOException {
+ LearnerHandler(Socket sock, BufferedInputStream bufferedInput,Leader leader) throws IOException {
super("LearnerHandler-" + sock.getRemoteSocketAddress());
this.sock = sock;
this.leader = leader;
- leader.addLearnerHandler(this);
+ this.bufferedInput = bufferedInput;
+
+ try {
+ if (leader.self != null) {
+ leader.self.authServer.authenticate(sock,
+ new DataInputStream(bufferedInput));
+ }
+ } catch (IOException e) {
+ LOG.error("Server failed to authenticate quorum learner, addr: {}, closing connection",
+ sock.getRemoteSocketAddress(), e);
+ try {
+ sock.close();
+ } catch (IOException ie) {
+ LOG.error("Exception while closing socket", ie);
+ }
+ throw new SaslException("Authentication failure: " + e.getMessage());
+ }
}
@Override
@@ -343,11 +362,11 @@ public class LearnerHandler extends ZooKeeperThread {
@Override
public void run() {
try {
+ leader.addLearnerHandler(this);
tickOfNextAckDeadline = leader.self.tick.get()
+ leader.self.initLimit + leader.self.syncLimit;
- ia = BinaryInputArchive.getArchive(new BufferedInputStream(sock
- .getInputStream()));
+ ia = BinaryInputArchive.getArchive(bufferedInput);
bufferedOutput = new BufferedOutputStream(sock.getOutputStream());
oa = BinaryOutputArchive.getArchive(bufferedOutput);
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/75411ab3/src/java/main/org/apache/zookeeper/server/quorum/Observer.java
----------------------------------------------------------------------
diff --git a/src/java/main/org/apache/zookeeper/server/quorum/Observer.java b/src/java/main/org/apache/zookeeper/server/quorum/Observer.java
index 27368c7..f0f724e 100644
--- a/src/java/main/org/apache/zookeeper/server/quorum/Observer.java
+++ b/src/java/main/org/apache/zookeeper/server/quorum/Observer.java
@@ -19,12 +19,12 @@
package org.apache.zookeeper.server.quorum;
import java.io.IOException;
-import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import org.apache.jute.Record;
import org.apache.zookeeper.server.ObserverBean;
import org.apache.zookeeper.server.Request;
+import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer;
import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;
import org.apache.zookeeper.server.util.SerializeUtils;
import org.apache.zookeeper.txn.SetDataTxn;
@@ -63,10 +63,10 @@ public class Observer extends Learner{
zk.registerJMX(new ObserverBean(this, zk), self.jmxLocalPeerBean);
try {
- InetSocketAddress addr = findLeader();
- LOG.info("Observing " + addr);
+ QuorumServer leaderServer = findLeader();
+ LOG.info("Observing " + leaderServer.addr);
try {
- connectToLeader(addr);
+ connectToLeader(leaderServer.addr, leaderServer.hostname);
long newLeaderZxid = registerWithLeader(Leader.OBSERVERINFO);
if (self.isReconfigStateChange())
throw new Exception("learned about role change");
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/75411ab3/src/java/main/org/apache/zookeeper/server/quorum/QuorumCnxManager.java
----------------------------------------------------------------------
diff --git a/src/java/main/org/apache/zookeeper/server/quorum/QuorumCnxManager.java b/src/java/main/org/apache/zookeeper/server/quorum/QuorumCnxManager.java
index c8f73a3..09da63a 100644
--- a/src/java/main/org/apache/zookeeper/server/quorum/QuorumCnxManager.java
+++ b/src/java/main/org/apache/zookeeper/server/quorum/QuorumCnxManager.java
@@ -18,6 +18,7 @@
package org.apache.zookeeper.server.quorum;
+import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
@@ -30,15 +31,24 @@ import java.net.SocketTimeoutException;
import java.nio.BufferUnderflowException;
import java.nio.ByteBuffer;
import java.nio.channels.UnresolvedAddressException;
+import java.util.Collections;
import java.util.Enumeration;
+import java.util.HashSet;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.NoSuchElementException;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
import org.apache.zookeeper.server.ZooKeeperThread;
+import org.apache.zookeeper.server.quorum.auth.QuorumAuthLearner;
+import org.apache.zookeeper.server.quorum.auth.QuorumAuthServer;
import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -79,8 +89,8 @@ public class QuorumCnxManager {
/*
* Negative counter for observer server ids.
*/
-
- private long observerCounter = -1;
+
+ private AtomicLong observerCounter = new AtomicLong(-1);
/*
* Protocol identifier used among peers
@@ -97,11 +107,26 @@ public class QuorumCnxManager {
*/
private int cnxTO = 5000;
-
+
+ final QuorumPeer self;
+
/*
* Local IP address
*/
- final QuorumPeer self;
+ final long mySid;
+ final int socketTimeout;
+ final Map<Long, QuorumPeer.QuorumServer> view;
+ final boolean listenOnAllIPs;
+ private ThreadPoolExecutor connectionExecutor;
+ private final Set<Long> inprogressConnections = Collections
+ .synchronizedSet(new HashSet<Long>());
+ private QuorumAuthServer authServer;
+ private QuorumAuthLearner authLearner;
+ private boolean quorumSaslAuthEnabled;
+ /*
+ * Counter to count connection processing threads.
+ */
+ private AtomicInteger connectionThreadCnt = new AtomicInteger(0);
/*
* Mapping from Peer to Thread number
@@ -217,7 +242,15 @@ public class QuorumCnxManager {
}
}
- public QuorumCnxManager(QuorumPeer self) {
+ public QuorumCnxManager(QuorumPeer self,
+ final long mySid,
+ Map<Long,QuorumPeer.QuorumServer> view,
+ QuorumAuthServer authServer,
+ QuorumAuthLearner authLearner,
+ int socketTimeout,
+ boolean listenOnAllIPs,
+ int quorumCnxnThreadsSize,
+ boolean quorumSaslAuthEnabled) {
this.recvQueue = new ArrayBlockingQueue<Message>(RECV_CAPACITY);
this.queueSendMap = new ConcurrentHashMap<Long, ArrayBlockingQueue<ByteBuffer>>();
this.senderWorkerMap = new ConcurrentHashMap<Long, SendWorker>();
@@ -230,11 +263,54 @@ public class QuorumCnxManager {
this.self = self;
- // Starts listener thread that waits for connection requests
+ this.mySid = mySid;
+ this.socketTimeout = socketTimeout;
+ this.view = view;
+ this.listenOnAllIPs = listenOnAllIPs;
+
+ initializeAuth(mySid, authServer, authLearner, quorumCnxnThreadsSize,
+ quorumSaslAuthEnabled);
+
+ // Starts listener thread that waits for connection requests
listener = new Listener();
listener.setName("QuorumPeerListener");
}
+ private void initializeAuth(final long mySid,
+ final QuorumAuthServer authServer,
+ final QuorumAuthLearner authLearner,
+ final int quorumCnxnThreadsSize,
+ final boolean quorumSaslAuthEnabled) {
+ this.authServer = authServer;
+ this.authLearner = authLearner;
+ this.quorumSaslAuthEnabled = quorumSaslAuthEnabled;
+ if (!this.quorumSaslAuthEnabled) {
+ LOG.debug("Not initializing connection executor as quorum sasl auth is disabled");
+ return;
+ }
+
+ // init connection executors
+ final AtomicInteger threadIndex = new AtomicInteger(1);
+ SecurityManager s = System.getSecurityManager();
+ final ThreadGroup group = (s != null) ? s.getThreadGroup()
+ : Thread.currentThread().getThreadGroup();
+ ThreadFactory daemonThFactory = new ThreadFactory() {
+
+ @Override
+ public Thread newThread(Runnable r) {
+ Thread t = new Thread(group, r, "QuorumConnectionThread-"
+ + "[myid=" + mySid + "]-"
+ + threadIndex.getAndIncrement());
+ return t;
+ }
+ };
+ this.connectionExecutor = new ThreadPoolExecutor(3,
+ quorumCnxnThreadsSize, 60, TimeUnit.SECONDS,
+ new SynchronousQueue<Runnable>(), daemonThFactory);
+ this.connectionExecutor.allowCoreThreadTimeOut(true);
+ }
+
+
/**
* Invokes initiateConnection for testing purposes
*
@@ -247,17 +323,80 @@ public class QuorumCnxManager {
sock.connect(self.getVotingView().get(sid).electionAddr, cnxTO);
initiateConnection(sock, sid);
}
-
+
/**
* If this server has initiated the connection, then it gives up on the
* connection if it loses challenge. Otherwise, it keeps the connection.
*/
- public boolean initiateConnection(Socket sock, Long sid) {
+ public void initiateConnection(final Socket sock, final Long sid) {
+ try {
+ startConnection(sock, sid);
+ } catch (IOException e) {
+ LOG.error("Exception while connecting, id: {}, addr: {}, closing learner connection",
+ new Object[] { sid, sock.getRemoteSocketAddress() }, e);
+ closeSocket(sock);
+ return;
+ }
+ }
+
+ /**
+ * Server will initiate the connection request to its peer server
+ * asynchronously via separate connection thread.
+ */
+ public void initiateConnectionAsync(final Socket sock, final Long sid) {
+ if(!inprogressConnections.add(sid)){
+ // simply return as there is a connection request to
+ // server 'sid' already in progress.
+ LOG.debug("Connection request to server id: {} is already in progress, so skipping this request",
+ sid);
+ closeSocket(sock);
+ return;
+ }
+ try {
+ connectionExecutor.execute(
+ new QuorumConnectionReqThread(sock, sid));
+ connectionThreadCnt.incrementAndGet();
+ } catch (Throwable e) {
+ // Imp: Safer side catching all type of exceptions and remove 'sid'
+ // from inprogress connections. This is to avoid blocking further
+ // connection requests from this 'sid' in case of errors.
+ inprogressConnections.remove(sid);
+ LOG.error("Exception while submitting quorum connection request", e);
+ closeSocket(sock);
+ }
+ }
+
+ /**
+ * Thread to send connection request to peer server.
+ */
+ private class QuorumConnectionReqThread extends ZooKeeperThread {
+ final Socket sock;
+ final Long sid;
+ QuorumConnectionReqThread(final Socket sock, final Long sid) {
+ super("QuorumConnectionReqThread-" + sid);
+ this.sock = sock;
+ this.sid = sid;
+ }
+
+ @Override
+ public void run() {
+ try{
+ initiateConnection(sock, sid);
+ } finally {
+ inprogressConnections.remove(sid);
+ }
+ }
+ }
+
+ private boolean startConnection(Socket sock, Long sid)
+ throws IOException {
+ DataOutputStream dout = null;
+ DataInputStream din = null;
try {
// Use BufferedOutputStream to reduce the number of IP packets. This is
// important for x-DC scenarios.
BufferedOutputStream buf = new BufferedOutputStream(sock.getOutputStream());
- DataOutputStream dout = new DataOutputStream(buf);
+ dout = new DataOutputStream(buf);
// Sending id and challenge
@@ -269,12 +408,22 @@ public class QuorumCnxManager {
dout.writeInt(addr_bytes.length);
dout.write(addr_bytes);
dout.flush();
+
+ din = new DataInputStream(
+ new BufferedInputStream(sock.getInputStream()));
} catch (IOException e) {
LOG.warn("Ignoring exception reading or writing challenge: ", e);
closeSocket(sock);
return false;
}
-
+
+ // authenticate learner
+ QuorumPeer.QuorumServer qps = self.getVotingView().get(sid);
+ if (qps != null) {
+ // TODO - investigate why reconfig makes qps null.
+ authLearner.authenticate(sock, qps.hostname);
+ }
+
// If lost the challenge, then drop the new connection
if (sid > self.getId()) {
LOG.info("Have smaller server identifier, so dropping the " +
@@ -283,7 +432,7 @@ public class QuorumCnxManager {
// Otherwise proceed with the connection
} else {
SendWorker sw = new SendWorker(sock, sid);
- RecvWorker rw = new RecvWorker(sock, sid, sw);
+ RecvWorker rw = new RecvWorker(sock, din, sid, sw);
sw.setRecv(rw);
SendWorker vsw = senderWorkerMap.get(sid);
@@ -312,13 +461,58 @@ public class QuorumCnxManager {
* possible long value to lose the challenge.
*
*/
- public void receiveConnection(Socket sock) {
+ public void receiveConnection(final Socket sock) {
+ DataInputStream din = null;
+ try {
+ din = new DataInputStream(
+ new BufferedInputStream(sock.getInputStream()));
+
+ handleConnection(sock, din);
+ } catch (IOException e) {
+ LOG.error("Exception handling connection, addr: {}, closing server connection",
+ sock.getRemoteSocketAddress());
+ closeSocket(sock);
+ }
+ }
+
+ /**
+ * Server receives a connection request and handles it asynchronously via
+ * separate thread.
+ */
+ public void receiveConnectionAsync(final Socket sock) {
+ try {
+ connectionExecutor.execute(
+ new QuorumConnectionReceiverThread(sock));
+ connectionThreadCnt.incrementAndGet();
+ } catch (Throwable e) {
+ LOG.error("Exception handling connection, addr: {}, closing server connection",
+ sock.getRemoteSocketAddress());
+ closeSocket(sock);
+ }
+ }
+
+ /**
+ * Thread to receive connection request from peer server.
+ */
+ private class QuorumConnectionReceiverThread extends ZooKeeperThread {
+ private final Socket sock;
+ QuorumConnectionReceiverThread(final Socket sock) {
+ super("QuorumConnectionReceiverThread-" + sock.getRemoteSocketAddress());
+ this.sock = sock;
+ }
+
+ @Override
+ public void run() {
+ receiveConnection(sock);
+ }
+ }
+
+ private void handleConnection(Socket sock, DataInputStream din)
+ throws IOException {
Long sid = null, protocolVersion = null;
InetSocketAddress electionAddr = null;
try {
- DataInputStream din = new DataInputStream(sock.getInputStream());
-
protocolVersion = din.readLong();
if (protocolVersion >= 0) { // this is a server id and not a protocol version
sid = protocolVersion;
@@ -339,16 +533,18 @@ public class QuorumCnxManager {
* Choose identifier at random. We need a value to identify
* the connection.
*/
-
- sid = observerCounter--;
- LOG.info("Setting arbitrary identifier to observer: {}", sid);
+ sid = observerCounter.getAndDecrement();
+ LOG.info("Setting arbitrary identifier to observer: " + sid);
}
} catch (IOException e) {
closeSocket(sock);
LOG.warn("Exception reading or writing challenge: {}", e.toString());
return;
}
-
+
+ // do authenticating learner
+ authServer.authenticate(sock, din);
+
//If wins the challenge, then close the new connection.
if (sid < self.getId()) {
/*
@@ -375,7 +571,7 @@ public class QuorumCnxManager {
} else { // Otherwise start worker threads to receive data.
SendWorker sw = new SendWorker(sock, sid);
- RecvWorker rw = new RecvWorker(sock, sid, sw);
+ RecvWorker rw = new RecvWorker(sock, din, sid, sw);
sw.setRecv(rw);
SendWorker vsw = senderWorkerMap.get(sid);
@@ -402,7 +598,7 @@ public class QuorumCnxManager {
/*
* If sending message to myself, then simply enqueue it (loopback).
*/
- if (self.getId() == sid) {
+ if (this.mySid == sid) {
b.position(0);
addToRecvQueue(new Message(b.duplicate(), sid));
/*
@@ -444,7 +640,15 @@ public class QuorumCnxManager {
setSockOpts(sock);
sock.connect(electionAddr, cnxTO);
LOG.debug("Connected to server " + sid);
- initiateConnection(sock, sid);
+ // Sends connection request asynchronously if the quorum
+ // sasl authentication is enabled. This is required because
+ // sasl server authentication process may take few seconds to
+ // finish, this may delay next peer connection requests.
+ if (quorumSaslAuthEnabled) {
+ initiateConnectionAsync(sock, sid);
+ } else {
+ initiateConnection(sock, sid);
+ }
return true;
} catch (UnresolvedAddressException e) {
// Sun doesn't include the address that causes this
@@ -547,6 +751,13 @@ public class QuorumCnxManager {
LOG.warn("Got interrupted before joining the listener", ex);
}
softHalt();
+
+ // clear data structures used for auth
+ if (connectionExecutor != null) {
+ connectionExecutor.shutdown();
+ }
+ inprogressConnections.clear();
+ resetConnectionThreadCount();
}
/**
@@ -595,11 +806,19 @@ public class QuorumCnxManager {
public long getThreadCount() {
return threadCnt.get();
}
+
+ /**
+ * Return number of connection processing threads.
+ */
+ public long getConnectionThreadCount() {
+ return connectionThreadCnt.get();
+ }
+
/**
- * Return reference to QuorumPeer
+ * Reset the value of connection processing threads count to zero.
*/
- public QuorumPeer getQuorumPeer() {
- return self;
+ private void resetConnectionThreadCount() {
+ connectionThreadCnt.set(0);
}
/**
@@ -645,7 +864,16 @@ public class QuorumCnxManager {
setSockOpts(client);
LOG.info("Received connection request "
+ client.getRemoteSocketAddress());
- receiveConnection(client);
+ // Receive and handle the connection request
+ // asynchronously if the quorum sasl authentication is
+ // enabled. This is required because sasl server
+ // authentication process may take few seconds to finish,
+ // this may delay next peer connection requests.
+ if (quorumSaslAuthEnabled) {
+ receiveConnectionAsync(client);
+ } else {
+ receiveConnection(client);
+ }
numRetries = 0;
} catch (SocketTimeoutException e) {
LOG.warn("The socket is listening for the election accepted "
@@ -695,7 +923,8 @@ public class QuorumCnxManager {
try{
LOG.debug("Trying to close listener: " + ss);
if(ss != null) {
- LOG.debug("Closing listener: " + self.getId());
+ LOG.debug("Closing listener: "
+ + QuorumCnxManager.this.mySid);
ss.close();
}
} catch (IOException e){
@@ -847,8 +1076,9 @@ public class QuorumCnxManager {
}
}
} catch (Exception e) {
- LOG.warn("Exception when using channel: for id " + sid + " my id = " +
- self.getId() + " error = " + e);
+ LOG.warn("Exception when using channel: for id " + sid
+ + " my id = " + QuorumCnxManager.this.mySid
+ + " error = " + e);
}
this.finish();
LOG.warn("Send worker leaving thread " + " id " + sid + " my id = " + self.getId());
@@ -863,16 +1093,16 @@ public class QuorumCnxManager {
Long sid;
Socket sock;
volatile boolean running = true;
- DataInputStream din;
+ final DataInputStream din;
final SendWorker sw;
- RecvWorker(Socket sock, Long sid, SendWorker sw) {
+ RecvWorker(Socket sock, DataInputStream din, Long sid, SendWorker sw) {
super("RecvWorker:" + sid);
this.sid = sid;
this.sock = sock;
this.sw = sw;
+ this.din = din;
try {
- din = new DataInputStream(sock.getInputStream());
// OK to wait until socket disconnects while reading.
sock.setSoTimeout(0);
} catch (IOException e) {
@@ -925,8 +1155,8 @@ public class QuorumCnxManager {
addToRecvQueue(new Message(message.duplicate(), sid));
}
} catch (Exception e) {
- LOG.warn("Connection broken for id " + sid + ", my id = " +
- self.getId() + ", error = " , e);
+ LOG.warn("Connection broken for id " + sid + ", my id = "
+ + QuorumCnxManager.this.mySid + ", error = " , e);
} finally {
LOG.warn("Interrupting SendWorker");
sw.finish();
@@ -1046,4 +1276,8 @@ public class QuorumCnxManager {
throws InterruptedException {
return recvQueue.poll(timeout, unit);
}
+
+ public boolean connectedToPeer(long peerSid) {
+ return senderWorkerMap.get(peerSid) != null;
+ }
}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/75411ab3/src/java/main/org/apache/zookeeper/server/quorum/QuorumPeer.java
----------------------------------------------------------------------
diff --git a/src/java/main/org/apache/zookeeper/server/quorum/QuorumPeer.java b/src/java/main/org/apache/zookeeper/server/quorum/QuorumPeer.java
index 01e5947..45f32b9 100644
--- a/src/java/main/org/apache/zookeeper/server/quorum/QuorumPeer.java
+++ b/src/java/main/org/apache/zookeeper/server/quorum/QuorumPeer.java
@@ -42,6 +42,8 @@ import java.util.Properties;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
+import javax.security.sasl.SaslException;
+
import org.apache.zookeeper.KeeperException.BadArgumentsException;
import org.apache.zookeeper.common.AtomicFileWritingIdiom;
import org.apache.zookeeper.common.AtomicFileWritingIdiom.WriterStatement;
@@ -52,6 +54,13 @@ import org.apache.zookeeper.server.ServerCnxnFactory;
import org.apache.zookeeper.server.ZKDatabase;
import org.apache.zookeeper.server.ZooKeeperServer;
import org.apache.zookeeper.server.ZooKeeperThread;
+import org.apache.zookeeper.server.quorum.auth.QuorumAuth;
+import org.apache.zookeeper.server.quorum.auth.QuorumAuthLearner;
+import org.apache.zookeeper.server.quorum.auth.QuorumAuthServer;
+import org.apache.zookeeper.server.quorum.auth.SaslQuorumAuthLearner;
+import org.apache.zookeeper.server.quorum.auth.SaslQuorumAuthServer;
+import org.apache.zookeeper.server.quorum.auth.NullQuorumAuthLearner;
+import org.apache.zookeeper.server.quorum.auth.NullQuorumAuthServer;
import org.apache.zookeeper.server.admin.AdminServer;
import org.apache.zookeeper.server.admin.AdminServer.AdminServerException;
import org.apache.zookeeper.server.admin.AdminServerFactory;
@@ -98,6 +107,8 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
private Map<Long, RemotePeerBean> jmxRemotePeerBean;
LeaderElectionBean jmxLeaderElectionBean;
private QuorumCnxManager qcm;
+ QuorumAuthServer authServer;
+ QuorumAuthLearner authLearner;
/**
* ZKDatabase is a top level member of quorumpeer
@@ -116,6 +127,8 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
public InetSocketAddress clientAddr = null;
public long id;
+
+ public String hostname;
public LearnerType type = LearnerType.PARTICIPANT;
@@ -131,6 +144,7 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
this(id, addr, electionAddr, (InetSocketAddress)null, LearnerType.PARTICIPANT);
}
+ // VisibleForTesting
public QuorumServer(long id, InetSocketAddress addr) {
this(id, addr, (InetSocketAddress)null, (InetSocketAddress)null, LearnerType.PARTICIPANT);
}
@@ -217,7 +231,7 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
}
// is client_config a host:port or just a port
- String hostname = (clientParts.length == 2) ? clientParts[0] : "0.0.0.0";
+ hostname = (clientParts.length == 2) ? clientParts[0] : "0.0.0.0";
try {
clientAddr = new InetSocketAddress(hostname,
Integer.parseInt(clientParts[clientParts.length - 1]));
@@ -245,6 +259,8 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
setType(serverParts[3]);
}
+ this.hostname = serverParts[0];
+
setMyAddrs();
}
@@ -529,6 +545,50 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
private long electionTimeTaken = -1;
/**
+ * Enable/Disables quorum authentication using sasl. Defaulting to false.
+ */
+ protected boolean quorumSaslEnableAuth;
+
+ /**
+ * If this is false, quorum peer server will accept another quorum peer client
+ * connection even if the authentication did not succeed. This can be used while
+ * upgrading ZooKeeper server. Defaulting to false (required).
+ */
+ protected boolean quorumServerSaslAuthRequired;
+
+ /**
+ * If this is false, quorum peer learner will talk to quorum peer server
+ * without authentication. This can be used while upgrading ZooKeeper
+ * server. Defaulting to false (required).
+ */
+ protected boolean quorumLearnerSaslAuthRequired;
+
+ /**
+ * Kerberos quorum service principal. Defaulting to 'zkquorum/localhost'.
+ */
+ protected String quorumServicePrincipal;
+
+ /**
+ * Quorum learner login context name in jaas-conf file to read the kerberos
+ * security details. Defaulting to 'QuorumLearner'.
+ */
+ protected String quorumLearnerLoginContext;
+
+ /**
+ * Quorum server login context name in jaas-conf file to read the kerberos
+ * security details. Defaulting to 'QuorumServer'.
+ */
+ protected String quorumServerLoginContext;
+
+ // TODO: need to tune the default value of thread size
+ private static final int QUORUM_CNXN_THREADS_SIZE_DEFAULT_VALUE = 20;
+ /**
+ * The maximum number of threads to allow in the connectionExecutors thread
+ * pool which will be used to initiate quorum server connections.
+ */
+ protected int quorumCnxnThreadsSize = QUORUM_CNXN_THREADS_SIZE_DEFAULT_VALUE;
+
+ /**
* @deprecated As of release 3.4.0, this class has been deprecated, since
* it is used with one of the udp-based versions of leader election, which
* we are also deprecating.
@@ -717,14 +777,18 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
AdminServer adminServer;
- public QuorumPeer() {
+ public static QuorumPeer testingQuorumPeer() throws SaslException {
+ return new QuorumPeer();
+ }
+
+ public QuorumPeer() throws SaslException {
super("QuorumPeer");
quorumStats = new QuorumStats(this);
jmxRemotePeerBean = new HashMap<Long, RemotePeerBean>();
adminServer = AdminServerFactory.createAdminServer();
+ initialize();
}
-
/**
* For backward compatibility purposes, we instantiate QuorumMaj by default.
*/
@@ -759,6 +823,23 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
adminServer = AdminServerFactory.createAdminServer();
}
+ public void initialize() throws SaslException {
+ // init quorum auth server & learner
+ if (isQuorumSaslAuthEnabled()) {
+ Set<String> authzHosts = new HashSet<String>();
+ for (QuorumServer qs : getView().values()) {
+ authzHosts.add(qs.hostname);
+ }
+ authServer = new SaslQuorumAuthServer(isQuorumServerSaslAuthRequired(),
+ quorumServerLoginContext, authzHosts);
+ authLearner = new SaslQuorumAuthLearner(isQuorumLearnerSaslAuthRequired(),
+ quorumServicePrincipal, quorumLearnerLoginContext);
+ } else {
+ authServer = new NullQuorumAuthServer();
+ authLearner = new NullQuorumAuthLearner();
+ }
+ }
+
QuorumStats quorumStats() {
return quorumStats;
}
@@ -939,26 +1020,26 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
//TODO: use a factory rather than a switch
switch (electionAlgorithm) {
- case 1:
- le = new AuthFastLeaderElection(this);
- break;
- case 2:
- le = new AuthFastLeaderElection(this, true);
- break;
- case 3:
- qcm = new QuorumCnxManager(this);
- QuorumCnxManager.Listener listener = qcm.listener;
- if(listener != null){
- listener.start();
- FastLeaderElection fle = new FastLeaderElection(this, qcm);
- fle.start();
- le = fle;
- } else {
- LOG.error("Null listener when initializing cnx manager");
- }
- break;
- default:
- assert false;
+ case 1:
+ le = new AuthFastLeaderElection(this);
+ break;
+ case 2:
+ le = new AuthFastLeaderElection(this, true);
+ break;
+ case 3:
+ qcm = createCnxnManager();
+ QuorumCnxManager.Listener listener = qcm.listener;
+ if(listener != null){
+ listener.start();
+ FastLeaderElection fle = new FastLeaderElection(this, qcm);
+ fle.start();
+ le = fle;
+ } else {
+ LOG.error("Null listener when initializing cnx manager");
+ }
+ break;
+ default:
+ assert false;
}
return le;
}
@@ -1921,4 +2002,75 @@ public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider
long getElectionTimeTaken() {
return electionTimeTaken;
}
+
+ void setQuorumServerSaslRequired(boolean serverSaslRequired) {
+ quorumServerSaslAuthRequired = serverSaslRequired;
+ LOG.info("{} set to {}", QuorumAuth.QUORUM_SERVER_SASL_AUTH_REQUIRED,
+ serverSaslRequired);
+ }
+
+ void setQuorumLearnerSaslRequired(boolean learnerSaslRequired) {
+ quorumLearnerSaslAuthRequired = learnerSaslRequired;
+ LOG.info("{} set to {}", QuorumAuth.QUORUM_LEARNER_SASL_AUTH_REQUIRED,
+ learnerSaslRequired);
+ }
+
+ void setQuorumSaslEnabled(boolean enableAuth) {
+ quorumSaslEnableAuth = enableAuth;
+ if (!quorumSaslEnableAuth) {
+ LOG.info("QuorumPeer communication is not secured!");
+ } else {
+ LOG.info("{} set to {}",
+ QuorumAuth.QUORUM_SASL_AUTH_ENABLED, enableAuth);
+ }
+ }
+
+ void setQuorumServicePrincipal(String servicePrincipal) {
+ quorumServicePrincipal = servicePrincipal;
+ LOG.info("{} set to {}", QuorumAuth.QUORUM_KERBEROS_SERVICE_PRINCIPAL,
+ quorumServicePrincipal);
+ }
+
+ void setQuorumLearnerLoginContext(String learnerContext) {
+ quorumLearnerLoginContext = learnerContext;
+ LOG.info("{} set to {}", QuorumAuth.QUORUM_LEARNER_SASL_LOGIN_CONTEXT,
+ quorumLearnerLoginContext);
+ }
+
+ void setQuorumServerLoginContext(String serverContext) {
+ quorumServerLoginContext = serverContext;
+ LOG.info("{} set to {}", QuorumAuth.QUORUM_SERVER_SASL_LOGIN_CONTEXT,
+ quorumServerLoginContext);
+ }
+
+ void setQuorumCnxnThreadsSize(int qCnxnThreadsSize) {
+ if (qCnxnThreadsSize > QUORUM_CNXN_THREADS_SIZE_DEFAULT_VALUE) {
+ quorumCnxnThreadsSize = qCnxnThreadsSize;
+ }
+ LOG.info("quorum.cnxn.threads.size set to {}", quorumCnxnThreadsSize);
+ }
+
+ boolean isQuorumSaslAuthEnabled() {
+ return quorumSaslEnableAuth;
+ }
+
+ private boolean isQuorumServerSaslAuthRequired() {
+ return quorumServerSaslAuthRequired;
+ }
+
+ private boolean isQuorumLearnerSaslAuthRequired() {
+ return quorumLearnerSaslAuthRequired;
+ }
+
+ public QuorumCnxManager createCnxnManager() {
+ return new QuorumCnxManager(this,
+ this.getId(),
+ this.getView(),
+ this.authServer,
+ this.authLearner,
+ this.tickTime * this.syncLimit,
+ this.getQuorumListenOnAllIPs(),
+ this.quorumCnxnThreadsSize,
+ this.isQuorumSaslAuthEnabled());
+ }
}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/75411ab3/src/java/main/org/apache/zookeeper/server/quorum/QuorumPeerConfig.java
----------------------------------------------------------------------
diff --git a/src/java/main/org/apache/zookeeper/server/quorum/QuorumPeerConfig.java b/src/java/main/org/apache/zookeeper/server/quorum/QuorumPeerConfig.java
index 7afbedd..7159139 100644
--- a/src/java/main/org/apache/zookeeper/server/quorum/QuorumPeerConfig.java
+++ b/src/java/main/org/apache/zookeeper/server/quorum/QuorumPeerConfig.java
@@ -49,6 +49,7 @@ import org.apache.zookeeper.common.PathUtils;
import org.apache.zookeeper.server.ZooKeeperServer;
import org.apache.zookeeper.server.quorum.QuorumPeer.LearnerType;
import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer;
+import org.apache.zookeeper.server.quorum.auth.QuorumAuth;
import org.apache.zookeeper.server.quorum.flexible.QuorumHierarchical;
import org.apache.zookeeper.server.quorum.flexible.QuorumMaj;
import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;
@@ -94,6 +95,17 @@ public class QuorumPeerConfig {
protected LearnerType peerType = LearnerType.PARTICIPANT;
/**
+ * Configurations for the quorumpeer-to-quorumpeer sasl authentication
+ */
+ protected boolean quorumServerRequireSasl = false;
+ protected boolean quorumLearnerRequireSasl = false;
+ protected boolean quorumEnableSasl = false;
+ protected String quorumServicePrincipal = QuorumAuth.QUORUM_KERBEROS_SERVICE_PRINCIPAL_DEFAULT_VALUE;
+ protected String quorumLearnerLoginContext = QuorumAuth.QUORUM_LEARNER_SASL_LOGIN_CONTEXT_DFAULT_VALUE;
+ protected String quorumServerLoginContext = QuorumAuth.QUORUM_SERVER_SASL_LOGIN_CONTEXT_DFAULT_VALUE;
+ protected int quorumCnxnThreadsSize;
+
+ /**
* Minimum snapshot retain count.
* @see org.apache.zookeeper.server.PurgeTxnLog#purge(File, File, int)
*/
@@ -296,11 +308,47 @@ public class QuorumPeerConfig {
}
} else if ((key.startsWith("server.") || key.startsWith("group") || key.startsWith("weight")) && zkProp.containsKey("dynamicConfigFile")) {
throw new ConfigException("parameter: " + key + " must be in a separate dynamic config file");
+ } else if (key.equals(QuorumAuth.QUORUM_SASL_AUTH_ENABLED)) {
+ quorumEnableSasl = Boolean.parseBoolean(value);
+ } else if (key.equals(QuorumAuth.QUORUM_SERVER_SASL_AUTH_REQUIRED)) {
+ quorumServerRequireSasl = Boolean.parseBoolean(value);
+ } else if (key.equals(QuorumAuth.QUORUM_LEARNER_SASL_AUTH_REQUIRED)) {
+ quorumLearnerRequireSasl = Boolean.parseBoolean(value);
+ } else if (key.equals(QuorumAuth.QUORUM_LEARNER_SASL_LOGIN_CONTEXT)) {
+ quorumLearnerLoginContext = value;
+ } else if (key.equals(QuorumAuth.QUORUM_SERVER_SASL_LOGIN_CONTEXT)) {
+ quorumServerLoginContext = value;
+ } else if (key.equals(QuorumAuth.QUORUM_KERBEROS_SERVICE_PRINCIPAL)) {
+ quorumServicePrincipal = value;
+ } else if (key.equals("quorum.cnxn.threads.size")) {
+ quorumCnxnThreadsSize = Integer.parseInt(value);
} else {
System.setProperty("zookeeper." + key, value);
}
}
+ if (!quorumEnableSasl && quorumServerRequireSasl) {
+ throw new IllegalArgumentException(
+ QuorumAuth.QUORUM_SASL_AUTH_ENABLED
+ + " is disabled, so cannot enable "
+ + QuorumAuth.QUORUM_SERVER_SASL_AUTH_REQUIRED);
+ }
+ if (!quorumEnableSasl && quorumLearnerRequireSasl) {
+ throw new IllegalArgumentException(
+ QuorumAuth.QUORUM_SASL_AUTH_ENABLED
+ + " is disabled, so cannot enable "
+ + QuorumAuth.QUORUM_LEARNER_SASL_AUTH_REQUIRED);
+ }
+ // If quorumpeer learner is not auth enabled then self won't be able to
+ // join quorum. So this condition is ensuring that the quorumpeer learner
+ // is also auth enabled while enabling quorum server require sasl.
+ if (!quorumLearnerRequireSasl && quorumServerRequireSasl) {
+ throw new IllegalArgumentException(
+ QuorumAuth.QUORUM_LEARNER_SASL_AUTH_REQUIRED
+ + " is disabled, so cannot enable "
+ + QuorumAuth.QUORUM_SERVER_SASL_AUTH_REQUIRED);
+ }
+
// Reset to MIN_SNAP_RETAIN_COUNT if invalid (less than 3)
// PurgeTxnLog.purge(File, File, int) will not allow to purge less
// than 3.