KafkaUserOperator extends AbstractOperator<KafkaUser, KafkaUserSpec, KafkaUserStatus,
        CrdOperator<KubernetesClient, KafkaUser, KafkaUserList, DoneableKafkaUser>> { @Override protected Future<KafkaUserStatus> createOrUpdate(Reconciliation reconciliation, KafkaUser resource) { Secret clientsCaCert = secretOperations.get(caNamespace, caCertName); Secret clientsCaKey = secretOperations.get(caNamespace, caKeyName); Secret userSecret = secretOperations.get(reconciliation.namespace(), KafkaUserModel.getSecretName(reconciliation.name())); KafkaUserStatus userStatus = new KafkaUserStatus(); String namespace = reconciliation.namespace(); String userName = reconciliation.name(); KafkaUserModel user; try { user = KafkaUserModel.fromCrd(certManager, passwordGenerator, resource, clientsCaCert, clientsCaKey, userSecret); } catch (Exception e) { StatusUtils.setStatusConditionAndObservedGeneration(resource, userStatus, Future.failedFuture(e)); return Future.failedFuture(new ReconciliationException(userStatus, e)); } log.debug("{}: Updating User {} in namespace {}", reconciliation, userName, namespace); Secret desired = user.generateSecret(); String password = null; if (desired != null && desired.getData().get("password") != null) { password = new String(Base64.getDecoder().decode(desired.getData().get("password")), StandardCharsets.US_ASCII); } Set<SimpleAclRule> tlsAcls = null; Set<SimpleAclRule> scramOrNoneAcls = null; KafkaUserQuotas tlsQuotas = null; KafkaUserQuotas scramOrNoneQuotas = null; if (user.isTlsUser()) { tlsAcls = user.getSimpleAclRules(); tlsQuotas = user.getQuotas(); } else if (user.isScramUser() || user.isNoneUser()) { scramOrNoneAcls = user.getSimpleAclRules(); scramOrNoneQuotas = user.getQuotas(); } KafkaUserQuotas finalScramOrNoneQuotas = scramOrNoneQuotas; KafkaUserQuotas finalTlsQuotas = tlsQuotas; Promise<KafkaUserStatus> handler = Promise.promise(); CompositeFuture.join( scramShaCredentialOperator.reconcile(user.getName(), password) .compose(ignore -> CompositeFuture.join(kafkaUserQuotasOperator.reconcile(KafkaUserModel.getTlsUserName(userName), finalTlsQuotas), kafkaUserQuotasOperator.reconcile(KafkaUserModel.getScramUserName(userName), finalScramOrNoneQuotas))), reconcileSecretAndSetStatus(namespace, user, desired, userStatus), aclOperations.reconcile(KafkaUserModel.getTlsUserName(userName), tlsAcls), aclOperations.reconcile(KafkaUserModel.getScramUserName(userName), scramOrNoneAcls)) .onComplete(reconciliationResult -> { StatusUtils.setStatusConditionAndObservedGeneration(resource, userStatus, reconciliationResult.mapEmpty()); userStatus.setUsername(user.getUserName()); if (reconciliationResult.succeeded()) { handler.complete(userStatus); } else { handler.fail(new ReconciliationException(userStatus, reconciliationResult.cause())); } }); return handler.future(); } KafkaUserOperator(Vertx vertx,
                             CertManager certManager,
                             CrdOperator<KubernetesClient, KafkaUser, KafkaUserList, DoneableKafkaUser> crdOperator,
                             Labels labels,
                             SecretOperator secretOperations,
                             ScramShaCredentialsOperator scramShaCredentialOperator,
                             KafkaUserQuotasOperator kafkaUserQuotasOperator,
                             SimpleAclOperator aclOperations, String caCertName, String caKeyName, String caNamespace); @Override Optional<LabelSelector> selector(); @Override Future<Set<NamespaceAndName>> allResourceNames(String namespace);  }
@Test public void testCreateTlsUser(VertxTestContext context) { CrdOperator mockCrdOps = mock(CrdOperator.class); SecretOperator mockSecretOps = mock(SecretOperator.class); SimpleAclOperator aclOps = mock(SimpleAclOperator.class); ScramShaCredentialsOperator scramOps = mock(ScramShaCredentialsOperator.class); KafkaUserQuotasOperator quotasOps = mock(KafkaUserQuotasOperator.class); ArgumentCaptor<String> secretNamespaceCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<String> secretNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<Secret> secretCaptor = ArgumentCaptor.forClass(Secret.class); when(mockSecretOps.reconcile(secretNamespaceCaptor.capture(), secretNameCaptor.capture(), secretCaptor.capture())).thenReturn(Future.succeededFuture()); ArgumentCaptor<String> aclNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<Set<SimpleAclRule>> aclRulesCaptor = ArgumentCaptor.forClass(Set.class); when(aclOps.reconcile(aclNameCaptor.capture(), aclRulesCaptor.capture())).thenReturn(Future.succeededFuture()); when(scramOps.reconcile(any(), any())).thenReturn(Future.succeededFuture()); when(quotasOps.reconcile(any(), any())).thenReturn(Future.succeededFuture()); KafkaUserOperator op = new KafkaUserOperator(vertx, mockCertManager, mockCrdOps, Labels.EMPTY, mockSecretOps, scramOps, quotasOps, aclOps, ResourceUtils.CA_CERT_NAME, ResourceUtils.CA_KEY_NAME, ResourceUtils.NAMESPACE); KafkaUser user = ResourceUtils.createKafkaUserTls(); Secret clientsCa = ResourceUtils.createClientsCaCertSecret(); Secret clientsCaKey = ResourceUtils.createClientsCaKeySecret(); when(mockSecretOps.get(anyString(), eq("user-cert"))).thenReturn(clientsCa); when(mockSecretOps.get(anyString(), eq("user-key"))).thenReturn(clientsCaKey); when(mockCrdOps.getAsync(anyString(), anyString())).thenReturn(Future.succeededFuture(user)); when(mockCrdOps.updateStatusAsync(any(KafkaUser.class))).thenReturn(Future.succeededFuture()); Checkpoint async = context.checkpoint(); op.createOrUpdate(new Reconciliation("test-trigger", KafkaUser.RESOURCE_KIND, ResourceUtils.NAMESPACE, ResourceUtils.NAME), user) .onComplete(context.succeeding(v -> context.verify(() -> { List<String> capturedNames = secretNameCaptor.getAllValues(); assertThat(capturedNames, hasSize(1)); assertThat(capturedNames.get(0), is(ResourceUtils.NAME)); List<String> capturedNamespaces = secretNamespaceCaptor.getAllValues(); assertThat(capturedNamespaces, hasSize(1)); assertThat(capturedNamespaces.get(0), is(ResourceUtils.NAMESPACE)); List<Secret> capturedSecrets = secretCaptor.getAllValues(); assertThat(capturedSecrets, hasSize(1)); Secret captured = capturedSecrets.get(0); assertThat(captured.getMetadata().getName(), is(user.getMetadata().getName())); assertThat(captured.getMetadata().getNamespace(), is(user.getMetadata().getNamespace())); assertThat(captured.getMetadata().getLabels(), is(Labels.fromMap(user.getMetadata().getLabels()) .withStrimziKind(KafkaUser.RESOURCE_KIND) .withKubernetesName(KafkaUserModel.KAFKA_USER_OPERATOR_NAME) .withKubernetesInstance(ResourceUtils.NAME) .withKubernetesPartOf(ResourceUtils.NAME) .withKubernetesManagedBy(KafkaUserModel.KAFKA_USER_OPERATOR_NAME) .toMap())); assertThat(new String(Base64.getDecoder().decode(captured.getData().get("ca.crt"))), is("clients-ca-crt")); assertThat(new String(Base64.getDecoder().decode(captured.getData().get("user.crt"))), is("crt file")); assertThat(new String(Base64.getDecoder().decode(captured.getData().get("user.key"))), is("key file")); List<String> capturedAclNames = aclNameCaptor.getAllValues(); assertThat(capturedAclNames, hasSize(2)); assertThat(capturedAclNames.get(0), is(KafkaUserModel.getTlsUserName(ResourceUtils.NAME))); assertThat(capturedAclNames.get(1), is(KafkaUserModel.getScramUserName(ResourceUtils.NAME))); List<Set<SimpleAclRule>> capturedAcls = aclRulesCaptor.getAllValues(); assertThat(capturedAcls, hasSize(2)); Set<SimpleAclRule> aclRules = capturedAcls.get(0); assertThat(aclRules, hasSize(ResourceUtils.createExpectedSimpleAclRules(user).size())); assertThat(aclRules, is(ResourceUtils.createExpectedSimpleAclRules(user))); assertThat(capturedAcls.get(1), is(nullValue())); async.flag(); }))); }
@Test public void testUpdateUserNoChange(VertxTestContext context) { CrdOperator mockCrdOps = mock(CrdOperator.class); SecretOperator mockSecretOps = mock(SecretOperator.class); SimpleAclOperator aclOps = mock(SimpleAclOperator.class); ScramShaCredentialsOperator scramOps = mock(ScramShaCredentialsOperator.class); KafkaUserQuotasOperator quotasOps = mock(KafkaUserQuotasOperator.class); ArgumentCaptor<String> secretNamespaceCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<String> secretNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<Secret> secretCaptor = ArgumentCaptor.forClass(Secret.class); when(mockSecretOps.reconcile(secretNamespaceCaptor.capture(), secretNameCaptor.capture(), secretCaptor.capture())).thenReturn(Future.succeededFuture()); when(scramOps.reconcile(any(), any())).thenReturn(Future.succeededFuture()); ArgumentCaptor<String> aclNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<Set<SimpleAclRule>> aclRulesCaptor = ArgumentCaptor.forClass(Set.class); when(aclOps.reconcile(aclNameCaptor.capture(), aclRulesCaptor.capture())).thenReturn(Future.succeededFuture()); KafkaUser user = ResourceUtils.createKafkaUserTls(); when(mockCrdOps.getAsync(anyString(), anyString())).thenReturn(Future.succeededFuture(user)); when(mockCrdOps.updateStatusAsync(any(KafkaUser.class))).thenReturn(Future.succeededFuture()); KafkaUserOperator op = new KafkaUserOperator(vertx, mockCertManager, mockCrdOps, Labels.EMPTY, mockSecretOps, scramOps, quotasOps, aclOps, ResourceUtils.CA_CERT_NAME, ResourceUtils.CA_KEY_NAME, ResourceUtils.NAMESPACE); Secret clientsCa = ResourceUtils.createClientsCaCertSecret(); Secret clientsCaKey = ResourceUtils.createClientsCaKeySecret(); Secret userCert = ResourceUtils.createUserSecretTls(); when(mockSecretOps.get(anyString(), eq("user-cert"))).thenReturn(clientsCa); when(mockSecretOps.get(anyString(), eq("user-key"))).thenReturn(clientsCaKey); when(mockSecretOps.get(anyString(), eq(KafkaUserModel.getSecretName(user.getMetadata().getName())))).thenReturn(userCert); when(quotasOps.reconcile(any(), any())).thenReturn(Future.succeededFuture()); Checkpoint async = context.checkpoint(); op.createOrUpdate(new Reconciliation("test-trigger", KafkaUser.RESOURCE_KIND, ResourceUtils.NAMESPACE, ResourceUtils.NAME), user) .onComplete(context.succeeding(v -> context.verify(() -> { List<String> capturedNames = secretNameCaptor.getAllValues(); assertThat(capturedNames, hasSize(1)); assertThat(capturedNames.get(0), is(ResourceUtils.NAME)); List<String> capturedNamespaces = secretNamespaceCaptor.getAllValues(); assertThat(capturedNamespaces, hasSize(1)); assertThat(capturedNamespaces.get(0), is(ResourceUtils.NAMESPACE)); List<Secret> capturedSecrets = secretCaptor.getAllValues(); assertThat(capturedSecrets, hasSize(1)); Secret captured = capturedSecrets.get(0); assertThat(captured.getMetadata().getName(), is(userCert.getMetadata().getName())); assertThat(captured.getMetadata().getNamespace(), is(userCert.getMetadata().getNamespace())); assertThat(captured.getMetadata().getLabels(), is(userCert.getMetadata().getLabels())); assertThat(captured.getData().get("ca.crt"), is(userCert.getData().get("ca.crt"))); assertThat(captured.getData().get("user.crt"), is(userCert.getData().get("user.crt"))); assertThat(captured.getData().get("user.key"), is(userCert.getData().get("user.key"))); List<String> capturedAclNames = aclNameCaptor.getAllValues(); assertThat(capturedAclNames, hasSize(2)); assertThat(capturedAclNames.get(0), is(KafkaUserModel.getTlsUserName(ResourceUtils.NAME))); assertThat(capturedAclNames.get(1), is(KafkaUserModel.getScramUserName(ResourceUtils.NAME))); List<Set<SimpleAclRule>> capturedAcls = aclRulesCaptor.getAllValues(); assertThat(capturedAcls, hasSize(2)); Set<SimpleAclRule> aclRules = capturedAcls.get(0); assertThat(aclRules, hasSize(ResourceUtils.createExpectedSimpleAclRules(user).size())); assertThat(aclRules, is(ResourceUtils.createExpectedSimpleAclRules(user))); assertThat(capturedAcls.get(1), is(nullValue())); async.flag(); }))); }
@Test public void testUpdateUserNoAuthenticationAndNoAuthorization(VertxTestContext context) { CrdOperator mockCrdOps = mock(CrdOperator.class); SecretOperator mockSecretOps = mock(SecretOperator.class); SimpleAclOperator aclOps = mock(SimpleAclOperator.class); ScramShaCredentialsOperator scramOps = mock(ScramShaCredentialsOperator.class); KafkaUserQuotasOperator quotasOps = mock(KafkaUserQuotasOperator.class); ArgumentCaptor<String> secretNamespaceCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<String> secretNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<Secret> secretCaptor = ArgumentCaptor.forClass(Secret.class); when(mockSecretOps.reconcile(secretNamespaceCaptor.capture(), secretNameCaptor.capture(), secretCaptor.capture())).thenReturn(Future.succeededFuture()); when(scramOps.reconcile(any(), any())).thenReturn(Future.succeededFuture()); ArgumentCaptor<String> aclNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<Set<SimpleAclRule>> aclRulesCaptor = ArgumentCaptor.forClass(Set.class); when(aclOps.reconcile(aclNameCaptor.capture(), aclRulesCaptor.capture())).thenReturn(Future.succeededFuture()); KafkaUser user = ResourceUtils.createKafkaUserTls(); user.getSpec().setAuthorization(null); user.getSpec().setAuthentication(null); when(mockCrdOps.getAsync(anyString(), anyString())).thenReturn(Future.succeededFuture(user)); when(mockCrdOps.updateStatusAsync(any(KafkaUser.class))).thenReturn(Future.succeededFuture()); KafkaUserOperator op = new KafkaUserOperator(vertx, mockCertManager, mockCrdOps, Labels.EMPTY, mockSecretOps, scramOps, quotasOps, aclOps, ResourceUtils.CA_CERT_NAME, ResourceUtils.CA_KEY_NAME, ResourceUtils.NAMESPACE); when(quotasOps.reconcile(any(), any())).thenReturn(Future.succeededFuture()); Checkpoint async = context.checkpoint(); op.createOrUpdate(new Reconciliation("test-trigger", KafkaUser.RESOURCE_KIND, ResourceUtils.NAMESPACE, ResourceUtils.NAME), user) .onComplete(context.succeeding(v -> context.verify(() -> { List<String> capturedNames = secretNameCaptor.getAllValues(); assertThat(capturedNames, hasSize(1)); assertThat(capturedNames.get(0), is(ResourceUtils.NAME)); List<String> capturedNamespaces = secretNamespaceCaptor.getAllValues(); assertThat(capturedNamespaces, hasSize(1)); assertThat(capturedNamespaces.get(0), is(ResourceUtils.NAMESPACE)); List<Secret> capturedSecrets = secretCaptor.getAllValues(); assertThat(capturedSecrets, hasSize(1)); Secret captured = capturedSecrets.get(0); assertThat(captured, is(nullValue())); List<String> capturedAclNames = aclNameCaptor.getAllValues(); assertThat(capturedAclNames, hasSize(2)); assertThat(capturedAclNames.get(0), is(KafkaUserModel.getTlsUserName(ResourceUtils.NAME))); assertThat(capturedAclNames.get(1), is(KafkaUserModel.getScramUserName(ResourceUtils.NAME))); List<Set<SimpleAclRule>> capturedAcls = aclRulesCaptor.getAllValues(); assertThat(capturedAcls, hasSize(2)); assertThat(capturedAcls.get(0), is(nullValue())); assertThat(capturedAcls.get(1), is(nullValue())); async.flag(); }))); }
@Test public void testUpdateUserNewCert(VertxTestContext context) { CrdOperator mockCrdOps = mock(CrdOperator.class); SecretOperator mockSecretOps = mock(SecretOperator.class); SimpleAclOperator aclOps = mock(SimpleAclOperator.class); ScramShaCredentialsOperator scramOps = mock(ScramShaCredentialsOperator.class); KafkaUserQuotasOperator quotasOps = mock(KafkaUserQuotasOperator.class); ArgumentCaptor<String> secretNamespaceCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<String> secretNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<Secret> secretCaptor = ArgumentCaptor.forClass(Secret.class); when(mockSecretOps.reconcile(secretNamespaceCaptor.capture(), secretNameCaptor.capture(), secretCaptor.capture())).thenReturn(Future.succeededFuture()); ArgumentCaptor<String> aclNameCaptor = ArgumentCaptor.forClass(String.class); ArgumentCaptor<Set<SimpleAclRule>> aclRulesCaptor = ArgumentCaptor.forClass(Set.class); when(aclOps.reconcile(aclNameCaptor.capture(), aclRulesCaptor.capture())).thenReturn(Future.succeededFuture()); when(scramOps.reconcile(any(), any())).thenReturn(Future.succeededFuture()); when(quotasOps.reconcile(any(), any())).thenReturn(Future.succeededFuture()); when(mockCrdOps.getAsync(anyString(), anyString())).thenReturn(Future.succeededFuture()); KafkaUserOperator op = new KafkaUserOperator(vertx, mockCertManager, mockCrdOps, Labels.EMPTY, mockSecretOps, scramOps, quotasOps, aclOps, ResourceUtils.CA_CERT_NAME, ResourceUtils.CA_KEY_NAME, ResourceUtils.NAMESPACE); KafkaUser user = ResourceUtils.createKafkaUserTls(); Secret clientsCa = ResourceUtils.createClientsCaCertSecret(); clientsCa.getData().put("ca.crt", Base64.getEncoder().encodeToString("different-clients-ca-crt".getBytes())); Secret clientsCaKey = ResourceUtils.createClientsCaKeySecret(); clientsCaKey.getData().put("ca.key", Base64.getEncoder().encodeToString("different-clients-ca-key".getBytes())); Secret userCert = ResourceUtils.createUserSecretTls(); when(mockSecretOps.get(anyString(), eq("user-cert"))).thenReturn(clientsCa); when(mockSecretOps.get(anyString(), eq("user-key"))).thenReturn(clientsCaKey); when(mockSecretOps.get(anyString(), eq(KafkaUserModel.getSecretName(user.getMetadata().getName())))).thenReturn(userCert); when(mockCrdOps.getAsync(anyString(), anyString())).thenReturn(Future.succeededFuture(user)); when(mockCrdOps.updateStatusAsync(any(KafkaUser.class))).thenReturn(Future.succeededFuture()); Checkpoint async = context.checkpoint(); op.createOrUpdate(new Reconciliation("test-trigger", KafkaUser.RESOURCE_KIND, ResourceUtils.NAMESPACE, ResourceUtils.NAME), user) .onComplete(context.succeeding(v -> context.verify(() -> { List<String> capturedNames = secretNameCaptor.getAllValues(); assertThat(capturedNames, hasSize(1)); assertThat(capturedNames.get(0), is(ResourceUtils.NAME)); List<String> capturedNamespaces = secretNamespaceCaptor.getAllValues(); assertThat(capturedNamespaces, hasSize(1)); assertThat(capturedNamespaces.get(0), is(ResourceUtils.NAMESPACE)); List<Secret> capturedSecrets = secretCaptor.getAllValues(); assertThat(capturedSecrets, hasSize(1)); Secret captured = capturedSecrets.get(0); assertThat(captured.getMetadata().getName(), is(userCert.getMetadata().getName())); assertThat(captured.getMetadata().getNamespace(), is(userCert.getMetadata().getNamespace())); assertThat(captured.getMetadata().getLabels(), is(userCert.getMetadata().getLabels())); assertThat(new String(Base64.getDecoder().decode(captured.getData().get("ca.crt"))), is("different-clients-ca-crt")); assertThat(new String(Base64.getDecoder().decode(captured.getData().get("user.crt"))), is("crt file")); assertThat(new String(Base64.getDecoder().decode(captured.getData().get("user.key"))), is("key file")); async.flag(); }))); }