Java 类org.apache.curator.framework.AuthInfo 实例源码

private void createConnection() throws Exception {
  // Curator connection
  CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder();
  builder = builder.connectString(zkHostPort)
      .connectionTimeoutMs(zkSessionTimeout)
      .retryPolicy(new RetryNTimes(numRetries, zkRetryInterval));

  // Set up authorization based on fencing scheme
  List<AuthInfo> authInfos = new ArrayList<>();
  for (ZKUtil.ZKAuthInfo zkAuth : zkAuths) {
    authInfos.add(new AuthInfo(zkAuth.getScheme(), zkAuth.getAuth()));
  }
  if (useDefaultFencingScheme) {
    byte[] defaultFencingAuth =
        (zkRootNodeUsername + ":" + zkRootNodePassword).getBytes(
            Charset.forName("UTF-8"));
    authInfos.add(new AuthInfo(zkRootNodeAuthScheme, defaultFencingAuth));
  }
  builder = builder.authorization(authInfos);

  // Connect to ZK
  curatorFramework = builder.build();
  curatorFramework.start();
}

@VisibleForTesting
void enhanceBuilderWithSecurityParameters(HAConfiguration.ZookeeperProperties zookeeperProperties,
                                          CuratorFrameworkFactory.Builder builder) {

    ACLProvider aclProvider = getAclProvider(zookeeperProperties);

    AuthInfo authInfo = null;
    if (zookeeperProperties.hasAuth()) {
        authInfo = AtlasZookeeperSecurityProperties.parseAuth(zookeeperProperties.getAuth());
    }

    if (aclProvider != null) {
        LOG.info("Setting up acl provider.");
        builder.aclProvider(aclProvider);
        if (authInfo != null) {
            byte[] auth = authInfo.getAuth();
            LOG.info("Setting up auth provider with scheme: {} and id: {}", authInfo.getScheme(),
                    getIdForLogging(authInfo.getScheme(), new String(auth, Charsets.UTF_8)));
            builder.authorization(authInfo.getScheme(), auth);
        }
    }
}

private ZookeeperFactory makeZookeeperFactory(final ZookeeperFactory actualZookeeperFactory)
{
    return new ZookeeperFactory()
    {
        @Override
        public ZooKeeper newZooKeeper(String connectString, int sessionTimeout, Watcher watcher, boolean canBeReadOnly) throws Exception
        {
            ZooKeeper zooKeeper = actualZookeeperFactory.newZooKeeper(connectString, sessionTimeout, watcher, canBeReadOnly);
            for ( AuthInfo auth : authInfos )
            {
                zooKeeper.addAuthInfo(auth.getScheme(), auth.getAuth());
            }

            return zooKeeper;
        }
    };
}

public List<AuthInfo> authInfo() {
    String username = env.getProperty("rpc.client.zookeeper.username");
    String password = env.getProperty("rpc.client.zookeeper.password");
    List<AuthInfo> info = new ArrayList<AuthInfo>();
    info.add(new DigestAuthInfo(username, password));
    return info;
}

public List<AuthInfo> authInfo() {
    String username = env.getProperty("rpc.server.zookeeper.username");
    String password = env.getProperty("rpc.server.zookeeper.password");
    List<AuthInfo> info = new ArrayList<AuthInfo>();
    info.add(new DigestAuthInfo(username, password));
    return info;
}

@Bean
public List<AuthInfo> authInfo() {
    String username = env.getProperty("rpc.client.zookeeper.username");
    String password = env.getProperty("rpc.client.zookeeper.password");
    List<AuthInfo> info = new ArrayList<AuthInfo>();
    info.add(new DigestAuthInfo(username, password));
    return info;
}

@Override
protected CuratorFramework createInstance() throws Exception {
    String connectionString = resolveConnectionString();
    if(connectionString==null) {
        throw new IllegalArgumentException("Cannot resolve zookeeper connection string");
    }

    RetryPolicy retryPolicy = new ExponentialBackoffRetry(baseSleepTime, maxRetries);
    Builder curatorFrameworkBuilder = CuratorFrameworkFactory.builder()
        .connectString(connectionString)
        .retryPolicy(retryPolicy)
        .canBeReadOnly(canReadOnly);        

    String credentialString = resolveCredentialString();
    if(credentialString!=null) {
          String[] credentials = StringUtils.tokenizeToStringArray(credentialString, STRING_ARRAY_SEPARATOR);

          List<AuthInfo> authList = new ArrayList<AuthInfo>();
          for(String cred : credentials){
              String[] aclId = cred.split(":");
              String passwd = new String(Base64.decodeBase64(aclId[1].trim()),"UTF-8");
              authList.add(new AuthInfo(
                  SCHEME_DIGEST, 
                  String.format("%s:%s", aclId[0].trim(), passwd).getBytes()));    
          }

          if(!authList.isEmpty()) {
              curatorFrameworkBuilder.authorization(authList);
          }
    }

    CuratorFramework client = curatorFrameworkBuilder.build();
    client.start();

    return client;
}

private List<AuthInfo> buildAuths(CuratorFrameworkFactory.Builder builder)
{
    ImmutableList.Builder<AuthInfo> builder1 = ImmutableList.builder();
    if ( builder.getAuthInfos() != null )
    {
        builder1.addAll(builder.getAuthInfos());
    }
    return builder1.build();
}

private List<AuthInfo> authInfo() {
    List<AuthInfo> info = new ArrayList<AuthInfo>();
    info.add(new DigestAuthInfo(this.zkUsername, this.zkPassword));
    return info;
}

@Test
public void shouldGetAuth() {
    AuthInfo authInfo = AtlasZookeeperSecurityProperties.parseAuth("digest:user:password");
    assertEquals(authInfo.getScheme(), "digest");
    assertEquals(authInfo.getAuth(), "user:password".getBytes(Charsets.UTF_8));
}

public CuratorFramework createAndStartCurator(Configuration conf)
    throws Exception {
  String zkHostPort = conf.get(YarnConfiguration.RM_ZK_ADDRESS);
  if (zkHostPort == null) {
    throw new YarnRuntimeException(
        YarnConfiguration.RM_ZK_ADDRESS + " is not configured.");
  }
  int numRetries = conf.getInt(YarnConfiguration.RM_ZK_NUM_RETRIES,
      YarnConfiguration.DEFAULT_ZK_RM_NUM_RETRIES);
  int zkSessionTimeout = conf.getInt(YarnConfiguration.RM_ZK_TIMEOUT_MS,
      YarnConfiguration.DEFAULT_RM_ZK_TIMEOUT_MS);
  int zkRetryInterval = conf.getInt(YarnConfiguration.RM_ZK_RETRY_INTERVAL_MS,
      YarnConfiguration.DEFAULT_RM_ZK_RETRY_INTERVAL_MS);

  // set up zk auths
  List<ZKUtil.ZKAuthInfo> zkAuths = RMZKUtils.getZKAuths(conf);
  List<AuthInfo> authInfos = new ArrayList<>();
  for (ZKUtil.ZKAuthInfo zkAuth : zkAuths) {
    authInfos.add(new AuthInfo(zkAuth.getScheme(), zkAuth.getAuth()));
  }

  if (HAUtil.isHAEnabled(conf) && HAUtil.getConfValueForRMInstance(
      YarnConfiguration.ZK_RM_STATE_STORE_ROOT_NODE_ACL, conf) == null) {
    String zkRootNodeUsername = HAUtil
        .getConfValueForRMInstance(YarnConfiguration.RM_ADDRESS,
            YarnConfiguration.DEFAULT_RM_ADDRESS, conf);
    byte[] defaultFencingAuth =
        (zkRootNodeUsername + ":" + zkRootNodePassword)
            .getBytes(Charset.forName("UTF-8"));
    authInfos.add(new AuthInfo(new DigestAuthenticationProvider().getScheme(),
        defaultFencingAuth));
  }

  CuratorFramework client =  CuratorFrameworkFactory.builder()
      .connectString(zkHostPort)
      .sessionTimeoutMs(zkSessionTimeout)
      .retryPolicy(new RetryNTimes(numRetries, zkRetryInterval))
      .authorization(authInfos).build();
  client.start();
  return client;
}

/**
 * Get an {@link AuthInfo} by parsing input string.
 * @param authString A string of the form scheme:authString
 * @return {@link AuthInfo} with the scheme and auth taken from configuration values.
 */
public static AuthInfo parseAuth(String authString) {
    String[] authComponents = getComponents(authString, "authString", "scheme:authString");
    return new AuthInfo(authComponents[0], authComponents[1].getBytes(Charsets.UTF_8));
}