Adding X509ZNodeGroupAclProvider support in NettyServerCnxnFactory. (#65)

* Adding X509ZNodeGroupAclProvider support in NettyServerCnxnFactory with backward compatibility

* Addressed review comments

* Addressing review comments about typo and other comments.

Co-authored-by: [Rahul Rane] <“[[email protected]]”>

Author: rahulrane50

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

@@ -71,7 +71,9 @@
 import org.apache.zookeeper.common.X509Exception;
 import org.apache.zookeeper.common.X509Exception.SSLContextException;
 import org.apache.zookeeper.server.NettyServerCnxn.HandshakeState;
+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.auth.X509AuthenticationProvider;
 import org.apache.zookeeper.server.quorum.QuorumPeerConfig;
 import org.slf4j.Logger;
@@ -423,15 +425,29 @@ public void operationComplete(Future<Channel> future) {
 
                     String authProviderProp = System.getProperty(x509Util.getSslAuthProviderProperty(), "x509");
 
-                    X509AuthenticationProvider authProvider = (X509AuthenticationProvider) ProviderRegistry.getProvider(authProviderProp);
+                    // All implementations of the AuthenticationProvider interface should be supported here. Currently
+                    // any custom implementation of X509AuthenticationProvider or ServerAuthenticationProvider is
+                    // supported with backward compatability.
+                    X509AuthenticationProvider authProvider = null;
+                    ServerAuthenticationProvider serverAuthProvider = null;
+                    try {
+                        authProvider = (X509AuthenticationProvider) ProviderRegistry.getProvider(authProviderProp);
+                    } catch (ClassCastException e) {
+                        serverAuthProvider = ProviderRegistry.getServerProvider(authProviderProp);
+                    }
 
-                    if (authProvider == null) {
+                    if (authProvider == null && serverAuthProvider == null) {
                         LOG.error("X509 Auth provider not found: {}", authProviderProp);
                         cnxn.close(ServerCnxn.DisconnectReason.AUTH_PROVIDER_NOT_FOUND);
                         return;
                     }
 
-                    KeeperException.Code code = authProvider.handleAuthentication(cnxn, null);
+                    KeeperException.Code code = KeeperException.Code.AUTHFAILED;
+                    if (authProvider != null) {
+                        code = authProvider.handleAuthentication(cnxn, null);
+                    } else if (serverAuthProvider != null) {
+                        code = serverAuthProvider.handleAuthentication(new ServerAuthenticationProvider.ServerObjs(zkServer, cnxn), null);
+                    }
                     if (KeeperException.Code.OK != code) {
                         zkServer.serverStats().incrementAuthFailedCount();
                         LOG.error("Authentication failed for session 0x{}", Long.toHexString(cnxn.getSessionId()));

zookeeper-server/src/main/java/org/apache/zookeeper/server/auth/znode/groupacl/ZkClientUriDomainMappingHelper.java

@@ -72,14 +72,15 @@ public ZkClientUriDomainMappingHelper(ZooKeeperServer zks) {
 
     this.rootPath =
         X509AuthenticationConfig.getInstance().getZnodeGroupAclClientUriDomainMappingRootPath();
+    LOG.info("ZkClientUriDomainMappingHelper::ClientUriDomainMapping Client URI domain mapping root path: {}", this.rootPath);
     if (rootPath == null) {
       throw new IllegalStateException(
           "ZkClientUriDomainMappingHelper::ClientUriDomainMapping root path config is not set!");
     }
 
     if (zks.getZKDatabase().getNode(rootPath) == null) {
       throw new IllegalStateException(
-          "ZkClientUriDomainMappingHelper::ClientUriDomainMapping root path does not exist!");
+          "ZkClientUriDomainMappingHelper::ClientUriDomainMapping root path does not exist :" + rootPath);
     }
 
     addWatches();
@@ -125,8 +126,11 @@ private void parseZNodeMapping() {
         try {
           List<String> clientUris =
               zks.getZKDatabase().getChildren(rootPath + "/" + domainName, null, null);
-          clientUris.forEach(
-              clientUri -> newClientUriToDomainNames.computeIfAbsent(clientUri, k -> new HashSet<>()).add(domainName));
+          clientUris.forEach(clientUri -> {
+              LOG.info("ZkClientUriDomainMappingHelper::parseZNodeMapping(): Adding client uri mapping: domainName : {},"
+                  + " clientUri: {}", domainName, clientUri);
+              newClientUriToDomainNames.computeIfAbsent(clientUri, k -> new HashSet<>()).add(domainName);
+          });
         } catch (KeeperException.NoNodeException e) {
           LOG.warn(
               "ZkClientUriDomainMappingHelper::parseZNodeMapping(): No clientUri ZNodes found under domain: {}",