SimpleAclOperator { public Future<ReconcileResult<Set<SimpleAclRule>>> reconcile(String username, Set<SimpleAclRule> desired) { Promise<ReconcileResult<Set<SimpleAclRule>>> promise = Promise.promise(); vertx.createSharedWorkerExecutor("kubernetes-ops-pool").executeBlocking( future -> { Set<SimpleAclRule> current; try { current = getAcls(username); } catch (Exception e) { if (e instanceof InvalidResourceException && (desired == null || desired.isEmpty())) { future.complete(); return; } else { log.error("Reconciliation failed for user {}", username, e); future.fail(e); return; } } if (desired == null || desired.isEmpty()) { if (current.size() == 0) { log.debug("User {}: No expected Acl rules and no existing Acl rules -> NoOp", username); future.complete(ReconcileResult.noop(desired)); } else { log.debug("User {}: No expected Acl rules, but {} existing Acl rules -> Deleting rules", username, current.size()); internalDelete(username, current).onComplete(future); } } else { if (current.isEmpty()) { log.debug("User {}: {} expected Acl rules, but no existing Acl rules -> Adding rules", username, desired.size()); internalCreate(username, desired).onComplete(future); } else { log.debug("User {}: {} expected Acl rules and {} existing Acl rules -> Reconciling rules", username, desired.size(), current.size()); internalUpdate(username, desired, current).onComplete(future); } } }, false, promise ); return promise.future(); } SimpleAclOperator(Vertx vertx, Admin adminClient); Future<ReconcileResult<Set<SimpleAclRule>>> reconcile(String username, Set<SimpleAclRule> desired); Set<SimpleAclRule> getAcls(String username); Set<String> getUsersWithAcls();  }
@Test public void testReconcileInternalCreateAddsAclsToAuthorizer(VertxTestContext context) { Admin mockAdminClient = mock(AdminClient.class); SimpleAclOperator aclOp = new SimpleAclOperator(vertx, mockAdminClient); ResourcePattern resource1 = new ResourcePattern(ResourceType.CLUSTER, "kafka-cluster", PatternType.LITERAL); ResourcePattern resource2 = new ResourcePattern(ResourceType.TOPIC, "my-topic", PatternType.LITERAL); KafkaPrincipal foo = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "CN=foo"); AclBinding describeAclBinding = new AclBinding(resource1, new AccessControlEntry(foo.toString(), "*", org.apache.kafka.common.acl.AclOperation.DESCRIBE, AclPermissionType.ALLOW)); AclBinding readAclBinding = new AclBinding(resource2, new AccessControlEntry(foo.toString(), "*", org.apache.kafka.common.acl.AclOperation.READ, AclPermissionType.ALLOW)); AclBinding writeAclBinding = new AclBinding(resource2, new AccessControlEntry(foo.toString(), "*", org.apache.kafka.common.acl.AclOperation.WRITE, AclPermissionType.ALLOW)); SimpleAclRuleResource ruleResource1 = new SimpleAclRuleResource("kafka-cluster", SimpleAclRuleResourceType.CLUSTER, AclResourcePatternType.LITERAL); SimpleAclRuleResource ruleResource2 = new SimpleAclRuleResource("my-topic", SimpleAclRuleResourceType.TOPIC, AclResourcePatternType.LITERAL); SimpleAclRule resource1DescribeRule = new SimpleAclRule(AclRuleType.ALLOW, ruleResource1, "*", AclOperation.DESCRIBE); SimpleAclRule resource2ReadRule = new SimpleAclRule(AclRuleType.ALLOW, ruleResource2, "*", AclOperation.READ); SimpleAclRule resource2WriteRule = new SimpleAclRule(AclRuleType.ALLOW, ruleResource2, "*", AclOperation.WRITE); ArgumentCaptor<Collection<AclBinding>> aclBindingsCaptor = ArgumentCaptor.forClass(Collection.class); assertDoesNotThrow(() -> { mockDescribeAcls(mockAdminClient, null, emptyList()); mockCreateAcls(mockAdminClient, aclBindingsCaptor); }); Checkpoint async = context.checkpoint(); aclOp.reconcile("CN=foo", new LinkedHashSet<>(asList(resource2ReadRule, resource2WriteRule, resource1DescribeRule))) .onComplete(context.succeeding(rr -> context.verify(() -> { Collection<AclBinding> capturedAclBindings = aclBindingsCaptor.getValue(); assertThat(capturedAclBindings, hasSize(3)); assertThat(capturedAclBindings, hasItems(describeAclBinding, readAclBinding, writeAclBinding)); Set<ResourcePattern> capturedResourcePatterns = capturedAclBindings.stream().map(AclBinding::pattern).collect(Collectors.toSet()); assertThat(capturedResourcePatterns, hasSize(2)); assertThat(capturedResourcePatterns, hasItems(resource1, resource2)); async.flag(); }))); }
@Test public void testReconcileInternalUpdateCreatesNewAclsAndDeletesOldAcls(VertxTestContext context) { Admin mockAdminClient = mock(AdminClient.class); SimpleAclOperator aclOp = new SimpleAclOperator(vertx, mockAdminClient); ResourcePattern resource1 = new ResourcePattern(ResourceType.TOPIC, "my-topic", PatternType.LITERAL); ResourcePattern resource2 = new ResourcePattern(ResourceType.TOPIC, "my-topic2", PatternType.LITERAL); KafkaPrincipal foo = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "CN=foo"); AclBinding readAclBinding = new AclBinding(resource1, new AccessControlEntry(foo.toString(), "*", org.apache.kafka.common.acl.AclOperation.READ, AclPermissionType.ALLOW)); AclBinding writeAclBinding = new AclBinding(resource2, new AccessControlEntry(foo.toString(), "*", org.apache.kafka.common.acl.AclOperation.WRITE, AclPermissionType.ALLOW)); SimpleAclRuleResource resource = new SimpleAclRuleResource("my-topic2", SimpleAclRuleResourceType.TOPIC, AclResourcePatternType.LITERAL); SimpleAclRule rule1 = new SimpleAclRule(AclRuleType.ALLOW, resource, "*", AclOperation.WRITE); ArgumentCaptor<Collection<AclBinding>> aclBindingsCaptor = ArgumentCaptor.forClass(Collection.class); ArgumentCaptor<Collection<AclBindingFilter>> aclBindingFiltersCaptor = ArgumentCaptor.forClass(Collection.class); assertDoesNotThrow(() -> { mockDescribeAcls(mockAdminClient, null, Collections.singleton(readAclBinding)); mockCreateAcls(mockAdminClient, aclBindingsCaptor); mockDeleteAcls(mockAdminClient, Collections.singleton(readAclBinding), aclBindingFiltersCaptor); }); Checkpoint async = context.checkpoint(); aclOp.reconcile("CN=foo", new LinkedHashSet(asList(rule1))) .onComplete(context.succeeding(rr -> context.verify(() -> { Collection<AclBinding> capturedAclBindings = aclBindingsCaptor.getValue(); assertThat(capturedAclBindings, hasSize(1)); assertThat(capturedAclBindings, hasItem(writeAclBinding)); Set<ResourcePattern> capturedResourcePatterns = capturedAclBindings.stream().map(AclBinding::pattern).collect(Collectors.toSet()); assertThat(capturedResourcePatterns, hasSize(1)); assertThat(capturedResourcePatterns, hasItem(resource2)); Collection<AclBindingFilter> capturedAclBindingFilters = aclBindingFiltersCaptor.getValue(); assertThat(capturedAclBindingFilters, hasSize(1)); assertThat(capturedAclBindingFilters, hasItem(readAclBinding.toFilter())); Set<ResourcePatternFilter> capturedResourcePatternFilters = capturedAclBindingFilters.stream().map(AclBindingFilter::patternFilter).collect(Collectors.toSet()); assertThat(capturedResourcePatternFilters, hasSize(1)); assertThat(capturedResourcePatternFilters, hasItem(resource1.toFilter())); async.flag(); }))); }
@Test public void testReconcileInternalDelete(VertxTestContext context) { Admin mockAdminClient = mock(AdminClient.class); SimpleAclOperator aclOp = new SimpleAclOperator(vertx, mockAdminClient); ResourcePattern resource = new ResourcePattern(ResourceType.TOPIC, "my-topic", PatternType.LITERAL); KafkaPrincipal foo = new KafkaPrincipal("User", "CN=foo"); AclBinding readAclBinding = new AclBinding(resource, new AccessControlEntry(foo.toString(), "*", org.apache.kafka.common.acl.AclOperation.READ, AclPermissionType.ALLOW)); ArgumentCaptor<Collection<AclBindingFilter>> aclBindingFiltersCaptor = ArgumentCaptor.forClass(Collection.class); assertDoesNotThrow(() -> { mockDescribeAcls(mockAdminClient, null, Collections.singleton(readAclBinding)); mockDeleteAcls(mockAdminClient, Collections.singleton(readAclBinding), aclBindingFiltersCaptor); }); Checkpoint async = context.checkpoint(); aclOp.reconcile("CN=foo", null) .onComplete(context.succeeding(rr -> context.verify(() -> { Collection<AclBindingFilter> capturedAclBindingFilters = aclBindingFiltersCaptor.getValue(); assertThat(capturedAclBindingFilters, hasSize(1)); assertThat(capturedAclBindingFilters, hasItem(readAclBinding.toFilter())); Set<ResourcePatternFilter> capturedResourcePatternFilters = capturedAclBindingFilters.stream().map(AclBindingFilter::patternFilter).collect(Collectors.toSet()); assertThat(capturedResourcePatternFilters, hasSize(1)); assertThat(capturedResourcePatternFilters, hasItem(resource.toFilter())); async.flag(); }))); }