Merge remote-tracking branch 'origin' into add-subordinate-support-wa…

…it_until

lib/charms/data_platform_libs/v0/data_interfaces.py

@@ -331,7 +331,7 @@ def _on_topic_requested(self, event: TopicRequestedEvent):
 
 # Increment this PATCH version before using `charmcraft publish-lib` or reset
 # to 0 if you are raising the major API version
-LIBPATCH = 39
+LIBPATCH = 40
 
 PYDEPS = ["ops>=2.0.0"]
 
@@ -391,6 +391,10 @@ class IllegalOperationError(DataInterfacesError):
     """To be used when an operation is not allowed to be performed."""
 
 
+class PrematureDataAccessError(DataInterfacesError):
+    """To be raised when the Relation Data may be accessed (written) before protocol init complete."""
+
+
 ##############################################################################
 # Global helpers / utilities
 ##############################################################################
@@ -1453,6 +1457,8 @@ def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
 class ProviderData(Data):
     """Base provides-side of the data products relation."""
 
+    RESOURCE_FIELD = "database"
+
     def __init__(
         self,
         model: Model,
@@ -1618,6 +1624,15 @@ def _fetch_my_specific_relation_data(
     def _update_relation_data(self, relation: Relation, data: Dict[str, str]) -> None:
         """Set values for fields not caring whether it's a secret or not."""
         req_secret_fields = []
+
+        keys = set(data.keys())
+        if self.fetch_relation_field(relation.id, self.RESOURCE_FIELD) is None and (
+            keys - {"endpoints", "read-only-endpoints", "replset"}
+        ):
+            raise PrematureDataAccessError(
+                "Premature access to relation data, update is forbidden before the connection is initialized."
+            )
+
         if relation.app:
             req_secret_fields = get_encoded_list(relation, relation.app, REQ_SECRET_FIELDS)
 
@@ -3290,6 +3305,8 @@ class KafkaRequiresEvents(CharmEvents):
 class KafkaProviderData(ProviderData):
     """Provider-side of the Kafka relation."""
 
+    RESOURCE_FIELD = "topic"
+
     def __init__(self, model: Model, relation_name: str) -> None:
         super().__init__(model, relation_name)
 
@@ -3539,6 +3556,8 @@ class OpenSearchRequiresEvents(CharmEvents):
 class OpenSearchProvidesData(ProviderData):
     """Provider-side of the OpenSearch relation."""
 
+    RESOURCE_FIELD = "index"
+
     def __init__(self, model: Model, relation_name: str) -> None:
         super().__init__(model, relation_name)
 

lib/charms/opensearch/v0/helper_conf_setter.py

@@ -279,23 +279,21 @@ def replace(
         with open(path, "r+") as f:
             data = f.read()
 
-            if regex and old_val and re.compile(old_val).match(data):
-                data = re.sub(r"{}".format(old_val), f"{new_val}", data)
-            elif old_val and old_val in data:
-                data = data.replace(old_val, new_val)
-            elif add_line_if_missing:
-                data += f"{data.rstrip()}\n{new_val}\n"
-
-            if output_type in [OutputType.console, OutputType.all]:
-                logger.info(data)
-
-            if output_type in [OutputType.file, OutputType.all]:
-                if output_file is None or output_file == config_file:
-                    f.seek(0)
-                    f.write(data)
-                else:
-                    with open(output_file, "w") as g:
-                        g.write(data)
+        if regex and old_val and re.compile(old_val, re.MULTILINE).findall(data):
+            data = re.sub(r"{}".format(old_val), f"{new_val}", data)
+        elif old_val and old_val in data:
+            data = data.replace(old_val, new_val)
+        elif add_line_if_missing:
+            data = f"{data.rstrip()}\n{new_val}\n"
+
+        if output_type in [OutputType.console, OutputType.all]:
+            logger.info(data)
+
+        if output_file is None:
+            output_file = config_file
+
+        with open(output_file, "w") as f:
+            f.write(data)
 
     @override
     def append(

lib/charms/opensearch/v0/opensearch_base_charm.py

@@ -642,6 +642,14 @@ def _on_update_status(self, event: UpdateStatusEvent):  # noqa: C901
         ):
             self.opensearch_provider.remove_lingering_relation_users_and_roles()
 
+        # If the unit reloads its certs but the other units are not ready yet
+        # we need to wait for them all to be ready before deleting the old CA
+        if (
+            self.tls.read_stored_ca("old-ca")
+            and self.tls.ca_and_certs_rotation_complete_in_cluster()
+        ):
+            logger.debug("update_status: Detected CA rotation complete in cluster")
+            self.tls.on_ca_certs_rotation_complete()
         # If relation not broken - leave
         if self.model.get_relation("certificates") is not None:
             return
@@ -818,14 +826,17 @@ def on_tls_conf_set(
                     logger.error("Could not reload TLS certificates via API, will restart.")
                     self._restart_opensearch_event.emit()
                 else:
-                    # the chain.pem file should only be updated after applying the new certs
-                    # otherwise there could be TLS verification errors after renewing the CA
-                    self.tls.update_request_ca_bundle()
                     self.status.clear(TLSNotFullyConfigured)
                     self.tls.reset_ca_rotation_state()
-                    # cleaning the former CA certificate from the truststore
-                    # must only be done AFTER all renewed certificates are available and loaded
-                    self.tls.remove_old_ca()
+                    # if all certs are stored and CA rotation is complete in the cluster
+                    # we delete the old ca and update the chain to only include the new one
+                    if (
+                        self.tls.read_stored_ca("old-ca")
+                        and self.tls.ca_and_certs_rotation_complete_in_cluster()
+                    ):
+                        logger.info("on_tls_conf_set: Detected CA rotation complete in cluster")
+                        self.tls.on_ca_certs_rotation_complete()
+
             else:
                 event.defer()
                 return
@@ -949,12 +960,6 @@ def _start_opensearch(self, event: _StartOpenSearch) -> None:  # noqa: C901
             return
 
         if not self._can_service_start():
-            # after rotating the CA and certificates:
-            # the last host in the cluster to restart might not be able to connect to the other
-            # hosts anymore, because it is the last to renew the pem-file for requests
-            # in this case we update the pem-file to be able to connect and start the host
-            if self.peers_data.get(Scope.UNIT, "tls_ca_renewed", False):
-                self.tls.update_request_ca_bundle()
             self.node_lock.release()
             logger.info("Could not start opensearch service. Will retry next event.")
             event.defer()
@@ -989,11 +994,6 @@ def _start_opensearch(self, event: _StartOpenSearch) -> None:  # noqa: C901
             self.unit.status = BlockedStatus(str(e))
             return
 
-        # we should update the chain.pem file to avoid TLS verification errors
-        # this happens on restarts after applying a new admin cert on CA rotation
-        if self.peers_data.get(Scope.UNIT, "tls_ca_renewed", False):
-            self.tls.update_request_ca_bundle()
-
         try:
             self.opensearch.start(
                 wait_until_http_200=(
@@ -1161,10 +1161,6 @@ def _post_start_init(self, event: _StartOpenSearch):  # noqa: C901
         if self.opensearch_peer_cm.is_provider():
             self.peer_cluster_provider.refresh_relation_data(event, can_defer=False)
 
-        # before resetting the CA rotation state, we remove the old ca from the truststore
-        if self.peers_data.get(Scope.UNIT, "tls_ca_renewed", False):
-            self.tls.remove_old_ca()
-
         # update the peer relation data for TLS CA rotation routine
         self.tls.reset_ca_rotation_state()
         if self.is_tls_full_configured_in_cluster():
@@ -1181,6 +1177,15 @@ def _post_start_init(self, event: _StartOpenSearch):  # noqa: C901
                 self.tls.request_new_admin_certificate()
             else:
                 self.tls.store_admin_tls_secrets_if_applies()
+        # If the reload through API failed, we restart the service
+        # We remove the old CA and update the chain to only include the new one
+        # if all certs are stored and CA rotation is complete in the cluster
+        if (
+            self.tls.read_stored_ca("old-ca")
+            and self.tls.ca_and_certs_rotation_complete_in_cluster()
+        ):
+            logger.info("post_start_init: Detected CA rotation complete in cluster")
+            self.tls.on_ca_certs_rotation_complete()
 
     def _stop_opensearch(self, *, restart: bool = False) -> None:
         """Stop OpenSearch if possible."""

lib/charms/opensearch/v0/opensearch_tls.py

@@ -21,6 +21,7 @@
 import tempfile
 import typing
 from os.path import exists
+from pathlib import Path
 from typing import Any, Dict, List, Optional, Tuple, Union
 
 from charms.opensearch.v0.constants_charm import (
@@ -227,7 +228,7 @@ def _on_certificate_available(self, event: CertificateAvailableEvent) -> None:
             merge=True,
         )
 
-        current_stored_ca = self._read_stored_ca()
+        current_stored_ca = self.read_stored_ca()
         if current_stored_ca != event.ca:
             if not self.store_new_ca(self.charm.secrets.get_object(scope, cert_type.val)):
                 logger.debug("Could not store new CA certificate.")
@@ -253,7 +254,7 @@ def _on_certificate_available(self, event: CertificateAvailableEvent) -> None:
 
         # apply the chain.pem file for API requests, only if the CA cert has not been updated
         admin_secrets = self.charm.secrets.get_object(Scope.APP, CertType.APP_ADMIN.val) or {}
-        if admin_secrets.get("chain") and not self._read_stored_ca(alias="old-ca"):
+        if admin_secrets.get("chain") and not self.read_stored_ca(alias="old-ca"):
             self.update_request_ca_bundle()
 
         # store the admin certificates in non-leader units
@@ -272,7 +273,7 @@ def _on_certificate_available(self, event: CertificateAvailableEvent) -> None:
         if self.charm.unit.is_leader() and self.charm.opensearch_peer_cm.is_provider(typ="main"):
             self.charm.peer_cluster_provider.refresh_relation_data(event, can_defer=False)
 
-        renewal = self._read_stored_ca(alias="old-ca") is not None or (
+        renewal = self.read_stored_ca(alias="old-ca") is not None or (
             old_cert is not None and old_cert != event.certificate
         )
 
@@ -556,9 +557,11 @@ def store_new_ca(self, secrets: Dict[str, Any]) -> bool:  # noqa: C901
                 logging.error(f"Error storing the ca-cert: {e}")
                 return False
 
+        self._add_ca_to_request_bundle(secrets.get("chain"))
+
         return True
 
-    def _read_stored_ca(self, alias: str = "ca") -> Optional[str]:
+    def read_stored_ca(self, alias: str = "ca") -> Optional[str]:
         """Load stored CA cert."""
         secrets = self.charm.secrets.get_object(Scope.APP, CertType.APP_ADMIN.val)
 
@@ -608,6 +611,8 @@ def remove_old_ca(self) -> None:
             if f"Alias <{old_alias}> does not exist" in e.out:
                 return
 
+        old_ca_content = self.read_stored_ca(alias=old_alias)
+
         run_cmd(
             f"""{keytool} \
             -delete \
@@ -617,6 +622,8 @@ def remove_old_ca(self) -> None:
             -storetype PKCS12"""
         )
         logger.info(f"Removed {old_alias} from truststore.")
+        # remove it from the request bundle
+        self._remove_ca_from_request_bundle(old_ca_content)
 
     def update_request_ca_bundle(self) -> None:
         """Create a new chain.pem file for requests module"""
@@ -689,7 +696,7 @@ def all_tls_resources_stored(self, only_unit_resources: bool = False) -> bool:
 
         # compare issuer of the cert with the issuer of the CA
         # if they don't match, certs are not up-to-date and need to be renewed after CA rotation
-        if not (current_ca := self._read_stored_ca()):
+        if not (current_ca := self.read_stored_ca()):
             return False
 
         # to make sure the content is processed correctly by openssl, temporary store it in a file
@@ -870,6 +877,45 @@ def ca_rotation_complete_in_cluster(self) -> bool:
 
         return rotation_complete
 
+    def ca_and_certs_rotation_complete_in_cluster(self) -> bool:
+        """Check whether the CA rotation completed in all units."""
+        rotation_complete = True
+
+        # the current unit is not in the relation.units list
+        if (
+            self.charm.peers_data.get(Scope.UNIT, "tls_ca_renewing")
+            or self.charm.peers_data.get(
+                Scope.UNIT,
+                "tls_ca_renewed",
+            )
+            or self.charm.peers_data.get(Scope.UNIT, "tls_configured") is not True
+        ):
+            logger.debug("TLS CA rotation ongoing on this unit.")
+            return False
+
+        for relation_type in [
+            PeerRelationName,
+            PeerClusterRelationName,
+            PeerClusterOrchestratorRelationName,
+        ]:
+            for relation in self.model.relations[relation_type]:
+                logger.debug(f"Checking relation {relation}: units: {relation.units}")
+                for unit in relation.units:
+                    if (
+                        "tls_ca_renewing" in relation.data[unit]
+                        or "tls_ca_renewed" in relation.data[unit]
+                        or relation.data[unit].get("tls_configured") != "True"
+                    ):
+                        logger.debug(
+                            f"TLS CA rotation not complete for unit {unit}: {relation} \
+                                | tls_ca_renewing: {relation.data[unit].get('tls_ca_renewing')} \
+                                | tls_ca_renewed: {relation.data[unit].get('tls_ca_renewed')} \
+                                | tls_configured: {relation.data[unit].get('tls_configured')}"
+                        )
+                        rotation_complete = False
+                        break
+        return rotation_complete
+
     def is_ca_rotation_ongoing(self) -> bool:
         """Check whether the CA rotation is currently in progress."""
         if (
@@ -884,10 +930,35 @@ def is_ca_rotation_ongoing(self) -> bool:
         return False
 
     def update_ca_rotation_flag_to_peer_cluster_relation(self, flag: str, operation: str) -> None:
-        """Add a CA rotation flag to all related peer clusters in large deployments."""
+        """Add or remove a CA rotation flag to all related peer clusters in large deployments."""
         for relation_type in [PeerClusterRelationName, PeerClusterOrchestratorRelationName]:
             for relation in self.model.relations[relation_type]:
                 if operation == "add":
                     relation.data[self.charm.unit][flag] = "True"
                 elif operation == "remove":
                     relation.data[self.charm.unit].pop(flag, None)
+
+    def on_ca_certs_rotation_complete(self) -> None:
+        """Handle the completion of CA rotation."""
+        logger.info("CA rotation completed. Deleting old CA and updating request bundle.")
+        self.remove_old_ca()
+        self.update_request_ca_bundle()
+
+    def _add_ca_to_request_bundle(self, ca_cert: str) -> None:
+        """Add the CA cert to the request bundle for the requests module."""
+        bundle_path = Path(self.certs_path) / "chain.pem"
+        if not bundle_path.exists():
+            return
+
+        bundle_content = bundle_path.read_text()
+        if ca_cert not in bundle_content:
+            bundle_path.write_text(f"{bundle_content}\n{ca_cert}")
+
+    def _remove_ca_from_request_bundle(self, ca_cert: str) -> None:
+        """Remove the CA cert from the request bundle for the requests module."""
+        bundle_path = Path(self.certs_path) / "chain.pem"
+        if not bundle_path.exists():
+            return
+
+        bundle_content = bundle_path.read_text()
+        bundle_path.write_text(bundle_content.replace(ca_cert, ""))