zookeeper认证机制及dubbo、kafka集成、zooviewer/idea zk插件配置

ZooKeeper的ACL机制

zookeeper经过ACL机制控制znode节点的访问权限。

首先介绍下znode的5种操做权限：
CREATE、READ、WRITE、DELETE、ADMIN 也就是 增、删、改、查、管理权限，这5种权限简写为crwda(即：每一个单词的首字符缩写)
注：这5种权限中，delete是指对子节点的删除权限，其它4种权限指对自身节点的操做权限

身份的认证有4种方式：
world：默认方式，至关于全世界都能访问
auth：表明已经认证经过的用户(cli中能够经过addauth digest user:pwd 来添加当前上下文中的受权用户)
digest：即用户名:密码这种方式认证，这也是业务系统中最经常使用的，下面解决dubbo认证用的就是这种模式。
ip：使用Ip地址认证

咱们在zk的客户端能够进行节点权限的查看和设置。

[zk: localhost:2181(CONNECTED) 3] create /test data
Created /test
[zk: localhost:2181(CONNECTED) 2] getAcl /test
'world,'anyone
: cdrwa
[zk: localhost:2181(CONNECTED) 3] addauth digest user:password
[zk: localhost:2181(CONNECTED) 4] setAcl /test auth:user:password:cdrwa
[zk: localhost:2181(CONNECTED) 5] getAcl /test
'digest,'user:V28q/NynI4JI3Rk54h0r8O5kMug=
: cdrwa    

从上述操做能够看出，zk新建立的znode默认访问方式为world。咱们经过addauth和setAcl给/test节点设置访问权限为digest，操做权限为cdrwa，用户名为user，密码为password。

另启zk客户端，执行ls /test,发现当前用户已经没法访问/test节点，提示信息为“Authentication is not valid”。解决方法就是addauth添加认证用户了，而且必须使用用户名和密码明文进行认证。

[zk: localhost:2181(CONNECTED) 0] ls /test
Authentication is not valid : /test
[zk: localhost:2181(CONNECTED) 4] addauth digest user:password
[zk: localhost:2181(CONNECTED) 5] ls /test
[] 
[zk: localhost:2181(CONNECTED) 6] create /test/leaf data
Created /test/leaf
[zk: localhost:2181(CONNECTED) 7] getAcl /test/leaf
'world,'anyone
: cdrwa

addauth添加digest认证用户user后，便可正常访问/test节点了。

另外，还有一点须要注意，znode的ACL是相互独立的。也就是说，任意不一样节点能够用不一样的acl列表，互不影响，而且ACL是不可被继承的，且是节点级别控制的，有些节点能够启用acl，有些能够不启用，最典型的好比dubbo启用、kafka不启用。
咱们在/test下建立leaf节点，可发现，leaf节点的认证方式为world，即任何用户都有访问权限。

分布式服务Dubbo+Zookeeper安全认证:KeeperErrorCode = NoAuth解决

网上搜了一圈，只有https://www.jianshu.com/p/02ada8d1858a和https://www.zhihu.com/question/45720203/answer/140188334提到了相关可行的解决方法。由于咱们使用的是内部集成后的dubbo（maven-shade-plugin二次打包的LZ也常常这么干），且已经走的是curator而非zkclient，因此仅仅将dubbo.registry.client改为curator不解决问题。

public CuratorZookeeperClient(URL url) {
        super(url);
        Builder builder = CuratorFrameworkFactory.builder().connectString(url.getBackupAddress()).retryPolicy(new RetryNTimes(2147483647, 1000)).connectionTimeoutMs(url.getParameter("timeout", 5000)).sessionTimeoutMs(url.getParameter("session", 60000));
        String authority = url.getAuthority();
        if (authority != null && authority.length() > 0) {
            builder = builder.authorization("digest", authority.getBytes());
        }

        this.client = builder.build();
        this.client.getConnectionStateListenable().addListener(new ConnectionStateListener() {
            public void stateChanged(CuratorFramework client, ConnectionState state) {
                if (state == ConnectionState.LOST) {
                    CuratorZookeeperClient.this.stateChanged(0);
                } else if (state == ConnectionState.CONNECTED) {
                    CuratorZookeeperClient.this.stateChanged(1);
                } else if (state == ConnectionState.RECONNECTED) {
                    CuratorZookeeperClient.this.stateChanged(2);
                }

            }
        });
        this.client.start();
    }

因此简单的方法就是本身修改CuratorZookeeperClient，将zk的认证用户名和密码注入进来，以下：

public CuratorZookeeperClient(URL url) {
        super(url);
        String username = null;
        String password = null;
        // 加载配置文件
        try {
            ResourceBundle bundle = ResourceBundle.getBundle("application");
            username = bundle.getString("rpc.registry.username");
            password = bundle.getString("rpc.registry.password");
        } catch (Exception e) {
            // NOP
        }

        if (username != null && password != null) {
            url = url.setUsername(username).setPassword(password);
        }
        Builder builder = CuratorFrameworkFactory.builder().connectString(url.getBackupAddress()).retryPolicy(new RetryNTimes(2147483647, 1000)).connectionTimeoutMs(url.getParameter("timeout", 5000)).sessionTimeoutMs(url.getParameter("session", 60000));
        String authority = url.getAuthority();
        if (authority != null && authority.length() > 0) {
            builder = builder.authorization("digest", authority.getBytes());
        }

        this.client = builder.build();
        this.client.getConnectionStateListenable().addListener(new ConnectionStateListener() {
            public void stateChanged(CuratorFramework client, ConnectionState state) {
                if (state == ConnectionState.LOST) {
                    CuratorZookeeperClient.this.stateChanged(0);
                } else if (state == ConnectionState.CONNECTED) {
                    CuratorZookeeperClient.this.stateChanged(1);
                } else if (state == ConnectionState.RECONNECTED) {
                    CuratorZookeeperClient.this.stateChanged(2);
                }

            }
        });
        this.client.start();
    }

 dubbo认证是解决了，还有kafka、平常管理用的zooviewer和idea集成的zk插件呢。。。。因此继续kafka。。

kafka链接zookeeper认证

注：仅仅启用认证的话，dubbo客户端即便不配置SASL，也是能够正常访问的，可是这样就失去了意义。

这一部分主要参考了https://www.orchome.com/500、https://codeforgeek.com/how-to-set-up-authentication-in-kafka-cluster/以及https://blog.csdn.net/sdksdk0/article/details/95336382。关于SASL和kerberos的详细介绍，参见：kerberos与sasl入坑指南

除了配置认证外，还能够选择启用或者不启用acl。 

客户端工具

zooinspector，3.4.14以后的版本都支持认证，能够下载使用。