ZOOKEEPER-3156: Add in option to canonicalize host name

This is the master and 3.5 version of #648.  It should apply cleanly to both lines, but if you want a separate pull request for each I am happy to do it.

Author: Robert Evans <evans@yahoo-inc.com>

Reviewers: fangmin@apache.org, andor@apache.org

Closes #652 from revans2/ZOOKEEPER-3156

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

@@ -793,7 +793,7 @@ public class ClientCnxn {
             super(msg);
         }
     }
-    
+
     /**
      * This class services the outgoing request queue and generates the heart
      * beats. It also spawns the ReadThread.
@@ -1080,7 +1080,8 @@ public class ClientCnxn {
                     if (zooKeeperSaslClient != null) {
                         zooKeeperSaslClient.shutdown();
                     }
-                    zooKeeperSaslClient = new ZooKeeperSaslClient(getServerPrincipal(addr), clientConfig);
+                    zooKeeperSaslClient = new ZooKeeperSaslClient(SaslServerPrincipal.getServerPrincipal(addr, clientConfig),
+                        clientConfig);
                 } catch (LoginException e) {
                     // An authentication error occurred when the SASL client tried to initialize:
                     // for Kerberos this means that the client failed to authenticate with the KDC.
@@ -1099,13 +1100,6 @@ public class ClientCnxn {
             clientCnxnSocket.connect(addr);
         }
 
-        private String getServerPrincipal(InetSocketAddress addr) {
-            String principalUserName = clientConfig.getProperty(ZKClientConfig.ZK_SASL_CLIENT_USERNAME,
-                    ZKClientConfig.ZK_SASL_CLIENT_USERNAME_DEFAULT);
-            String serverPrincipal = principalUserName + "/" + addr.getHostString();
-            return serverPrincipal;
-        }
-
         private void logStartConnect(InetSocketAddress addr) {
             String msg = "Opening socket connection to server " + addr;
             if (zooKeeperSaslClient != null) {

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

@@ -0,0 +1,132 @@
+/**
+ * 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 java.net.InetAddress;
+import java.net.InetSocketAddress;
+import org.apache.zookeeper.client.ZKClientConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Computes the Server Principal for a SASL client.
+ */
+public class SaslServerPrincipal {
+    private static final Logger LOG = LoggerFactory.getLogger(SaslServerPrincipal.class);
+
+    /**
+     * Get the name of the server principal for a SASL client.
+     * @param addr the address of the host.
+     * @param clientConfig the configuration for the client.
+     * @return the name of the principal.
+     */
+    static String getServerPrincipal(InetSocketAddress addr, ZKClientConfig clientConfig) {
+        return getServerPrincipal(new WrapperInetSocketAddress(addr), clientConfig);
+    }
+
+    /**
+     * Get the name of the server principal for a SASL client.  This is visible for testing purposes.
+     * @param addr the address of the host.
+     * @param clientConfig the configuration for the client.
+     * @return the name of the principal.
+     */
+    static String getServerPrincipal(WrapperInetSocketAddress addr, ZKClientConfig clientConfig) {
+        String principalUserName = clientConfig.getProperty(ZKClientConfig.ZK_SASL_CLIENT_USERNAME,
+            ZKClientConfig.ZK_SASL_CLIENT_USERNAME_DEFAULT);
+        String hostName = addr.getHostName();
+
+        boolean canonicalize = true;
+        String canonicalizeText = clientConfig.getProperty(ZKClientConfig.ZK_SASL_CLIENT_CANONICALIZE_HOSTNAME,
+            ZKClientConfig.ZK_SASL_CLIENT_CANONICALIZE_HOSTNAME_DEFAULT);
+        try {
+            canonicalize = Boolean.parseBoolean(canonicalizeText);
+        } catch (IllegalArgumentException ea) {
+            LOG.warn("Could not parse config {} \"{}\" into a boolean using default {}", ZKClientConfig
+                .ZK_SASL_CLIENT_CANONICALIZE_HOSTNAME, canonicalizeText, canonicalize);
+        }
+
+        if (canonicalize) {
+            WrapperInetAddress ia = addr.getAddress();
+            if (ia == null) {
+                throw new IllegalArgumentException("Unable to canonicalize address " + addr + " because it's not resolvable");
+            }
+
+            String canonicalHostName = ia.getCanonicalHostName();
+            //avoid using literal IP address when security check fails
+            if (!canonicalHostName.equals(ia.getHostAddress())) {
+                hostName = canonicalHostName;
+            }
+            if (LOG.isDebugEnabled()) {
+                LOG.debug("Canonicalized address to {}", hostName);
+            }
+        }
+        String serverPrincipal = principalUserName + "/" + hostName;
+        return serverPrincipal;
+    }
+
+    /**
+     * This is here to provide a way to unit test the core logic as the methods for
+     * InetSocketAddress are marked as final.
+     */
+    static class WrapperInetSocketAddress {
+        private final InetSocketAddress addr;
+
+        WrapperInetSocketAddress(InetSocketAddress addr) {
+            this.addr = addr;
+        }
+
+        public String getHostName() {
+            return addr.getHostName();
+        }
+
+        public WrapperInetAddress getAddress() {
+            InetAddress ia = addr.getAddress();
+            return ia == null ? null : new WrapperInetAddress(ia);
+        }
+
+        @Override
+        public String toString() {
+            return addr.toString();
+        }
+    }
+
+    /**
+     * This is here to provide a way to unit test the core logic as the methods for
+     * InetAddress are marked as final.
+     */
+    static class WrapperInetAddress {
+        private final InetAddress ia;
+
+        WrapperInetAddress(InetAddress ia) {
+            this.ia = ia;
+        }
+
+        public String getCanonicalHostName() {
+            return ia.getCanonicalHostName();
+        }
+
+        public String getHostAddress() {
+            return ia.getHostAddress();
+        }
+
+        @Override
+        public String toString() {
+            return ia.toString();
+        }
+    }
+}

zookeeper-server/src/main/java/org/apache/zookeeper/client/ZKClientConfig.java

@@ -33,6 +33,9 @@ import org.apache.zookeeper.server.quorum.QuorumPeerConfig.ConfigException;
 public class ZKClientConfig extends ZKConfig {
     public static final String ZK_SASL_CLIENT_USERNAME = "zookeeper.sasl.client.username";
     public static final String ZK_SASL_CLIENT_USERNAME_DEFAULT = "zookeeper";
+    public static final String ZK_SASL_CLIENT_CANONICALIZE_HOSTNAME =
+        "zookeeper.sasl.client.canonicalize.hostname";
+    public static final String ZK_SASL_CLIENT_CANONICALIZE_HOSTNAME_DEFAULT = "true";
     @SuppressWarnings("deprecation")
     public static final String LOGIN_CONTEXT_NAME_KEY = ZooKeeperSaslClient.LOGIN_CONTEXT_NAME_KEY;;
     public static final String LOGIN_CONTEXT_NAME_KEY_DEFAULT = "Client";

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

@@ -0,0 +1,76 @@
+/**
+ * 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 java.io.IOException;
+import org.apache.zookeeper.client.ZKClientConfig;
+import org.junit.Assert;
+import org.junit.Test;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class ClientCanonicalizeTest extends ZKTestCase {
+    @Test
+    public void testClientCanonicalization() throws IOException, InterruptedException {
+        SaslServerPrincipal.WrapperInetSocketAddress addr = mock(SaslServerPrincipal.WrapperInetSocketAddress.class);
+        SaslServerPrincipal.WrapperInetAddress ia = mock(SaslServerPrincipal.WrapperInetAddress.class);
+
+        when(addr.getHostName()).thenReturn("zookeeper.apache.org");
+        when(addr.getAddress()).thenReturn(ia);
+        when(ia.getCanonicalHostName()).thenReturn("zk1.apache.org");
+        when(ia.getHostAddress()).thenReturn("127.0.0.1");
+
+        ZKClientConfig conf = new ZKClientConfig();
+        String principal = SaslServerPrincipal.getServerPrincipal(addr, conf);
+        Assert.assertEquals("The computed principal does not appear to have been canonicalized", "zookeeper/zk1.apache.org", principal);
+    }
+
+    @Test
+    public void testClientNoCanonicalization() throws IOException, InterruptedException {
+        SaslServerPrincipal.WrapperInetSocketAddress addr = mock(SaslServerPrincipal.WrapperInetSocketAddress.class);
+        SaslServerPrincipal.WrapperInetAddress ia = mock(SaslServerPrincipal.WrapperInetAddress.class);
+
+        when(addr.getHostName()).thenReturn("zookeeper.apache.org");
+        when(addr.getAddress()).thenReturn(ia);
+        when(ia.getCanonicalHostName()).thenReturn("zk1.apache.org");
+        when(ia.getHostAddress()).thenReturn("127.0.0.1");
+
+        ZKClientConfig conf = new ZKClientConfig();
+        conf.setProperty(ZKClientConfig.ZK_SASL_CLIENT_CANONICALIZE_HOSTNAME, "false");
+        String principal = SaslServerPrincipal.getServerPrincipal(addr, conf);
+        Assert.assertEquals("The computed principal does appears to have been canonicalized incorrectly", "zookeeper/zookeeper.apache.org",
+            principal);
+    }
+
+    @Test
+    public void testClientCanonicalizationToIp() throws IOException, InterruptedException {
+        SaslServerPrincipal.WrapperInetSocketAddress addr = mock(SaslServerPrincipal.WrapperInetSocketAddress.class);
+        SaslServerPrincipal.WrapperInetAddress ia = mock(SaslServerPrincipal.WrapperInetAddress.class);
+
+        when(addr.getHostName()).thenReturn("zookeeper.apache.org");
+        when(addr.getAddress()).thenReturn(ia);
+        when(ia.getCanonicalHostName()).thenReturn("127.0.0.1");
+        when(ia.getHostAddress()).thenReturn("127.0.0.1");
+
+        ZKClientConfig conf = new ZKClientConfig();
+        String principal = SaslServerPrincipal.getServerPrincipal(addr, conf);
+        Assert.assertEquals("The computed principal does appear to have falled back to the original host name",
+            "zookeeper/zookeeper.apache.org", principal);
+    }
+}