Add test to confirm service certificate refreshes do not disturb clie…

…nt connections (#6578)

tests/governance.py

@@ -512,6 +512,22 @@ def test_service_cert_renewal_extended(network, args):
         else:
             assert expected_exception is None, "Proposal should have not succeeded"
 
+    # Confirm that we can renew the service certificate multiple times
+    # without issue
+    for _ in range(0, 5):
+        new_duration = random.randint(1, 100)
+        LOG.info(f"Renewing service certificate for {new_duration} days")
+        renew_service_certificate(network, args, now, new_duration)
+        # Old cert still valid
+        primary, _ = network.find_primary()
+        with primary.client() as c:
+            c.get("/node/network/nodes")
+        # Update service identity used by clients to connect to the network
+        network.refresh_service_identity_file(args)
+        # Using new cert also works
+        with primary.client() as c:
+            c.get("/node/network/nodes")
+
     return network
 
 

tests/infra/network.py

@@ -22,6 +22,7 @@
 import pprint
 import functools
 import re
+import hashlib
 from datetime import datetime, timedelta, timezone
 from infra.consortium import slurp_file
 from infra.snp import IS_SNP
@@ -1645,6 +1646,25 @@ def verify_service_certificate_validity_period(self, expected_validity_days):
             f"Certificate validity period for service: {valid_from} - {valid_to} (for {validity_period})"
         )
 
+    def refresh_service_identity_file(self, args):
+        """
+        Refresh service_cert.pem from the current primary node, so that future client
+        connections pick up the new service certificate.
+        """
+        primary = self.find_random_node()
+        with primary.client() as c:
+            r = c.get("/node/network")
+            assert r.status_code == 200, r
+            new_service_identity = r.body.json()["service_certificate"]
+        identity_filepath = os.path.join(self.common_dir, "service_cert.pem")
+        with open(identity_filepath, "r", encoding="utf-8") as f:
+            before_digest = hashlib.sha256(f.read().encode()).hexdigest()
+        LOG.info(f"Before refresh, service_cert.pem was sha256:{before_digest}")
+        after_digest = hashlib.sha256(new_service_identity.encode()).hexdigest()
+        LOG.info(f"After refresh, service_cert.pem is sha256:{after_digest}")
+        with open(identity_filepath, "w", encoding="utf-8") as f:
+            f.write(new_service_identity)
+
     def save_service_identity(self, args):
         n = self.find_random_node()
         with n.client() as c: