[DPE-2195] Add certificate_extra_sans config option

config.yaml

@@ -82,3 +82,7 @@ options:
      description: 'Profile representing the scope of deployment, and used to enable high-level customisation of sysconfigs, resource checks/allocation, warning levels, etc. Allowed values are: “production”, “staging” and “testing”'
      type: string
      default: production
+  certificate_extra_sans:
+     description: Config options 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.
+     type: string
+     default: ""

src/structured_config.py

@@ -62,6 +62,7 @@ class CharmConfig(BaseConfigModel):
     replication_quota_window_num: int
     zookeeper_ssl_cipher_suites: Optional[str]
     profile: str
+    certificate_extra_sans: Optional[str]
 
     @validator("*", pre=True)
     @classmethod

src/tls.py

@@ -24,7 +24,7 @@
     RelationJoinedEvent,
 )
 from ops.framework import Object
-from ops.model import ActiveStatus, BlockedStatus, Relation
+from ops.model import ActiveStatus, BlockedStatus
 
 from literals import TLS_RELATION, TRUSTED_CA_RELATION, TRUSTED_CERTIFICATE_RELATION
 from utils import (
@@ -90,10 +90,10 @@ def __init__(self, charm):
 
     def _tls_relation_created(self, _) -> None:
         """Handler for `certificates_relation_created` event."""
-        if not self.charm.unit.is_leader() or not self.peer_relation:
+        if not self.charm.unit.is_leader() or not self.charm.peer_relation:
             return
 
-        self.peer_relation.data[self.charm.app].update({"tls": "enabled"})
+        self.charm.app_peer_data.update({"tls": "enabled"})
 
     def _tls_relation_joined(self, _) -> None:
         """Handler for `certificates_relation_joined` event."""
@@ -238,7 +238,7 @@ def _trusted_relation_broken(self, event: RelationBrokenEvent) -> None:
 
     def _on_certificate_available(self, event: CertificateAvailableEvent) -> None:
         """Handler for `certificates_available` event after provider updates signed certs."""
-        if not self.peer_relation:
+        if not self.charm.peer_relation:
             logger.warning("No peer relation on certificate available")
             event.defer()
             return
@@ -259,12 +259,12 @@ def _on_certificate_available(self, event: CertificateAvailableEvent) -> None:
 
     def _on_certificate_expiring(self, _) -> None:
         """Handler for `certificate_expiring` event."""
-        if not self.private_key or not self.csr or not self.peer_relation:
+        if not self.private_key or not self.csr or not self.charm.peer_relation:
             logger.error("Missing unit private key and/or old csr")
             return
         new_csr = generate_csr(
             private_key=self.private_key.encode("utf-8"),
-            subject=self.peer_relation.data[self.charm.unit].get("private-address", ""),
+            subject=self.charm.unit_peer_data.get("private-address", ""),
             sans_ip=self._sans["sans_ip"],
             sans_dns=self._sans["sans_dns"],
         )
@@ -288,11 +288,6 @@ def _set_tls_private_key(self, event: ActionEvent) -> None:
 
         self._on_certificate_expiring(event)
 
-    @property
-    def peer_relation(self) -> Optional[Relation]:
-        """Get the peer relation of the charm."""
-        return self.charm.peer_relation
-
     @property
     def enabled(self) -> bool:
         """Flag to check if the cluster should run with TLS.
@@ -373,26 +368,40 @@ def truststore_password(self) -> Optional[str]:
 
     def _request_certificate(self):
         """Generates and submits CSR to provider."""
-        if not self.private_key or not self.peer_relation:
+        if not self.private_key or not self.charm.peer_relation:
             logger.error("Can't request certificate, missing private key")
             return
 
         csr = generate_csr(
             private_key=self.private_key.encode("utf-8"),
-            subject=self.peer_relation.data[self.charm.unit].get("private-address", ""),
+            subject=self.charm.unit_peer_data.get("private-address", ""),
             sans_ip=self._sans["sans_ip"],
             sans_dns=self._sans["sans_dns"],
         )
         self.charm.set_secret(scope="unit", key="csr", value=csr.decode("utf-8").strip())
 
         self.certificates.request_certificate_creation(certificate_signing_request=csr)
 
+    @property
+    def _extra_sans(self) -> List[str]:
+        """Parse the certificate_extra_sans config option."""
+        extra_sans = self.charm.config.certificate_extra_sans or ""
+        parsed_sans = []
+
+        if extra_sans == "":
+            return parsed_sans
+
+        for sans in extra_sans.split(","):
+            parsed_sans.append(sans.replace("{unit}", self.charm.unit.name.split("/")[1]))
+
+        return parsed_sans
+
     @property
     def _sans(self) -> Dict[str, List[str]]:
         """Builds a SAN dict of DNS names and IPs for the unit."""
         return {
             "sans_ip": [self.charm.unit_host],
-            "sans_dns": [self.charm.unit.name, socket.getfqdn()],
+            "sans_dns": [self.charm.unit.name, socket.getfqdn()] + self._extra_sans,
         }
 
     def generate_alias(self, app_name: str, relation_id: int) -> str:

tests/unit/test_tls.py

@@ -2,6 +2,7 @@
 # Copyright 2023 Canonical Ltd.
 # See LICENSE file for licensing details.
 
+import socket
 from pathlib import Path
 
 import pytest
@@ -80,3 +81,33 @@ def test_mtls_flag_added(harness: Harness):
     peer_relation_data = harness.get_relation_data(peer_relation_id, "kafka")
     assert peer_relation_data.get("mtls", "disabled") == "enabled"
     assert isinstance(harness.charm.app.status, ActiveStatus)
+
+
+def test_extra_sans_config(harness: Harness):
+    # Create peer relation
+    peer_relation_id = harness.add_relation(PEER, CHARM_KEY)
+    harness.add_relation_unit(peer_relation_id, "kafka/0")
+    harness.update_relation_data(peer_relation_id, "kafka/0", {"private-address": "treebeard"})
+
+    harness.update_config({"certificate_extra_sans": ""})
+    assert harness.charm.tls._extra_sans == []
+
+    harness.update_config({"certificate_extra_sans": "worker{unit}.com"})
+    assert harness.charm.tls._extra_sans == ["worker0.com"]
+
+    harness.update_config({"certificate_extra_sans": "worker{unit}.com,{unit}.example"})
+    assert harness.charm.tls._extra_sans == ["worker0.com", "0.example"]
+
+
+def test_sans(harness: Harness):
+    # Create peer relation
+    peer_relation_id = harness.add_relation(PEER, CHARM_KEY)
+    harness.add_relation_unit(peer_relation_id, "kafka/0")
+    harness.update_relation_data(peer_relation_id, "kafka/0", {"private-address": "treebeard"})
+    harness.update_config({"certificate_extra_sans": "worker{unit}.com"})
+
+    sock_dns = socket.getfqdn()
+    assert harness.charm.tls._sans == {
+        "sans_ip": ["treebeard"],
+        "sans_dns": ["kafka/0", sock_dns, "worker0.com"],
+    }