Fix CC not restarted when API secret changes

cluster-operator/src/main/java/io/strimzi/operator/cluster/model/CruiseControl.java

@@ -415,10 +415,11 @@ protected List<EnvVar> getEnvVars() {
     /**
      * Creates Cruise Control API auth usernames, passwords, and credentials file
      *
+     * @param passwordGenerator The password generator for API users
+     *
      * @return Map containing Cruise Control API auth credentials
      */
-    public static Map<String, String> generateCruiseControlApiCredentials() {
-        PasswordGenerator passwordGenerator = new PasswordGenerator(16);
+    public static Map<String, String> generateCruiseControlApiCredentials(PasswordGenerator passwordGenerator) {
         String apiAdminPassword = passwordGenerator.generate();
         String apiUserPassword = passwordGenerator.generate();
 
@@ -441,10 +442,13 @@ public static Map<String, String> generateCruiseControlApiCredentials() {
     /**
      * Generate the Secret containing the Cruise Control API auth credentials.
      *
+     * @param passwordGenerator The password generator for API users
+     * 
      * @return The generated Secret.
      */
-    public Secret generateApiSecret() {
-        return ModelUtils.createSecret(CruiseControlResources.apiSecretName(cluster), namespace, labels, ownerReference, generateCruiseControlApiCredentials(), Collections.emptyMap(), Collections.emptyMap());
+    public Secret generateApiSecret(PasswordGenerator passwordGenerator) {
+        return ModelUtils.createSecret(CruiseControlResources.apiSecretName(cluster), namespace, labels, ownerReference, 
+            generateCruiseControlApiCredentials(passwordGenerator), Collections.emptyMap(), Collections.emptyMap());
     }
 
     /**

cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/assembly/CruiseControlReconciler.java

@@ -20,11 +20,13 @@
 import io.strimzi.operator.cluster.model.KafkaVersion;
 import io.strimzi.operator.cluster.model.NodeRef;
 import io.strimzi.operator.cluster.operator.resource.ResourceOperatorSupplier;
+import io.strimzi.operator.common.Annotations;
 import io.strimzi.operator.common.Reconciliation;
 import io.strimzi.operator.common.ReconciliationLogger;
 import io.strimzi.operator.common.Util;
 import io.strimzi.operator.common.model.Ca;
 import io.strimzi.operator.common.model.Labels;
+import io.strimzi.operator.common.model.PasswordGenerator;
 import io.strimzi.operator.common.operator.resource.ConfigMapOperator;
 import io.strimzi.operator.common.operator.resource.DeploymentOperator;
 import io.strimzi.operator.common.operator.resource.NetworkPolicyOperator;
@@ -40,7 +42,6 @@
 import java.util.Map;
 import java.util.Set;
 
-
 /**
  * Class used for reconciliation of Cruise Control. This class contains both the steps of the Cruise Control
  * reconciliation pipeline and is also used to store the state between them.
@@ -63,12 +64,14 @@ public class CruiseControlReconciler {
     private final ServiceOperator serviceOperator;
     private final NetworkPolicyOperator networkPolicyOperator;
     private final ConfigMapOperator configMapOperator;
+    private final PasswordGenerator passwordGenerator;
 
     private boolean existingCertsChanged = false;
 
     private String serverConfigurationHash = "";
     private String capacityConfigurationHash = "";
-
+    private String apiSecretHash = "";
+
     /**
      * Constructs the Cruise Control reconciler
      *
@@ -81,6 +84,7 @@ public class CruiseControlReconciler {
      * @param kafkaBrokerStorage        A map with storage configuration used by the Kafka cluster and its broker pools
      * @param kafkaBrokerResources      A map with resource configuration used by the Kafka cluster and its broker pools
      * @param clusterCa                 The Cluster CA instance
+     * @param passwordGenerator         The password generator for API users
      */
     @SuppressWarnings({"checkstyle:ParameterNumber"})
     public CruiseControlReconciler(
@@ -92,7 +96,8 @@ public CruiseControlReconciler(
             Set<NodeRef> kafkaBrokerNodes,
             Map<String, Storage> kafkaBrokerStorage,
             Map<String, ResourceRequirements> kafkaBrokerResources,
-            ClusterCa clusterCa
+            ClusterCa clusterCa,
+            PasswordGenerator passwordGenerator
     ) {
         this.reconciliation = reconciliation;
         this.cruiseControl = CruiseControl.fromCrd(reconciliation, kafkaAssembly, versions, kafkaBrokerNodes, kafkaBrokerStorage, kafkaBrokerResources, supplier.sharedEnvironmentProvider);
@@ -109,6 +114,7 @@ public CruiseControlReconciler(
         this.serviceOperator = supplier.serviceOperations;
         this.networkPolicyOperator = supplier.networkPolicyOperator;
         this.configMapOperator = supplier.configMapOperations;
+        this.passwordGenerator = passwordGenerator;
     }
 
     /**
@@ -240,15 +246,17 @@ protected Future<Void> apiSecret() {
         if (cruiseControl != null) {
             return secretOperator.getAsync(reconciliation.namespace(), CruiseControlResources.apiSecretName(reconciliation.name()))
                     .compose(oldSecret -> {
-                        Secret newSecret = cruiseControl.generateApiSecret();
+                        Secret newSecret = cruiseControl.generateApiSecret(passwordGenerator);
 
                         if (oldSecret != null)  {
                             // The credentials should not change with every release
                             // So if the secret with credentials already exists, we re-use the values
                             // But we use the new secret to update labels etc. if needed
                             newSecret.setData(oldSecret.getData());
                         }
-
+
+                        this.apiSecretHash = ReconcilerUtils.hashSecretContent(newSecret, "password");
+
                         return secretOperator.reconcile(reconciliation, reconciliation.namespace(), CruiseControlResources.apiSecretName(reconciliation.name()), newSecret)
                                 .map((Void) null);
                     });
@@ -285,7 +293,8 @@ protected Future<Void> deployment(boolean isOpenShift, ImagePullPolicy imagePull
             podAnnotations.put(Ca.ANNO_STRIMZI_IO_CLUSTER_CA_KEY_GENERATION, String.valueOf(clusterCa.caKeyGeneration()));
             podAnnotations.put(CruiseControl.ANNO_STRIMZI_SERVER_CONFIGURATION_HASH, serverConfigurationHash);
             podAnnotations.put(CruiseControl.ANNO_STRIMZI_CAPACITY_CONFIGURATION_HASH, capacityConfigurationHash);
-
+            podAnnotations.put(Annotations.ANNO_STRIMZI_AUTH_HASH, apiSecretHash);
+
             Deployment deployment = cruiseControl.generateDeployment(podAnnotations, isOpenShift, imagePullPolicy, imagePullSecrets);
 
             return deploymentOperator

cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/assembly/KafkaAssemblyOperator.java

@@ -704,7 +704,8 @@ CruiseControlReconciler cruiseControlReconciler()   {
                     kafkaBrokerNodes,
                     kafkaBrokerStorage,
                     kafkaBrokerResources,
-                    clusterCa
+                    clusterCa,
+                    new PasswordGenerator(16)
             );
         }
 

cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/assembly/ReconcilerUtils.java

@@ -32,11 +32,15 @@
 import io.vertx.core.CompositeFuture;
 import io.vertx.core.Future;
 
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
 import java.util.ArrayList;
+import java.util.Base64;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Set;
+import java.util.regex.Pattern;
 
 import static io.strimzi.operator.common.Annotations.ANNO_STRIMZI_SERVER_CERT_HASH;
 
@@ -381,4 +385,34 @@
     public static boolean kraftEnabled(Kafka kafka) {
         return KafkaCluster.ENABLED_VALUE_STRIMZI_IO_KRAFT.equals(Annotations.stringAnnotation(kafka, Annotations.ANNO_STRIMZI_IO_KRAFT, "disabled").toLowerCase(Locale.ENGLISH));
     }
+
+    /**
+     * Creates a hash from Secret's content.
+     * @param secret Secret with data.
+     * @param keyword String used to find matching keys.
+     * @return Hash of the values whose keys match the keyword.
+     */
+    public static String hashSecretContent(Secret secret, String keyword) {
+        if (secret == null) {
+            return hash("Secret not found");
+        }
+        StringBuilder sb = new StringBuilder();
+        for (String key : secret.getData().keySet()) {
+            if (Pattern.compile(keyword, Pattern.CASE_INSENSITIVE).matcher(key).find()) {
+                sb.append(secret.getData().get(key));
+            }
+        }
+        return sb.toString().length() > 0 
+            ? hash(sb.toString()) 
+            : hash("Keyword not found");
+    }
+
+    private static String hash(String s) {
+        try {
+            byte[] hash = MessageDigest.getInstance("MD5").digest(s.getBytes());
+            return Base64.getEncoder().encodeToString(hash);
+        } catch (NoSuchAlgorithmException e) {
+            throw new RuntimeException(e);
+        }
+    }
 }

cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/assembly/CruiseControlReconcilerTest.java

@@ -25,6 +25,7 @@
 import io.strimzi.operator.cluster.model.KafkaVersion;
 import io.strimzi.operator.cluster.model.NodeRef;
 import io.strimzi.operator.cluster.operator.resource.ResourceOperatorSupplier;
+import io.strimzi.operator.common.Annotations;
 import io.strimzi.operator.common.Reconciliation;
 import io.strimzi.operator.common.model.PasswordGenerator;
 import io.strimzi.operator.common.operator.MockCertManager;
@@ -53,6 +54,7 @@
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
 @ExtendWith(VertxExtension.class)
@@ -78,14 +80,16 @@ public void reconcileEnabledCruiseControl(VertxTestContext context) {
         ServiceOperator mockServiceOps = supplier.serviceOperations;
         NetworkPolicyOperator mockNetPolicyOps = supplier.networkPolicyOperator;
         ConfigMapOperator mockCmOps = supplier.configMapOperations;
-
+        PasswordGenerator mockPasswordGenerator = mock(PasswordGenerator.class);
+
         ArgumentCaptor<ServiceAccount> saCaptor = ArgumentCaptor.forClass(ServiceAccount.class);
         when(mockSaOps.reconcile(any(), eq(NAMESPACE), eq(CruiseControlResources.serviceAccountName(NAME)), saCaptor.capture())).thenReturn(Future.succeededFuture());
-
+
+        when(mockPasswordGenerator.generate()).thenReturn("secret");
         ArgumentCaptor<Secret> secretCaptor = ArgumentCaptor.forClass(Secret.class);
         when(mockSecretOps.reconcile(any(), eq(NAMESPACE), eq(CruiseControlResources.secretName(NAME)), secretCaptor.capture())).thenReturn(Future.succeededFuture());
         when(mockSecretOps.reconcile(any(), eq(NAMESPACE), eq(CruiseControlResources.apiSecretName(NAME)), secretCaptor.capture())).thenReturn(Future.succeededFuture());
-
+        
         ArgumentCaptor<Service> serviceCaptor = ArgumentCaptor.forClass(Service.class);
         when(mockServiceOps.reconcile(any(), eq(NAMESPACE), eq(CruiseControlResources.serviceName(NAME)), serviceCaptor.capture())).thenReturn(Future.succeededFuture());
 
@@ -124,7 +128,8 @@ public void reconcileEnabledCruiseControl(VertxTestContext context) {
                 NODES,
                 Map.of("kafka", kafka.getSpec().getKafka().getStorage()),
                 Map.of(),
-                clusterCa
+                clusterCa,
+                mockPasswordGenerator
         );
 
         Checkpoint async = context.checkpoint();
@@ -151,6 +156,7 @@ public void reconcileEnabledCruiseControl(VertxTestContext context) {
                     assertThat(deployCaptor.getValue(), is(notNullValue()));
                     assertThat(deployCaptor.getAllValues().get(0).getSpec().getTemplate().getMetadata().getAnnotations().get(CruiseControl.ANNO_STRIMZI_SERVER_CONFIGURATION_HASH), is("f6dc41c7"));
                     assertThat(deployCaptor.getAllValues().get(0).getSpec().getTemplate().getMetadata().getAnnotations().get(CruiseControl.ANNO_STRIMZI_CAPACITY_CONFIGURATION_HASH), is("1eb49220"));
+                    assertThat(deployCaptor.getAllValues().get(0).getSpec().getTemplate().getMetadata().getAnnotations().get(Annotations.ANNO_STRIMZI_AUTH_HASH), is("bLEULIBSrWikUZUQPf7GZQ=="));
 
                     async.flag();
                 })));
@@ -207,7 +213,8 @@ public void reconcileDisabledCruiseControl(VertxTestContext context) {
                 NODES,
                 Map.of(NAME + "-kafka", kafka.getSpec().getKafka().getStorage()),
                 Map.of(),
-                clusterCa
+                clusterCa,
+                new PasswordGenerator(16)
         );
 
         Checkpoint async = context.checkpoint();

cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/assembly/ReconcilerUtilsTest.java

@@ -290,6 +290,17 @@ public void testEnabledJmxWithAuthWithExistingSecret(VertxTestContext context) {
                     async.flag();
                 })));
     }
+
+    @Test
+    public void testHashSecretContent() {
+        Secret secret = new SecretBuilder()
+            .addToData(Map.of("user-password", "changeit"))
+            .addToData(Map.of("userPassword", "changeit"))
+            .addToData(Map.of("PASSWORD", "changeit"))
+            .build();
+
+        assertThat(ReconcilerUtils.hashSecretContent(secret, "password"), is("F9yAQqlPifM1mCBESPQ7hA=="));
+    }
 
     static class MockJmxCluster implements SupportsJmx {
         private final JmxModel jmx;

cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/resource/cruisecontrol/MockCruiseControl.java

@@ -12,6 +12,7 @@
 import io.strimzi.operator.cluster.model.CruiseControl;
 import io.strimzi.operator.cluster.model.ModelUtils;
 import io.strimzi.operator.common.model.Labels;
+import io.strimzi.operator.common.model.PasswordGenerator;
 import io.strimzi.operator.common.model.cruisecontrol.CruiseControlEndpoints;
 import io.strimzi.operator.common.model.cruisecontrol.CruiseControlParameters;
 import io.strimzi.operator.common.operator.MockCertManager;
@@ -79,7 +80,7 @@ public class MockCruiseControl {
             .addToData("cruise-control.crt", MockCertManager.clusterCaCert())
             .build();
     public static final Secret CC_API_SECRET = ModelUtils.createSecret(CruiseControlResources.apiSecretName(CLUSTER), NAMESPACE, Labels.EMPTY, null,
-            CruiseControl.generateCruiseControlApiCredentials(), Collections.emptyMap(), Collections.emptyMap());
+            CruiseControl.generateCruiseControlApiCredentials(new PasswordGenerator(16)), Collections.emptyMap(), Collections.emptyMap());
 
     private static final Header AUTH_HEADER = convertToHeader(CruiseControlApiImpl.getAuthHttpHeader(true, CC_API_SECRET));