KafkaPrivilegeValidator implements PrivilegeValidator { @Override public void validate(PrivilegeValidatorContext context) throws ConfigurationException { List<String> splits = Lists.newArrayList(); for (String section : AUTHORIZABLE_SPLITTER.split(context.getPrivilege())) { splits.add(section); } if (splits.size() < 2 || splits.size() > 3) { throw new ConfigurationException(KafkaPrivilegeHelpMsg); } if (isAction(splits.get(0))) { throw new ConfigurationException("Kafka privilege can not start with an action.\n" + KafkaPrivilegeHelpMsg); } KafkaAuthorizable hostAuthorizable = KafkaModelAuthorizables.from(splits.get(0)); if (hostAuthorizable == null) { throw new ConfigurationException("No Kafka authorizable found for " + splits.get(0) + "\n." + KafkaPrivilegeHelpMsg); } if (!(hostAuthorizable instanceof Host)) { throw new ConfigurationException("Kafka privilege must begin with host authorizable.\n" + KafkaPrivilegeHelpMsg); } if (splits.size() == 3) { if (isAction(splits.get(1))) { throw new ConfigurationException("Kafka privilege can have action only at the end of privilege.\n" + KafkaPrivilegeHelpMsg); } KafkaAuthorizable authorizable = KafkaModelAuthorizables.from(splits.get(1)); if (authorizable == null) { throw new ConfigurationException("No Kafka authorizable found for " + splits.get(1) + "\n." + KafkaPrivilegeHelpMsg); } if (authorizable instanceof Host) { throw new ConfigurationException("Host authorizable can be specified just once in a Kafka privilege.\n" + KafkaPrivilegeHelpMsg); } } if (!isAction(splits.get(splits.size() - 1))) { throw new ConfigurationException("Kafka privilege must end with a valid action.\n" + KafkaPrivilegeHelpMsg); } } KafkaPrivilegeValidator(); @Override void validate(PrivilegeValidatorContext context); static final String KafkaPrivilegeHelpMsg; }
@Test public void testOnlyHostResource() { KafkaPrivilegeValidator kafkaPrivilegeValidator = new KafkaPrivilegeValidator(); try { kafkaPrivilegeValidator.validate(new PrivilegeValidatorContext("host=host1")); } catch (ConfigurationException ex) { Assert.assertEquals(KafkaPrivilegeValidator.KafkaPrivilegeHelpMsg, ex.getMessage()); } }
@Test public void testValidPrivileges() throws Exception { KafkaPrivilegeValidator kafkaPrivilegeValidator = new KafkaPrivilegeValidator(); try { kafkaPrivilegeValidator.validate(new PrivilegeValidatorContext("host=host1->cluster=kafka-cluster->action=read")); } catch (ConfigurationException ex) { Assert.fail("Not expected ConfigurationException"); } try { kafkaPrivilegeValidator.validate(new PrivilegeValidatorContext("host=host1->topic=t1->action=read")); } catch (ConfigurationException ex) { Assert.fail("Not expected ConfigurationException"); } try { kafkaPrivilegeValidator.validate(new PrivilegeValidatorContext("host=host1->consumergroup=g1->action=read")); } catch (ConfigurationException ex) { Assert.fail("Not expected ConfigurationException"); } }
@Test public void testInvalidHostResource() throws Exception { KafkaPrivilegeValidator kafkaPrivilegeValidator = new KafkaPrivilegeValidator(); try { kafkaPrivilegeValidator.validate(new PrivilegeValidatorContext("hhost=host1->cluster=kafka-cluster->action=read")); Assert.fail("Expected ConfigurationException"); } catch (ConfigurationException ex) { } }
@Test public void testInvalidClusterResource() throws Exception { KafkaPrivilegeValidator kafkaPrivilegeValidator = new KafkaPrivilegeValidator(); try { kafkaPrivilegeValidator.validate(new PrivilegeValidatorContext("host=host1->clluster=kafka-cluster->action=read")); Assert.fail("Expected ConfigurationException"); } catch (ConfigurationException ex) { } }
@Test public void testInvalidTopicResource() throws Exception { KafkaPrivilegeValidator kafkaPrivilegeValidator = new KafkaPrivilegeValidator(); try { kafkaPrivilegeValidator.validate(new PrivilegeValidatorContext("host=host1->ttopic=t1->action=read")); Assert.fail("Expected ConfigurationException"); } catch (ConfigurationException ex) { } }
@Test public void testInvalidConsumerGroupResource() throws Exception { KafkaPrivilegeValidator kafkaPrivilegeValidator = new KafkaPrivilegeValidator(); try { kafkaPrivilegeValidator.validate(new PrivilegeValidatorContext("host=host1->coonsumergroup=g1->action=read")); Assert.fail("Expected ConfigurationException"); } catch (ConfigurationException ex) { } }
@Test public void testPrivilegeMustHaveExcatlyOneHost() { KafkaPrivilegeValidator kafkaPrivilegeValidator = new KafkaPrivilegeValidator(); try { kafkaPrivilegeValidator.validate(new PrivilegeValidatorContext("host=host1->host=host2->action=read")); Assert.fail("Multiple Host resources are not allowed within a Kafka privilege."); } catch (ConfigurationException ex) { Assert.assertEquals("Host authorizable can be specified just once in a Kafka privilege.\n" + KafkaPrivilegeValidator.KafkaPrivilegeHelpMsg, ex.getMessage()); } }
@Test public void testPrivilegeCanNotStartWithAction() { KafkaPrivilegeValidator kafkaPrivilegeValidator = new KafkaPrivilegeValidator(); try { kafkaPrivilegeValidator.validate(new PrivilegeValidatorContext("action=write->host=host1->topic=t1")); Assert.fail("Kafka privilege can not start with an action."); } catch (ConfigurationException ex) { Assert.assertEquals("Kafka privilege can not start with an action.\n" + KafkaPrivilegeValidator.KafkaPrivilegeHelpMsg, ex.getMessage()); } }
@Test public void testPrivilegeWithMoreParts() { KafkaPrivilegeValidator kafkaPrivilegeValidator = new KafkaPrivilegeValidator(); try { kafkaPrivilegeValidator.validate(new PrivilegeValidatorContext("host=host1->topic=t1->consumergroup=cg1->action=read")); Assert.fail("Kafka privilege can have one Host authorizable, at most one non Host authorizable and one action."); } catch (ConfigurationException ex) { Assert.assertEquals(KafkaPrivilegeValidator.KafkaPrivilegeHelpMsg, ex.getMessage()); } }
@Test public void testPrivilegeNotEndingWithAction() { KafkaPrivilegeValidator kafkaPrivilegeValidator = new KafkaPrivilegeValidator(); try { kafkaPrivilegeValidator.validate(new PrivilegeValidatorContext("host=host1->topic=t1->consumergroup=cg1")); Assert.fail("Kafka privilege must end with a valid action."); } catch (ConfigurationException ex) { Assert.assertEquals("Kafka privilege must end with a valid action.\n" + KafkaPrivilegeValidator.KafkaPrivilegeHelpMsg, ex.getMessage()); } }
@Test public void testPrivilegeNotEndingWithValidAction() { KafkaPrivilegeValidator kafkaPrivilegeValidator = new KafkaPrivilegeValidator(); try { kafkaPrivilegeValidator.validate(new PrivilegeValidatorContext("host=host1->topic=t1->action=bla")); Assert.fail("Kafka privilege must end with a valid action."); } catch (ConfigurationException ex) { Assert.assertEquals("Kafka privilege must end with a valid action.\n" + KafkaPrivilegeValidator.KafkaPrivilegeHelpMsg, ex.getMessage()); } }