ZOOKEEPER-2858: Disable reverse DNS lookup for SASL java client

Author: anmolnar
Closes #2331 from anmolnar/ZOOKEEPER-2858

Author: anmolnar

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

@@ -1315,6 +1315,11 @@ and [SASL authentication for ZooKeeper](https://cwiki.apache.org/confluence/disp
     the fully qualified domain name belonging to the address. You can disable this 'canonicalization'
     by setting: zookeeper.sasl.client.canonicalize.hostname=false
 
+* *zookeeper.sasl.client.allowReverseDnsLookup* :
+    **New in 3.9.5:**
+    Controls whether reverse DNS lookup is enabled when constructing the server principal for the SASL client.
+    Default: false
+
 * *zookeeper.server.realm* :
     Realm part of the server principal. By default it is the client principal realm.
 

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

@@ -38,7 +38,7 @@ public class SaslServerPrincipal {
      * @return the name of the principal.
      */
     static String getServerPrincipal(InetSocketAddress addr, ZKClientConfig clientConfig) {
-        return getServerPrincipal(new WrapperInetSocketAddress(addr), clientConfig);
+        return getServerPrincipal(new WrapperInetSocketAddress(addr, clientConfig), clientConfig);
     }
 
     /**
@@ -96,13 +96,20 @@ static String getServerPrincipal(WrapperInetSocketAddress addr, ZKClientConfig c
     static class WrapperInetSocketAddress {
 
         private final InetSocketAddress addr;
+        private final ZKClientConfig clientConfig;
 
-        WrapperInetSocketAddress(InetSocketAddress addr) {
+        WrapperInetSocketAddress(InetSocketAddress addr, ZKClientConfig clientConfig) {
             this.addr = addr;
+            this.clientConfig = clientConfig;
         }
 
         public String getHostName() {
-            return addr.getHostName();
+            if (clientConfig.getBoolean(ZKClientConfig.ZK_SASL_CLIENT_ALLOW_REVERSE_DNS,
+                                        ZKClientConfig.ZK_SASL_CLIENT_ALLOW_REVERSE_DNS_DEFAULT)) {
+                return addr.getHostName();
+            } else {
+                return addr.getHostString();
+            }
         }
 
         public WrapperInetAddress getAddress() {

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

@@ -60,6 +60,8 @@ public class ZKClientConfig extends ZKConfig {
      * Feature is disabled by default.
      */
     public static final long ZOOKEEPER_REQUEST_TIMEOUT_DEFAULT = 0;
+    public static final String ZK_SASL_CLIENT_ALLOW_REVERSE_DNS = "zookeeper.sasl.client.allowReverseDnsLookup";
+    public static final boolean ZK_SASL_CLIENT_ALLOW_REVERSE_DNS_DEFAULT = false;
 
     public ZKClientConfig() {
         super();
@@ -120,6 +122,7 @@ protected void handleBackwardCompatibility() {
         setProperty(DISABLE_AUTO_WATCH_RESET, System.getProperty(DISABLE_AUTO_WATCH_RESET));
         setProperty(ZOOKEEPER_CLIENT_CNXN_SOCKET, System.getProperty(ZOOKEEPER_CLIENT_CNXN_SOCKET));
         setProperty(SECURE_CLIENT, System.getProperty(SECURE_CLIENT));
+        setProperty(ZK_SASL_CLIENT_ALLOW_REVERSE_DNS, System.getProperty(ZK_SASL_CLIENT_ALLOW_REVERSE_DNS));
     }
 
     /**

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

@@ -19,6 +19,8 @@
 package org.apache.zookeeper;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 import java.io.IOException;
@@ -85,4 +87,39 @@ public void testGetServerPrincipalReturnConfiguredPrincipalName() {
         assertEquals(configuredPrincipal, serverPrincipal);
     }
 
+    @Test
+    public void testAllowReverseDnsLookupDisabled() {
+        // Arrange
+        ZKClientConfig config = new ZKClientConfig();
+        config.setProperty(ZKClientConfig.ZK_SASL_CLIENT_ALLOW_REVERSE_DNS, "false");
+        InetSocketAddress addr = mock(InetSocketAddress.class);
+        SaslServerPrincipal.WrapperInetSocketAddress ia = new SaslServerPrincipal.WrapperInetSocketAddress(addr, config);
+        doReturn("this-is-the-right-hostname").when(addr).getHostString();
+        doThrow(new UnsupportedOperationException("getHostName() should not be called when reverse DNS is disabled"))
+            .when(addr).getHostName();
+
+        // Act
+        String hostname = ia.getHostName();
+
+        // Assert
+        assertEquals("this-is-the-right-hostname", hostname);
+    }
+
+    @Test
+    public void testAllowReverseDnsLookupEnabled() {
+        // Arrange
+        ZKClientConfig config = new ZKClientConfig();
+        config.setProperty(ZKClientConfig.ZK_SASL_CLIENT_ALLOW_REVERSE_DNS, "true");
+        InetSocketAddress addr = mock(InetSocketAddress.class);
+        SaslServerPrincipal.WrapperInetSocketAddress ia = new SaslServerPrincipal.WrapperInetSocketAddress(addr, config);
+        doReturn("this-is-the-right-hostname").when(addr).getHostName();
+        doThrow(new UnsupportedOperationException("getHostString() should not be called when reverse DNS is enabled"))
+            .when(addr).getHostString();
+
+        // Act
+        String hostname = ia.getHostName();
+
+        // Assert
+        assertEquals("this-is-the-right-hostname", hostname);
+    }
 }