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以後的版本都支持認證，能夠下載使用。