ZOOKEEPER-1467: Make server principal configurable at client side.

Make server principal configurable at the client side

Author: sujithsimon22 <sujith.abraham.simon@huawei.com>

Reviewers: Mohammad Arshad <arshad@apache.org>, enixon

Closes #1099 from sujithsimon22/ZOOKEEPER-1467

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

@@ -1231,6 +1231,12 @@ following reference
 * *zookeeper.kinit* :
     Specifies path to kinit binary. Default is "/usr/bin/kinit".
 
+* *zookeeper.server.principal* :
+    Specifies the server principal to be used by the client for authentication, while connecting to the zookeeper
+    server, when Kerberos authentication is enabled. A couple of ways to specify the server principal can be as
+    "zookeeper.server.principal = **zookeeper/zookeeper.apache.org@APACHE.ORG**" or
+    "zookeeper.server.principal = **zookeeper/zookeeper.apache.org**"
+
 <a name="C+Binding"></a>
 
 ### C Binding

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

@@ -48,6 +48,11 @@ public class SaslServerPrincipal {
      * @return the name of the principal.
      */
     static String getServerPrincipal(WrapperInetSocketAddress addr, ZKClientConfig clientConfig) {
+        String configuredServerPrincipal = clientConfig.getProperty(ZKClientConfig.ZOOKEEPER_SERVER_PRINCIPAL);
+        if (configuredServerPrincipal != null) {
+            // If server principal is already configured then return it
+            return configuredServerPrincipal;
+        }
         String principalUserName = clientConfig.getProperty(
             ZKClientConfig.ZK_SASL_CLIENT_USERNAME,
             ZKClientConfig.ZK_SASL_CLIENT_USERNAME_DEFAULT);

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

@@ -59,6 +59,7 @@ public class ZKClientConfig extends ZKConfig {
     public static final String SECURE_CLIENT = ZooKeeper.SECURE_CLIENT;
     public static final int CLIENT_MAX_PACKET_LENGTH_DEFAULT = 4096 * 1024; /* 4 MB */
     public static final String ZOOKEEPER_REQUEST_TIMEOUT = "zookeeper.request.timeout";
+    public static final String ZOOKEEPER_SERVER_PRINCIPAL = "zookeeper.server.principal";
     /**
      * Feature is disabled by default.
      */
@@ -83,6 +84,7 @@ public class ZKClientConfig extends ZKConfig {
      */
     private void initFromJavaSystemProperties() {
         setProperty(ZOOKEEPER_REQUEST_TIMEOUT, System.getProperty(ZOOKEEPER_REQUEST_TIMEOUT));
+        setProperty(ZOOKEEPER_SERVER_PRINCIPAL, System.getProperty(ZOOKEEPER_SERVER_PRINCIPAL));
     }
 
     @Override

zookeeper-server/src/main/java/org/apache/zookeeper/util/SecurityUtils.java

@@ -104,7 +104,12 @@ public final class SecurityUtils {
             // unless the system property
             // "zookeeper.server.realm" is set).
             String serverRealm = System.getProperty("zookeeper.server.realm", clientKerberosName.getRealm());
-            KerberosName serviceKerberosName = new KerberosName(servicePrincipal + "@" + serverRealm);
+            String modifiedServerPrincipal = servicePrincipal;
+            // If service principal does not contain realm, then add it
+            if (!modifiedServerPrincipal.contains("@")) {
+                modifiedServerPrincipal = modifiedServerPrincipal + "@" + serverRealm;
+            }
+            KerberosName serviceKerberosName = new KerberosName(modifiedServerPrincipal);
             final String serviceName = serviceKerberosName.getServiceName();
             final String serviceHostname = serviceKerberosName.getHostName();
             final String clientPrincipalName = clientKerberosName.toString();

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

@@ -22,6 +22,7 @@ import static org.junit.Assert.assertEquals;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 import java.io.IOException;
+import java.net.InetSocketAddress;
 import org.apache.zookeeper.client.ZKClientConfig;
 import org.junit.Test;
 
@@ -73,4 +74,15 @@ public class ClientCanonicalizeTest extends ZKTestCase {
         assertEquals("The computed principal does appear to have falled back to the original host name", "zookeeper/zookeeper.apache.org", principal);
     }
 
+    @Test
+    public void testGetServerPrincipalReturnConfiguredPrincipalName() {
+        ZKClientConfig config = new ZKClientConfig();
+        String configuredPrincipal = "zookeeper/zookeeper.apache.org@APACHE.ORG";
+        config.setProperty(ZKClientConfig.ZOOKEEPER_SERVER_PRINCIPAL, configuredPrincipal);
+
+        // Testing the case where server principal is configured, therefore InetSocketAddress is passed as null
+        String serverPrincipal = SaslServerPrincipal.getServerPrincipal((InetSocketAddress) null, config);
+        assertEquals(configuredPrincipal, serverPrincipal);
+    }
+
 }