@VisibleForTesting
  void setKeyACLs(Configuration conf) {
    Map<String, HashMap<KeyOpType, AccessControlList>> tempKeyAcls =
        new HashMap<String, HashMap<KeyOpType,AccessControlList>>();
    Map<String, String> allKeyACLS =
        conf.getValByRegex(KMSConfiguration.KEY_ACL_PREFIX_REGEX);
    for (Map.Entry<String, String> keyAcl : allKeyACLS.entrySet()) {
      String k = keyAcl.getKey();
      // this should be of type "key.acl.<KEY_NAME>.<OP_TYPE>"
      int keyNameStarts = KMSConfiguration.KEY_ACL_PREFIX.length();
      int keyNameEnds = k.lastIndexOf(".");
      if (keyNameStarts >= keyNameEnds) {
        LOG.warn("Invalid key name '{}'", k);
      } else {
        String aclStr = keyAcl.getValue();
        String keyName = k.substring(keyNameStarts, keyNameEnds);
        String keyOp = k.substring(keyNameEnds + 1);
        KeyOpType aclType = null;
        try {
          aclType = KeyOpType.valueOf(keyOp);
        } catch (IllegalArgumentException e) {
          LOG.warn("Invalid key Operation '{}'", keyOp);
        }
        if (aclType != null) {
          // On the assumption this will be single threaded.. else we need to
          // ConcurrentHashMap
          HashMap<KeyOpType,AccessControlList> aclMap =
              tempKeyAcls.get(keyName);
          if (aclMap == null) {
            aclMap = new HashMap<KeyOpType, AccessControlList>();
            tempKeyAcls.put(keyName, aclMap);
          }
          aclMap.put(aclType, new AccessControlList(aclStr));
          LOG.info("KEY_NAME '{}' KEY_OP '{}' ACL '{}'",
              keyName, aclType, aclStr);
        }
      }
    }
    keyAcls = tempKeyAcls;

    final Map<KeyOpType, AccessControlList> tempDefaults = new HashMap<>();
    final Map<KeyOpType, AccessControlList> tempWhitelists = new HashMap<>();
    for (KeyOpType keyOp : KeyOpType.values()) {
      parseAclsWithPrefix(conf, KMSConfiguration.DEFAULT_KEY_ACL_PREFIX,
          keyOp, tempDefaults);
      parseAclsWithPrefix(conf, KMSConfiguration.WHITELIST_KEY_ACL_PREFIX,
          keyOp, tempWhitelists);
    }
    defaultKeyAcls = tempDefaults;
    whitelistKeyAcls = tempWhitelists;
  }
@Test
  public void testKeyAclReload() {
    Configuration conf = new Configuration(false);
    conf.set(DEFAULT_KEY_ACL_PREFIX + "READ", "read1");
    conf.set(DEFAULT_KEY_ACL_PREFIX + "MANAGEMENT", "");
    conf.set(DEFAULT_KEY_ACL_PREFIX + "GENERATE_EEK", "*");
    conf.set(DEFAULT_KEY_ACL_PREFIX + "DECRYPT_EEK", "decrypt1");
    conf.set(KEY_ACL + "testuser1.ALL", "testkey1");
    conf.set(WHITELIST_KEY_ACL_PREFIX + "READ", "admin_read1");
    conf.set(WHITELIST_KEY_ACL_PREFIX + "MANAGEMENT", "");
    conf.set(WHITELIST_KEY_ACL_PREFIX + "GENERATE_EEK", "*");
    conf.set(WHITELIST_KEY_ACL_PREFIX + "DECRYPT_EEK", "admin_decrypt1");
    final KMSACLs acls = new KMSACLs(conf);

    // update config and hot-reload.
    conf.set(DEFAULT_KEY_ACL_PREFIX + "READ", "read2");
    conf.set(DEFAULT_KEY_ACL_PREFIX + "MANAGEMENT", "mgmt1,mgmt2");
    conf.set(DEFAULT_KEY_ACL_PREFIX + "GENERATE_EEK", "");
    conf.set(DEFAULT_KEY_ACL_PREFIX + "DECRYPT_EEK", "decrypt2");
    conf.set(KEY_ACL + "testkey1.ALL", "testkey1,testkey2");
    conf.set(WHITELIST_KEY_ACL_PREFIX + "READ", "admin_read2");
    conf.set(WHITELIST_KEY_ACL_PREFIX + "MANAGEMENT", "admin_mgmt,admin_mgmt1");
    conf.set(WHITELIST_KEY_ACL_PREFIX + "GENERATE_EEK", "");
    conf.set(WHITELIST_KEY_ACL_PREFIX + "DECRYPT_EEK", "admin_decrypt2");
    acls.setKeyACLs(conf);

    assertDefaultKeyAcl(acls, KeyOpType.READ, "read2");
    assertDefaultKeyAcl(acls, KeyOpType.MANAGEMENT, "mgmt1", "mgmt2");
    assertDefaultKeyAcl(acls, KeyOpType.GENERATE_EEK);
    assertDefaultKeyAcl(acls, KeyOpType.DECRYPT_EEK, "decrypt2");
    assertKeyAcl("testuser1", acls, KeyOpType.ALL, "testkey1");
    assertWhitelistKeyAcl(acls, KeyOpType.READ, "admin_read2");
    assertWhitelistKeyAcl(acls, KeyOpType.MANAGEMENT,
        "admin_mgmt", "admin_mgmt1");
    assertWhitelistKeyAcl(acls, KeyOpType.GENERATE_EEK);
    assertWhitelistKeyAcl(acls, KeyOpType.DECRYPT_EEK, "admin_decrypt2");

    // reloading same config, nothing should change.
    acls.setKeyACLs(conf);
    assertDefaultKeyAcl(acls, KeyOpType.READ, "read2");
    assertDefaultKeyAcl(acls, KeyOpType.MANAGEMENT, "mgmt1", "mgmt2");
    assertDefaultKeyAcl(acls, KeyOpType.GENERATE_EEK);
    assertDefaultKeyAcl(acls, KeyOpType.DECRYPT_EEK, "decrypt2");
    assertKeyAcl("testuser1", acls, KeyOpType.ALL, "testkey1");
    assertWhitelistKeyAcl(acls, KeyOpType.READ, "admin_read2");
    assertWhitelistKeyAcl(acls, KeyOpType.MANAGEMENT,
        "admin_mgmt", "admin_mgmt1");
    assertWhitelistKeyAcl(acls, KeyOpType.GENERATE_EEK);
    assertWhitelistKeyAcl(acls, KeyOpType.DECRYPT_EEK, "admin_decrypt2");

    // test wildcard.
    conf.set(DEFAULT_KEY_ACL_PREFIX + "DECRYPT_EEK", "*");
    acls.setKeyACLs(conf);
    AccessControlList acl = acls.defaultKeyAcls.get(KeyOpType.DECRYPT_EEK);
    Assert.assertTrue(acl.isAllAllowed());
    Assert.assertTrue(acl.getUsers().isEmpty());
    // everything else should still be the same.
    assertDefaultKeyAcl(acls, KeyOpType.READ, "read2");
    assertDefaultKeyAcl(acls, KeyOpType.MANAGEMENT, "mgmt1", "mgmt2");
    assertDefaultKeyAcl(acls, KeyOpType.GENERATE_EEK);
    assertKeyAcl("testuser1", acls, KeyOpType.ALL, "testkey1");
    assertWhitelistKeyAcl(acls, KeyOpType.READ, "admin_read2");
    assertWhitelistKeyAcl(acls, KeyOpType.MANAGEMENT,
        "admin_mgmt", "admin_mgmt1");
    assertWhitelistKeyAcl(acls, KeyOpType.GENERATE_EEK);
    assertWhitelistKeyAcl(acls, KeyOpType.DECRYPT_EEK, "admin_decrypt2");

    // test new configuration should clear other items
    conf = new Configuration();
    conf.set(DEFAULT_KEY_ACL_PREFIX + "DECRYPT_EEK", "new");
    acls.setKeyACLs(conf);
    assertDefaultKeyAcl(acls, KeyOpType.DECRYPT_EEK, "new");
    Assert.assertTrue(acls.keyAcls.isEmpty());
    Assert.assertTrue(acls.whitelistKeyAcls.isEmpty());
    Assert.assertEquals("Got unexpected sized acls:"
        + acls.defaultKeyAcls, 1, acls.defaultKeyAcls.size());
  }