[DPE-3822] handle invalidated certificates (#318)

## Issue

#204

## Solution
- implement handler for the `on.certificate_invalidated` event
- request new certificate for the invalidated one
- add unit test coverage in
`tests/unit/lib/test_opensearch_tls.py:test_on_certificate_invalidated`

lib/charms/opensearch/v0/opensearch_tls.py

@@ -18,7 +18,7 @@
 import re
 import socket
 import typing
-from typing import Dict, List, Optional, Tuple
+from typing import Dict, List, Optional, Tuple, Union
 
 from charms.opensearch.v0.constants_tls import TLS_RELATION, CertType
 from charms.opensearch.v0.helper_networking import get_host_public_ip
@@ -28,6 +28,7 @@
 from charms.tls_certificates_interface.v3.tls_certificates import (
     CertificateAvailableEvent,
     CertificateExpiringEvent,
+    CertificateInvalidatedEvent,
     TLSCertificatesRequiresV3,
     generate_csr,
     generate_private_key,
@@ -74,6 +75,9 @@ def __init__(self, charm: "OpenSearchBaseCharm", peer_relation: str):
 
         self.framework.observe(self.certs.on.certificate_available, self._on_certificate_available)
         self.framework.observe(self.certs.on.certificate_expiring, self._on_certificate_expiring)
+        self.framework.observe(
+            self.certs.on.certificate_invalidated, self._on_certificate_invalidated
+        )
 
     def _on_set_tls_private_key(self, event: ActionEvent) -> None:
         """Set the TLS private key, which will be used for requesting the certificate."""
@@ -186,7 +190,9 @@ def _on_certificate_available(self, event: CertificateAvailableEvent) -> None:
             logger.exception(e)
             event.defer()
 
-    def _on_certificate_expiring(self, event: CertificateExpiringEvent) -> None:
+    def _on_certificate_expiring(
+        self, event: Union[CertificateExpiringEvent, CertificateInvalidatedEvent]
+    ) -> None:
         """Request the new certificate when old certificate is expiring."""
         self.charm.peers_data.delete(Scope.UNIT, "tls_configured")
         try:
@@ -198,6 +204,11 @@ def _on_certificate_expiring(self, event: CertificateExpiringEvent) -> None:
 
         self._request_certificate_renewal(scope, cert_type, secrets)
 
+    def _on_certificate_invalidated(self, event: CertificateInvalidatedEvent) -> None:
+        """Handle a cert that was revoked or has expired"""
+        logger.debug(f"Received certificate invalidation. Reason: {event.reason}")
+        self._on_certificate_expiring(event)
+
     def _request_certificate(
         self,
         scope: Scope,

tests/unit/lib/test_opensearch_tls.py

@@ -240,3 +240,37 @@ def test_on_certificate_expiring(self, _, deployment_desc, request_certificate_c
         self.charm.tls._on_certificate_expiring(event_mock)
 
         request_certificate_creation.assert_called_once()
+
+    @patch(
+        "charms.tls_certificates_interface.v3.tls_certificates.TLSCertificatesRequiresV3.request_certificate_renewal"
+    )
+    @patch(
+        f"{BASE_LIB_PATH}.opensearch_peer_clusters.OpenSearchPeerClustersManager.deployment_desc"
+    )
+    @patch("charm.OpenSearchOperatorCharm._put_or_update_internal_user_leader")
+    def test_on_certificate_invalidated(self, _, deployment_desc, request_certificate_renewal):
+        """Test _on_certificate_invalidated event."""
+        csr = "csr_12345"
+        cert = "cert_12345"
+        key = create_utf8_encoded_private_key()
+        secret_key = CertType.UNIT_TRANSPORT.val
+
+        self.secret_store.put_object(
+            Scope.UNIT,
+            secret_key,
+            {"csr": csr, "cert": cert, "key": key},
+        )
+
+        deployment_desc.return_value = DeploymentDescription(
+            config=PeerClusterConfig(cluster_name="", init_hold=False, roles=[]),
+            start=StartMode.WITH_GENERATED_ROLES,
+            pending_directives=[],
+            typ=DeploymentType.MAIN_ORCHESTRATOR,
+            app=self.charm.app.name,
+            state=DeploymentState(value=State.ACTIVE),
+        )
+
+        event_mock = MagicMock(certificate=cert)
+        self.charm.tls._on_certificate_invalidated(event_mock)
+
+        request_certificate_renewal.assert_called_once()