[DPE-9663] TLS relation error field

config.yaml

@@ -16,3 +16,11 @@ options:
       - charmed-sentinel-operator: administrative user for operating valkey sentinel
       - charmed-sentinel-valkey: used by sentinel to authenticate against valkey
       - charmed-sentinel-peers: used by sentinel for inter-sentinel authentication
+
+  certificate-extra-sans:
+    type: string
+    description: |
+      Config option to add extra-sans to the ones used when requesting server certificates.
+      The extra-sans are specified by comma-separated names to be added when requesting signed certificates.
+      Use "{unit}" as a placeholder to be filled with the unit number, e.g. "worker-{unit}" will be translated as 
+      "worker-0" for unit 0 and "worker-1" for unit 1 when requesting the certificate.

pyproject.toml

@@ -13,6 +13,7 @@ charmlibs-snap = "^1.0.1"
 charmlibs-interfaces-tls-certificates = ">=1.6.0"
 tenacity = "*"
 data-platform-helpers = ">=0.1.7"
+validators = ">=0.35.0"
 
 [tool.poetry.requires-plugins]
 poetry-plugin-export = ">=1.8"

src/events/tls.py

@@ -10,6 +10,7 @@
 import ops
 from charmlibs.interfaces.tls_certificates import (
     CertificateAvailableEvent,
+    CertificateDeniedEvent,
     CertificateRequestAttributes,
     TLSCertificatesRequiresV4,
 )
@@ -71,13 +72,17 @@ def __init__(self, charm: "ValkeyCharm"):
         self.framework.observe(
             self.client_certificate.on.certificate_available, self._on_certificate_available
         )
+        self.framework.observe(
+            self.client_certificate.on.certificate_denied, self._on_certificate_denied
+        )
         self.framework.observe(
             self.charm.on[PEER_RELATION].relation_created, self._on_peer_relation_created
         )
         self.framework.observe(
             self.charm.on[PEER_RELATION].relation_changed, self._on_peer_relation_changed
         )
         self.framework.observe(self.charm.on.update_status, self._on_update_status)
+        self.framework.observe(self.charm.on.config_changed, self._on_config_changed)
 
     def _on_peer_relation_created(self, event: ops.RelationCreatedEvent) -> None:
         """Set up self-signed certificates for peer TLS by default."""
@@ -209,6 +214,14 @@ def _on_certificate_available(self, event: CertificateAvailableEvent) -> None:
             event.defer()
             return
 
+    def _on_certificate_denied(self, event: CertificateDeniedEvent) -> None:
+        """Handle the `certificate-denied` event from TLS provider."""
+        if event.certificate_signing_request in [
+            csr.certificate_signing_request
+            for csr in self.client_certificate.get_csrs_from_requirer_relation_data()
+        ]:
+            logger.error("Certificate request was denied: %s", event.error.message)
+
     def _on_tls_relation_broken(self, event: ops.RelationBrokenEvent) -> None:
         """Handle the `relation-broken` event."""
         if self.charm.app.planned_units() == 0 or self.charm.state.unit_server.is_being_removed:
@@ -282,6 +295,19 @@ def _on_update_status(self, event: ops.UpdateStatusEvent) -> None:
             logger.debug("Trigger peer relation change to orchestrate certificate/CA rotation")
             self.charm.on[PEER_RELATION].relation_changed.emit(self.charm.state.peer_relation)
 
+    def _on_config_changed(self, event: ops.ConfigChangedEvent) -> None:
+        """Handle TLS related config changes."""
+        if not self.charm.tls_manager.extra_sans_config_is_valid():
+            logger.warning("Invalid configuration for 'certificate-extra-sans'")
+            return
+
+        if (
+            self.charm.state.client_tls_relation
+            and self.charm.tls_manager.certificate_sans_require_update()
+        ):
+            logger.info("Configuration change for TLS, refresh TLS certificates")
+            self.refresh_tls_certificates_event.emit()
+
     def _enable_client_tls(self) -> None:
         """Check preconditions and enable TLS if possible."""
         if not all(server.model.client_cert_ready for server in self.charm.state.servers):

src/managers/tls.py

@@ -6,6 +6,7 @@
 
 import logging
 from datetime import timedelta
+from ipaddress import ip_address
 
 from charmlibs.interfaces.tls_certificates import (
     Certificate,
@@ -17,6 +18,7 @@
 from data_platform_helpers.advanced_statuses.models import StatusObject
 from data_platform_helpers.advanced_statuses.protocol import ManagerStatusProtocol
 from data_platform_helpers.advanced_statuses.types import Scope
+from validators import ValidationError, hostname
 
 from common.exceptions import ValkeyWorkloadCommandError
 from core.base_workload import WorkloadBase
@@ -94,6 +96,12 @@ def build_sans_ip(self) -> frozenset[str]:
         if not self.state.peer_relation:
             return frozenset(sans_ip)
 
+        if self.extra_sans_config_is_valid() and (
+            extra_sans_config := self.state.config.get("certificate-extra-sans")
+        ):
+            extra_sans = [san.strip() for san in extra_sans_config.split(",")]
+            sans_ip = {san for san in extra_sans if self._is_ip_address(san)}
+
         sans_ip.add(self.state.bind_address)
 
         if ingress_ip := self.state.ingress_address:
@@ -113,6 +121,16 @@ def build_sans_dns(self) -> frozenset[str]:
         if not self.state.peer_relation:
             return frozenset(sans_dns)
 
+        if self.extra_sans_config_is_valid() and (
+            extra_sans_config := self.state.config.get("certificate-extra-sans")
+        ):
+            extra_sans = [san.strip() for san in extra_sans_config.split(",")]
+            sans_dns = {
+                san.replace("{unit}", str(self.state.unit_server.unit_id))
+                for san in extra_sans
+                if not self._is_ip_address(san)
+            }
+
         sans_dns.add(self.state.unit_server.model.hostname)
 
         return frozenset(sans_dns)
@@ -246,14 +264,109 @@ def start_ca_rotation_if_required(
         )
         return True
 
-    def get_statuses(self, scope: Scope, recompute: bool = False) -> list[StatusObject]:
+    def extra_sans_config_is_valid(self) -> bool:
+        """Validate configuration value for certificate-extra-sans option.
+
+        Returns:
+            bool: True if config value is valid, False if invalid.
+        """
+        if not (extra_sans_config := self.state.config.get("certificate-extra-sans")):
+            return True
+
+        extra_sans = [san.strip() for san in extra_sans_config.split(",")]
+
+        for san in extra_sans:
+            if not self._is_ip_address(san):
+                if not self._is_hostname(
+                    san.replace("{unit}", str(self.state.unit_server.unit_id))
+                ):
+                    logger.error(f"certificate-extra-sans configuration is invalid for {san}")
+                    return False
+
+        return True
+
+    def _is_ip_address(self, input_value: str) -> bool:
+        """Validate a given str and return True if it is an IP address, False if not."""
+        try:
+            ip_address(input_value)
+            return True
+        except ValueError:
+            return False
+
+    def _is_hostname(self, input_value: str) -> bool:
+        """Validate a given str and return True if it is a hostname, False if not."""
+        try:
+            # Hostname string may only be hyphens and alpha-numerals.
+            return hostname(
+                input_value,
+                skip_ipv4_addr=True,
+                skip_ipv6_addr=True,
+                may_have_port=False,
+                maybe_simple=True,
+            )
+        except ValidationError:
+            return False
+
+    def get_current_sans(self) -> dict[str, set[str]]:
+        """Get the current SANs for a unit's cert."""
+        cert_file = self.workload.tls_paths.client_cert
+
+        sans_ip = set()
+        sans_dns = set()
+        if not (
+            san_lines := self.workload.exec(
+                [
+                    "openssl",
+                    "x509",
+                    "-ext",
+                    "subjectAltName",
+                    "-noout",
+                    "-in",
+                    cert_file.as_posix(),
+                ]
+            )[0].splitlines()
+        ):
+            return {"sans_ip": sans_ip, "sans_dns": sans_dns}
+
+        for line in san_lines:
+            for sans in line.split(", "):
+                san_type, san_value = sans.split(":")
+
+                if san_type.strip() == "DNS":
+                    sans_dns.add(san_value)
+                if san_type.strip() == "IP Address":
+                    sans_ip.add(san_value)
+
+        return {"sans_ip": sans_ip, "sans_dns": sans_dns}
+
+    def certificate_sans_require_update(self) -> bool:
+        """Check current certificate sans and determine if certificate requires update.
+
+        Returns:
+            bool: True if certificate sans have changed, False if they are still the same.
+        """
+        current_sans = self.get_current_sans()
+        new_sans_ip = self.build_sans_ip()
+        new_sans_dns = self.build_sans_dns()
+
+        if new_sans_ip ^ current_sans["sans_ip"] or new_sans_dns ^ current_sans["sans_dns"]:
+            return True
+
+        return False
+
+    def get_statuses(self, scope: Scope, recompute: bool = False) -> list[StatusObject]:  # noqa: C901
         """Compute the TLS statuses."""
         status_list: list[StatusObject] = []
 
         # Peer relation not established yet, or model not built yet for unit or app
         if not self.state.cluster.model or not self.state.unit_server.model:
             return status_list or [CharmStatuses.ACTIVE_IDLE.value]
 
+        if (relation := self.state.client_tls_relation) and relation.data[relation.app].get(
+            "request_errors"
+        ):
+            status_list.append(TLSStatuses.CERTIFICATE_DENIED.value)
+
         if self.state.unit_server.tls_client_state == TLSState.TO_TLS:
             status_list.append(TLSStatuses.ENABLING_CLIENT_TLS.value)
 
@@ -278,4 +391,7 @@ def get_statuses(self, scope: Scope, recompute: bool = False) -> list[StatusObje
         if self.state.unit_server.model.tls_certificate_expiring:
             status_list.append(TLSStatuses.CERTIFICATE_EXPIRING.value)
 
+        if not self.extra_sans_config_is_valid():
+            status_list.append(TLSStatuses.SANS_CONFIG_INVALID.value)
+
         return status_list if status_list else [CharmStatuses.ACTIVE_IDLE.value]

src/statuses.py

@@ -116,3 +116,11 @@ class TLSStatuses(Enum):
     CA_ROTATION_UPDATED = StatusObject(
         status="maintenance", message="TLS CA rotation: certificates updated"
     )
+    SANS_CONFIG_INVALID = StatusObject(
+        status="blocked",
+        message="Invalid value for config option 'certificate-extra-sans'",
+        short_message="Invalid value `certificate-extra-sans`",
+    )
+    CERTIFICATE_DENIED = StatusObject(
+        status="blocked", message="Certificate request was denied, check logs for details"
+    )