Add auth support for zookeeper registry (sofastack#421)

* 给ZookeeperRegistry添加digest认证功能

* fix code format

* override zk auth test case

* Fix format

extension-impl/registry-zk/src/main/java/com/alipay/sofa/rpc/registry/zk/ZookeeperRegistry.java

@@ -37,8 +37,10 @@
 import com.alipay.sofa.rpc.log.LoggerFactory;
 import com.alipay.sofa.rpc.registry.Registry;
 import org.apache.curator.RetryPolicy;
+import org.apache.curator.framework.AuthInfo;
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.framework.api.ACLProvider;
 import org.apache.curator.framework.imps.CuratorFrameworkState;
 import org.apache.curator.framework.recipes.cache.PathChildrenCache;
 import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
@@ -48,8 +50,11 @@
 import org.apache.curator.retry.ExponentialBackoffRetry;
 import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.ZooDefs;
+import org.apache.zookeeper.data.ACL;
 
 import java.net.URLEncoder;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
@@ -212,14 +217,22 @@ public synchronized void init() {
                 address, rootPath, preferLocalFile, ephemeralNode);
         }
         RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
-        zkClient = CuratorFrameworkFactory.builder()
+        CuratorFrameworkFactory.Builder zkClientuilder = CuratorFrameworkFactory.builder()
             .connectString(address)
             .sessionTimeoutMs(registryConfig.getConnectTimeout() * 3)
             .connectionTimeoutMs(registryConfig.getConnectTimeout())
             .canBeReadOnly(false)
             .retryPolicy(retryPolicy)
-            .defaultData(null)
-            .build();
+            .defaultData(null);
+
+        //是否需要添加zk的认证信息
+        List<AuthInfo> authInfos = buildAuthInfo();
+        if (CommonUtils.isNotEmpty(authInfos)) {
+            zkClientuilder = zkClientuilder.aclProvider(getDefaultAclProvider())
+                .authorization(authInfos);
+        }
+
+        zkClient = zkClientuilder.build();
 
         zkClient.getConnectionStateListenable().addListener(new ConnectionStateListener() {
             @Override
@@ -747,4 +760,44 @@ private void closePathChildrenCache(Map<String, PathChildrenCache> map) {
             }
         }
     }
+
+    /**
+     * 获取默认的AclProvider
+     * @return
+     */
+    private ACLProvider getDefaultAclProvider() {
+        return new ACLProvider() {
+            @Override
+            public List<ACL> getDefaultAcl() {
+                return ZooDefs.Ids.CREATOR_ALL_ACL;
+            }
+
+            @Override
+            public List<ACL> getAclForPath(String path) {
+                return ZooDefs.Ids.CREATOR_ALL_ACL;
+            }
+        };
+    }
+
+    /**
+     * 创建认证信息
+     * @return
+     */
+    private List<AuthInfo> buildAuthInfo() {
+        List<AuthInfo> info = new ArrayList<AuthInfo>();
+
+        String scheme = registryConfig.getParameter("scheme");
+
+        //如果存在多个认证信息，则在参数形式为为addAuth=user1:paasswd1,user2:passwd2
+        String addAuth = registryConfig.getParameter("addAuth");
+
+        if (StringUtils.isNotEmpty(addAuth)) {
+            String[] addAuths = addAuth.split(",");
+            for (String singleAuthInfo : addAuths) {
+                info.add(new AuthInfo(scheme, singleAuthInfo.getBytes()));
+            }
+        }
+
+        return info;
+    }
 }

test/test-integration-3rd/src/test/java/com/alipay/sofa/rpc/registry/zk/auth/ZookeeperAuthBoltServerTest.java

@@ -0,0 +1,90 @@
+/*
+ * 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 com.alipay.sofa.rpc.registry.zk.auth;
+
+import com.alipay.sofa.rpc.config.ConsumerConfig;
+import com.alipay.sofa.rpc.config.ProviderConfig;
+import com.alipay.sofa.rpc.config.RegistryConfig;
+import com.alipay.sofa.rpc.config.ServerConfig;
+import com.alipay.sofa.rpc.registry.base.BaseZkTest;
+import com.alipay.sofa.rpc.test.EchoService;
+import com.alipay.sofa.rpc.test.EchoServiceImpl;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * @author <a href="mailto:[email protected]">jianyang</a>
+ */
+public class ZookeeperAuthBoltServerTest extends BaseZkTest {
+
+    private static ServerConfig        serverConfig;
+    private static RegistryConfig      registryConfig;
+
+    private static Map<String, String> parameters = new HashMap<String, String>();
+
+    @BeforeClass
+    public static void setUp() {
+        parameters.put("scheme", "digest");
+        //如果存在多个认证信息，则在参数形式为为user1:passwd1,user2:passwd2
+        parameters.put("addAuth", "sofazk:rpc1");
+
+        registryConfig = new RegistryConfig()
+            .setProtocol("zookeeper")
+            .setAddress("127.0.0.1:2181/authtest")
+            .setParameters(parameters);
+
+        serverConfig = new ServerConfig()
+            .setProtocol("bolt") // 设置一个协议，默认bolt
+            .setPort(12200) // 设置一个端口，默认12200
+            .setDaemon(false); // 非守护线程
+    }
+
+    @AfterClass
+    public static void destroy() {
+        serverConfig.destroy();
+    }
+
+    @Test
+    public void testAll() {
+
+        ProviderConfig<EchoService> providerConfig = new ProviderConfig<EchoService>()
+            .setRegistry(registryConfig)
+            .setInterfaceId(EchoService.class.getName()) // 指定接口
+            .setRef(new EchoServiceImpl()) // 指定实现
+            .setServer(serverConfig); // 指定服务端
+        providerConfig.export(); // 发布服务
+
+        ConsumerConfig<EchoService> consumerConfig = new ConsumerConfig<EchoService>()
+            .setRegistry(registryConfig)
+            .setInterfaceId(EchoService.class.getName()) // 指定接口
+            .setProtocol("bolt") // 指定协议
+            .setTimeout(3000)
+            .setConnectTimeout(10 * 1000);
+        EchoService echoService = consumerConfig.refer();
+
+        String result = echoService.echoStr("auth test");
+
+        Assert.assertEquals("auth test", result);
+
+    }
+
+}