Add refresh_events to CertHandler

lib/charms/observability_libs/v1/cert_handler.py

@@ -59,15 +59,15 @@
 import logging
 
 from ops.charm import CharmBase
-from ops.framework import EventBase, EventSource, Object, ObjectEvents
+from ops.framework import BoundEvent, EventBase, EventSource, Object, ObjectEvents
 from ops.jujuversion import JujuVersion
 from ops.model import Relation, Secret, SecretNotFoundError
 
 logger = logging.getLogger(__name__)
 
 LIBID = "b5cd5cd580f3428fa5f59a8876dcbe6a"
 LIBAPI = 1
-LIBPATCH = 11
+LIBPATCH = 12
 
 VAULT_SECRET_LABEL = "cert-handler-private-vault"
 
@@ -283,6 +283,7 @@ def __init__(
         peer_relation_name: str = "peers",
         cert_subject: Optional[str] = None,
         sans: Optional[List[str]] = None,
+        refresh_events: List[BoundEvent] = [],
     ):
         """CertHandler is used to wrap TLS Certificates management operations for charms.
 
@@ -299,6 +300,10 @@ def __init__(
                 Must match metadata.yaml.
             cert_subject: Custom subject. Name collisions are under the caller's responsibility.
             sans: DNS names. If none are given, use FQDN.
+            refresh_events: an optional list of bound events which
+                will be observed to replace the current CSR with a new one
+                if there are changes in the CSR's DNS SANs, IP SANs, subject, or private key.
+                Then, subsequently, replace its corresponding certificate with a new one.
         """
         super().__init__(charm, key)
         self.charm = charm
@@ -355,6 +360,26 @@ def __init__(
             self._on_upgrade_charm,
         )
 
+        for ev in refresh_events:
+            self.framework.observe(ev, self._on_refresh_event)
+
+    def _on_refresh_event(self, _):
+        """Replace the latest current CSR with a new one if there are any CSR changes.
+
+        The following CSR changes will trigger a certificate renewal: DNS SANs, IP SANs, subject, and private key changes.
+        Instead of individually comparing the new values of each field with those from the current CSR, we will compare the
+        entire current CSR with a newly generated one, populated with the latest values, to determine if renewal is needed.
+        """
+        curr_csr = self._csr.encode() if self._csr else None
+        new_csr = generate_csr(
+            private_key=self.private_key.encode(),
+            subject=self.cert_subject,
+            sans_dns=self.sans_dns,
+            sans_ip=self.sans_ip,
+        )
+        if curr_csr is not None and curr_csr != new_csr:
+            self._generate_csr(renew=True)
+
     def _on_upgrade_charm(self, _):
         has_privkey = self.vault.get_value("private-key")
 

tests/unit/test_kubernetes_compute_resources.py

@@ -2,9 +2,10 @@
 # See LICENSE file for licensing details.
 import unittest
 from unittest import mock
-from unittest.mock import MagicMock, Mock
+from unittest.mock import MagicMock, Mock, patch
 
 import httpx
+import tenacity
 import yaml
 from charms.observability_libs.v0.kubernetes_compute_resources_patch import (
     KubernetesComputeResourcesPatch,
@@ -16,12 +17,22 @@
 from ops import BlockedStatus, WaitingStatus
 from ops.charm import CharmBase
 from ops.testing import Harness
+from pytest import fixture
 
 from tests.unit.helpers import PROJECT_DIR
 
 CL_PATH = "charms.observability_libs.v0.kubernetes_compute_resources_patch.KubernetesComputeResourcesPatch"
 
 
+@fixture(autouse=True)
+def patch_retry():
+    with patch.multiple(
+        KubernetesComputeResourcesPatch,
+        PATCH_RETRY_STOP=tenacity.stop_after_delay(0),
+    ):
+        yield
+
+
 class TestKubernetesComputeResourcesPatch(unittest.TestCase):
     class _TestCharm(CharmBase):
         def __init__(self, *args):