zookeeper acl认证机制及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地址认证

一般使用[scheme:id:permissions]来表示acl权限。

我们在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，即任何用户都有访问权限。

设置super超级管理员用户

一旦我们为某一个节点设置了acl，那么其余的未授权的节点是无法访问或者操作该节点的，那么系统用久了以后，假如忘记了某一个节点的密码，那么就无法再操作这个节点了，所以需要这个super超级管理员用户权限，其作用还是很大的。超级用户只能在启动服务器的时候添加，且digest必须是密文。

String m = DigestAuthenticationProvider.generateDigest("super:admin"); // 得到哈希值

打开zk目录下的/bin/zkServer.sh服务器脚本文件，找到如下一行：
nohup $JAVA "-Dzookeeper.log.dir=${ZOO_LOG_DIR}" "-Dzookeeper.root.logger=${ZOO_LOG4J_PROP}"
加一个超管的配置项，如下：

"-Dzookeeper.DigestAuthenticationProvider.superDigest=super:xQJmxLMiHGwaqBvst5y6rkB6HQs="

然后重启zkServer.sh即可。

四字命令

ZooKeeper中有一系列的命令可以查看服务器的运行状态，它们的长度通常都是4个英文字母，因此又被称之为“四字命令”。

使用方式： echo {command} | nc localhost 2181

需要首先使用yum install nc安装。

详细的四字命令介绍可参见https://www.jianshu.com/p/c96c9f8c2433。

分布式服务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之后的版本都支持认证，可以下载使用。