使用Java API、Curator操作zookeeper的acl权限

水深无声 2021-06-24 16:09 430阅读 0赞

### zk原生api操作acl权限 ###

### 默认匿名权限 ###

ZooKeeper提供了如下几种验证模式（scheme）：

*  digest：Client端由用户名和密码验证，譬如user:password，digest的密码生成方式是Sha1摘要的base64形式
 *  auth：不使用任何id，代表任何已确认用户。
 *  ip：Client端由IP地址验证，譬如172.2.0.0/24
 *  world：固定用户为anyone，为所有Client端开放权限
 *  super：在这种scheme情况下，对应的id拥有超级权限，可以做任何事情(cdrwa）

注意的是，exists操作和getAcl操作并不受ACL许可控制，因此任何客户端可以查询节点的状态和节点的ACL。

节点的权限（perms）主要有以下几种：

*  Create 允许对子节点Create操作
 *  Read 允许对本节点GetChildren和GetData操作
 *  Write 允许对本节点SetData操作
 *  Delete 允许对子节点Delete操作
 *  Admin 允许对本节点setAcl操作

Znode ACL权限用一个int型的十进制数字perms表示，perms的5个二进制位分别表示setacl、delete、create、write、read。比如0x1f=adcwr，0x1=----r，0x15=a-c-r。

以下示例在创建节点的时候，赋予该节点的权限为默认匿名权限，权限位为adcwr，换成十进制数字表示就是3  
1，十六进制则是0x1f。代码如下：

package org.zero01.zk.demo;
    
    import org.apache.zookeeper.*;
    import org.apache.zookeeper.data.ACL;
    import org.apache.zookeeper.data.Stat;
    
    import java.util.List;
    
    /**
     * @program: zookeeper-connection
     * @description: zookeeper 操作节点acl权限演示
     * @author: 01
     * @create: 2018-04-27 09:20
     **/
    public class ZKNodeAcl implements Watcher {
    
        // 集群模式则是多个ip
        private static final String zkServerIps = "192.168.190.128:2181,192.168.190.129:2181,192.168.190.130:2181";
        // 超时时间
        private static final Integer timeout = 5000;
        private static ZooKeeper zooKeeper;
        private static Stat stat;
    
        // Watch事件通知方法
        public void process(WatchedEvent watchedEvent) {
        }
    
        public static void main(String[] args) throws Exception {
            zooKeeper = new ZooKeeper(zkServerIps, timeout, new ZKNodeAcl());
    
            // 这样创建的节点是默认匿名权限的：ZooDefs.Ids.OPEN_ACL_UNSAFE
            String result = zooKeeper.create("/testAclNode", "test data".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    
            Thread.sleep(1000);
    
            // 获取该节点的acl权限信息
            List<ACL> aclList = zooKeeper.getACL("/testAclNode", stat);
            for (ACL acl : aclList) {
                System.out.println("权限scheme id：" + acl.getId());
                // 获取的是十进制的int型数字
                System.out.println("权限位：" + acl.getPerms());
            }
            // 避免与服务端的连接马上断开
            Thread.sleep(1000);
        }
    }

输出

权限scheme id：'world,'anyone
    
    权限位：31

ZooDefs.Ids可以直接使用的权限如下：

ZooDefs.Ids.OPEN_ACL_UNSAFE   // 默认匿名权限，权限scheme id：'world,'anyone，权限位：31（adcwr）
    ZooDefs.Ids.READ_ACL_UNSAFE  // 只读权限，权限scheme id：'world,'anyone，权限位：1（r）

### 自定义用户权限 ###

本节介绍如何自定义用户权限，这里使用digest进行演示，因为平时工作中digest是用得最多的。我们都知道digest是使用密文进行设置的，所以我们需要自定义一个工具类来加密明文密码得到密文密码。代码如下：

package org.zero01.zk.util;
    
    import org.apache.zookeeper.server.auth.DigestAuthenticationProvider;
    
    public class AclUtils {
    
        public static String getDigestUserPwd(String id) throws Exception {
            // 加密明文密码
            return DigestAuthenticationProvider.generateDigest(id);
        }
    }

然后修改 ZKNodeAcl 类的代码如下：

package org.zero01.zk.demo;
    
    import org.apache.zookeeper.*;
    import org.apache.zookeeper.data.ACL;
    import org.apache.zookeeper.data.Id;
    import org.apache.zookeeper.data.Stat;
    import org.zero01.zk.util.AclUtils;
    
    import java.util.ArrayList;
    import java.util.List;
    
    /**
     * @program: zookeeper-connection
     * @description: zookeeper 操作节点acl权限演示
     * @author: 01
     * @create: 2018-04-27 09:20
     **/
    public class ZKNodeAcl implements Watcher {
    
        // 集群模式则是多个ip
        private static final String zkServerIps = "192.168.190.128:2181,192.168.190.129:2181,192.168.190.130:2181";
        // 超时时间
        private static final Integer timeout = 5000;
        private static ZooKeeper zooKeeper;
        private static Stat stat;
    
        public void process(WatchedEvent watchedEvent) {
        }
    
        public static void main(String[] args) throws Exception {
            zooKeeper = new ZooKeeper(zkServerIps, timeout, new ZKNodeAcl());
    
            // 自定义用户认证访问
            List<ACL> acls = new ArrayList<ACL>();  // 权限列表
            // 第一个参数是权限scheme，第二个参数是加密后的用户名和密码
            Id user1 = new Id("digest", AclUtils.getDigestUserPwd("user1:123456a"));
            Id user2 = new Id("digest", AclUtils.getDigestUserPwd("user2:123456b"));
            acls.add(new ACL(ZooDefs.Perms.ALL, user1));  // 给予所有权限
            acls.add(new ACL(ZooDefs.Perms.READ, user2));  // 只给予读权限
            acls.add(new ACL(ZooDefs.Perms.DELETE | ZooDefs.Perms.CREATE, user2));  // 多个权限的给予方式，使用 | 位运算符
    
            // 使用自定义的权限列表去创建节点
            String result = zooKeeper.create("/testDigestNode", "test data".getBytes(), acls, CreateMode.PERSISTENT);
            if (result != null) {
                System.out.println("创建节点：\t" + result + "\t成功...");
            }
    
            Thread.sleep(1000);
    
            // 获取该节点的acl权限信息
            List<ACL> aclList = zooKeeper.getACL("/testDigestNode", stat);
            for (ACL acl : aclList) {
                System.out.println("\n-----------------------\n");
                System.out.println("权限scheme id：" + acl.getId());
                System.out.println("权限位：" + acl.getPerms());
            }
            Thread.sleep(1000);
        }
    }

控制台输出信息如下：

创建节点：   /testDigestNode 成功...
    
    -----------------------
    
    权限scheme id：'digest,'user1:TQYTqd46qVVbWpOd02tLO5qb+JM=
    
    权限位：31
    
    -----------------------
    
    权限scheme id：'digest,'user2:CV4ED0rE6SxA3h/DN/WyScDMbCs=
    
    权限位：1
    
    -----------------------
    
    权限scheme id：'digest,'user2:CV4ED0rE6SxA3h/DN/WyScDMbCs=
    
    权限位：12

我们可以到服务器上查看该节点的ACL信息是否一致：

[zk: localhost:2181(CONNECTED) 15] getAcl /testDigestNode
    'digest,'user1:TQYTqd46qVVbWpOd02tLO5qb+JM=
    : cdrwa
    'digest,'user2:CV4ED0rE6SxA3h/DN/WyScDMbCs=
    : r
    'digest,'user2:CV4ED0rE6SxA3h/DN/WyScDMbCs=
    : cd
    [zk: localhost:2181(CONNECTED) 16]

如果我们需要操作一个设置了digest权限的节点，那么就需要登录用于相应权限的用户才行。在代码上也是如此，我们需要先通过addAuthInfo添加用户信息（账户:明文密码）才能够操作其拥有权限的节点。以下示例演示如何使用addAuthInfo添加用户信息并操作相应的节点，修改main方法的代码如下：

...
    public class ZKNodeAcl implements Watcher {
        ...
        public static void main(String[] args) throws Exception {
            zooKeeper = new ZooKeeper(zkServerIps, timeout, new ZKNodeAcl());
    
            // 注册过的用户必须通过addAuthInfo才能操作节点，参考命令行 addauth
            zooKeeper.addAuthInfo("digest", "user1:123456a".getBytes());
            String result = zooKeeper.create("/testDigestNode/testOneNode", "test data".getBytes(), ZooDefs.Ids.CREATOR_ALL_ACL, CreateMode.PERSISTENT);
            if (result != null) {
                System.out.println("创建子节点：\t" + result + "\t成功...");
            }
            // 获取节点数据
            byte[] data = zooKeeper.getData("/testDigestNode/testOneNode", false, stat);
            System.out.println(new String(data));
            // 设置节点数据
            zooKeeper.setData("/testDigestNode/testOneNode", "new test data".getBytes(), 0);
            Thread.sleep(1000);
        }
    }

控制台输出信息如下：

创建子节点：  /testDigestNode/testOneNode 成功...
    test data

### ip权限 ###

以上我们简单演示了默认匿名权限以及自定义的digest权限，下面简单演示下自定义ip权限的设置方式。修改 ZKNodeAcl 类中的main方法代码如下：

...
    public class ZKNodeAcl implements Watcher {
        ...
        public static void main(String[] args) throws Exception {
            zooKeeper = new ZooKeeper(zkServerIps, timeout, new ZKNodeAcl());
    
            // ip方式的acl
            List<ACL> aclsIP = new ArrayList<ACL>();  // 权限列表
            // 第一个参数是权限scheme，第二个参数是ip地址
            Id ipId1 = new Id("ip", "192.168.190.1");
            aclsIP.add(new ACL(ZooDefs.Perms.ALL, ipId1));  // 给予所有权限
    
            // 使用自定义的权限列表去创建节点
            String result = zooKeeper.create("/testIpNode", "this is test ip node data".getBytes(), aclsIP, CreateMode.PERSISTENT);
            if (result != null) {
                System.out.println("创建节点：\t" + result + "\t成功...");
            }
    
            Thread.sleep(1000);
    
            // 获取该节点的acl权限信息
            List<ACL> aclList = zooKeeper.getACL(result, stat);
            for (ACL acl : aclList) {
                System.out.println("\n-----------------------\n");
                System.out.println("权限scheme id：" + acl.getId());
                System.out.println("权限位：" + acl.getPerms());
            }
            Thread.sleep(1000);
    
            // 获取节点数据，验证ip是否有权限
            byte[] data = zooKeeper.getData("/testIpNode", false, stat);
            System.out.println("\n-----------------------\n");
            System.out.println(result + " 节点当前的数据为：" + new String(data));
        }
    }

控制台输出信息如下：

创建节点：   /testIpNode 成功...
    
    -----------------------
    
    权限scheme id：'ip,'192.168.190.1
    
    权限位：31
    
    -----------------------
    
    /testIpNode 节点当前的数据为：this is test ip node data

### curator之acl权限操作与认证授权 ###

以上介绍了zk原生api操作acl，接下来使用curator对节点的acl权限进行操作。我们先演示在创建节点时设置acl权限，现在/workspace/super只有如下节点：

[zk: localhost:2181(CONNECTED) 27] ls /workspace/super
    [xxxnode, testNode]
    [zk: localhost:2181(CONNECTED) 28]

package org.zero01.zk.curator;
    
    import org.apache.curator.RetryPolicy;
    import org.apache.curator.framework.CuratorFramework;
    import org.apache.curator.framework.CuratorFrameworkFactory;
    import org.apache.curator.retry.RetryNTimes;
    import org.apache.zookeeper.CreateMode;
    import org.apache.zookeeper.ZooDefs;
    import org.apache.zookeeper.data.ACL;
    import org.apache.zookeeper.data.Id;
    import org.zero01.zk.util.AclUtils;
    
    import java.util.ArrayList;
    import java.util.List;
    
    /**
     * @program: zookeeper-connection
     * @description: curator操作zk节点acl权限演示demo
     * @author: 01
     * @create: 2018-04-29 19:53
     **/
    public class CuratorAcl {
    
        // Curator客户端
        public CuratorFramework client = null;
        // 集群模式则是多个ip
        private static final String zkServerIps = "192.168.190.128:2181,192.168.190.129:2181,192.168.190.130:2181";
    
        public CuratorAcl() {
            RetryPolicy retryPolicy = new RetryNTimes(3, 5000);
            client = CuratorFrameworkFactory.builder().authorization("digest", "user1:123456a".getBytes())  // 认证授权，登录用户
                    .connectString(zkServerIps)
                    .sessionTimeoutMs(10000).retryPolicy(retryPolicy)
                    .namespace("workspace").build();
            client.start();
        }
    
        public void closeZKClient() {
            if (client != null) {
                this.client.close();
            }
        }
    
        public static void main(String[] args) throws Exception {
    
            // 实例化
            CuratorAcl cto = new CuratorAcl();
            boolean isZkCuratorStarted = cto.client.isStarted();
            System.out.println("当前客户的状态：" + (isZkCuratorStarted ? "连接中" : "已关闭"));
    
            String nodePath = "/super/testAclNode/testOne";
    
            // 自定义权限列表
            List<ACL> acls = new ArrayList<ACL>();
            Id user1 = new Id("digest", AclUtils.getDigestUserPwd("user1:123456a"));
            Id user2 = new Id("digest", AclUtils.getDigestUserPwd("user2:123456b"));
            acls.add(new ACL(ZooDefs.Perms.ALL, user1));
            acls.add(new ACL(ZooDefs.Perms.READ, user2));
            acls.add(new ACL(ZooDefs.Perms.DELETE | ZooDefs.Perms.CREATE, user2));
    
            // 创建节点，使用自定义权限列表来设置节点的acl权限
            byte[] nodeData = "child-data".getBytes();
            cto.client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).withACL(acls).forPath(nodePath, nodeData);
    
            cto.closeZKClient();
            boolean isZkCuratorStarted2 = cto.client.isStarted();
            System.out.println("当前客户的状态：" + (isZkCuratorStarted2 ? "连接中" : "已关闭"));
        }
    }

运行该类，然后到zookeeper服务器上，通过命令行进行如下操作：

[zk: localhost:2181(CONNECTED) 19] ls /workspace/super/testAclNode    
    [testOne]
    [zk: localhost:2181(CONNECTED) 20] getAcl /workspace/super/testAclNode
    'world,'anyone
    : cdrwa
    [zk: localhost:2181(CONNECTED) 21] getAcl /workspace/super/testAclNode/testOne
    'digest,'user1:TQYTqd46qVVbWpOd02tLO5qb+JM=
    : cdrwa
    'digest,'user2:CV4ED0rE6SxA3h/DN/WyScDMbCs=
    : r
    'digest,'user2:CV4ED0rE6SxA3h/DN/WyScDMbCs=
    : cd
    [zk: localhost:2181(CONNECTED) 22]

可以看到，当递归创建节点时，只会对最末端的子节点赋予自定义的acl权限，父节点都是zk默认的匿名权限。

如果想要在递归创建节点时，父节点和子节点的acl权限都是我们自定义的权限，那么就需要在withACL方法中，传递一个true，表示递归创建时所有节点的权限，都是我们设置的权限。修改代码如下：

// 创建节点，使用自定义权限列表来设置节点的acl权限
    byte[] nodeData = "child-data".getBytes();
    cto.client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).withACL(acls, true).forPath(nodePath, nodeData);

运行该类，然后到zookeeper服务器上，通过命令行进行如下操作：

[zk: localhost:2181(CONNECTED) 28] getAcl /workspace/super/testAclNodeTwo
    'digest,'user1:TQYTqd46qVVbWpOd02tLO5qb+JM=
    : cdrwa
    'digest,'user2:CV4ED0rE6SxA3h/DN/WyScDMbCs=
    : r
    'digest,'user2:CV4ED0rE6SxA3h/DN/WyScDMbCs=
    : cd
    [zk: localhost:2181(CONNECTED) 30] getAcl /workspace/super/testAclNodeTwo/testOne
    'digest,'user1:TQYTqd46qVVbWpOd02tLO5qb+JM=
    : cdrwa
    'digest,'user2:CV4ED0rE6SxA3h/DN/WyScDMbCs=
    : r
    'digest,'user2:CV4ED0rE6SxA3h/DN/WyScDMbCs=
    : cd
    [zk: localhost:2181(CONNECTED) 31]

如上，可以看到，创建的全部节点的acl权限都是我们设置的自定义权限。最后我们再来演示如何修改一个已存在的节点的acl权限，修改 CuratorAcl 类中的main方法代码如下：

public static void main(String[] args) throws Exception {
            // 实例化
            CuratorAcl cto = new CuratorAcl();
            boolean isZkCuratorStarted = cto.client.isStarted();
            System.out.println("当前客户的状态：" + (isZkCuratorStarted ? "连接中" : "已关闭"));
    
            String nodePath = "/super/testAclNodeTwo/testOne";
    
            // 自定义权限列表
            List<ACL> acls = new ArrayList<ACL>();
            Id user1 = new Id("digest", AclUtils.getDigestUserPwd("user1:123456a"));
            Id user2 = new Id("digest", AclUtils.getDigestUserPwd("user2:123456b"));
            acls.add(new ACL(ZooDefs.Perms.READ | ZooDefs.Perms.CREATE | ZooDefs.Perms.ADMIN, user1));
            acls.add(new ACL(ZooDefs.Perms.READ | ZooDefs.Perms.DELETE | ZooDefs.Perms.CREATE, user2));
    
            // 设置指定节点的acl权限
            cto.client.setACL().withACL(acls).forPath(nodePath);
    
            cto.closeZKClient();
            boolean isZkCuratorStarted2 = cto.client.isStarted();
            System.out.println("当前客户的状态：" + (isZkCuratorStarted2 ? "连接中" : "已关闭"));
        }

运行该类，然后到zookeeper服务器上，通过命令行进行如下操作：

[zk: localhost:2181(CONNECTED) 31] getAcl /workspace/super/testAclNodeTwo/testOne
    'digest,'user1:TQYTqd46qVVbWpOd02tLO5qb+JM=
    : cra
    'digest,'user2:CV4ED0rE6SxA3h/DN/WyScDMbCs=
    : cdr
    [zk: localhost:2181(CONNECTED) 32]

可以看到，成功修改了该节点的acl权限。

参考

[http://blog.51cto.com/zero01/2108483][http_blog.51cto.com_zero01_2108483]

[http_blog.51cto.com_zero01_2108483]: http://blog.51cto.com/zero01/2108483