[DPE-9318] Config option for certificate extra sans

config.yaml

@@ -21,3 +21,12 @@ options:
     type: secret
     description: |
       A Juju secret URI of a secret containing the private key for client TLS certificates.
+
+  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

@@ -305,17 +305,23 @@ def _on_secret_changed(self, event: ops.SecretChangedEvent) -> None:
 
     def _on_config_changed(self, event: ops.ConfigChangedEvent) -> None:
         """Handle TLS related config changes."""
-        if not (secret_id := self.charm.config.get(TLS_CLIENT_PRIVATE_KEY_CONFIG)):
-            return
-
-        if not (private_key := self.charm.tls_manager.read_and_validate_private_key(secret_id)):
-            logger.error("Invalid private key provided, cannot update TLS certificates.")
+        if not self.charm.tls_manager.extra_sans_config_is_valid():
+            logger.warning("Invalid configuration for 'certificate-extra-sans'")
             return
 
-        if self.charm.unit.is_leader():
-            self.charm.state.cluster.update({"tls_client_private_key": private_key.raw})
+        if (secret_id := self.charm.config.get(TLS_CLIENT_PRIVATE_KEY_CONFIG)) and (
+            private_key := self.charm.tls_manager.read_and_validate_private_key(secret_id)
+        ):
+            if self.charm.unit.is_leader():
+                self.charm.state.cluster.update({"tls_client_private_key": private_key.raw})
+            # refresh event will be ignored by the tls lib if the csr is unchanged
+            self.refresh_tls_certificates_event.emit()
 
-        if self.charm.state.client_tls_relation:
+        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:

src/managers/tls.py

@@ -9,6 +9,7 @@
 import logging
 import re
 from datetime import timedelta
+from ipaddress import ip_address
 
 from charmlibs.interfaces.tls_certificates import (
     Certificate,
@@ -21,6 +22,7 @@
 from data_platform_helpers.advanced_statuses.protocol import ManagerStatusProtocol
 from data_platform_helpers.advanced_statuses.types import Scope
 from ops import ModelError, SecretNotFoundError
+from validators import ValidationError, hostname
 
 from common.exceptions import ValkeyWorkloadCommandError
 from core.base_workload import WorkloadBase
@@ -98,6 +100,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:
@@ -117,6 +125,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)
@@ -304,6 +322,96 @@ def start_ca_rotation_if_required(
         )
         return True
 
+    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] = []
@@ -344,4 +452,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

@@ -123,3 +123,8 @@ class TLSStatuses(Enum):
         status="blocked",
         message="The private key provided is not valid. Please provide a valid private key",
     )
+    SANS_CONFIG_INVALID = StatusObject(
+        status="blocked",
+        message="Invalid value for config option 'certificate-extra-sans'",
+        short_message="Invalid value `certificate-extra-sans`",
+    )

tests/integration/tls/test_certificate_options.py

@@ -0,0 +1,100 @@
+#!/usr/bin/env python3
+# Copyright 2026 Canonical Ltd.
+# See LICENSE file for licensing details.
+import logging
+import subprocess
+
+import jubilant
+
+from literals import Substrate
+from statuses import TLSStatuses
+from tests.integration.helpers import (
+    APP_NAME,
+    IMAGE_RESOURCE,
+    TLS_CERT_FILE,
+    TLS_CHANNEL,
+    TLS_NAME,
+    are_agents_idle,
+    are_apps_active_and_agents_idle,
+    does_status_match,
+    download_client_certificate_from_unit,
+)
+
+logger = logging.getLogger(__name__)
+
+NUM_UNITS = 3
+
+
+def test_build_and_deploy(charm: str, juju: jubilant.Juju, substrate: Substrate) -> None:
+    """Deploy the charm under test and a TLS provider."""
+    juju.deploy(
+        charm,
+        resources=IMAGE_RESOURCE if substrate == Substrate.K8S else None,
+        num_units=NUM_UNITS,
+        trust=True,
+    )
+    juju.deploy(TLS_NAME, channel=TLS_CHANNEL)
+    juju.integrate(f"{APP_NAME}:client-certificates", TLS_NAME)
+    juju.wait(
+        lambda status: are_agents_idle(status, APP_NAME, idle_period=30, unit_count=NUM_UNITS),
+        timeout=600,
+    )
+
+
+def test_extra_sans_config_option(juju: jubilant.Juju) -> None:
+    """Configure extra sans for the TLS certificates."""
+    logger.info("Set config to invalid sans value")
+    config_value = "-my.hostname"
+    juju.config(app=APP_NAME, values={"certificate-extra-sans": config_value})
+
+    juju.wait(
+        lambda status: does_status_match(
+            status,
+            expected_unit_statuses={APP_NAME: [TLSStatuses.SANS_CONFIG_INVALID.value]},
+            num_units={APP_NAME: NUM_UNITS},
+        ),
+        timeout=100,
+    )
+
+    download_client_certificate_from_unit(juju, APP_NAME)
+    client_cert_sans = subprocess.getoutput(
+        f"openssl x509 -noout -ext subjectAltName -in {TLS_CERT_FILE}"
+    )
+    assert config_value not in client_cert_sans, (
+        f"config value {config_value} found in certificate sans {client_cert_sans}"
+    )
+
+    logger.info("Configure valid extra-sans")
+    config_value = "server-{unit}.valkey"
+    juju.config(app=APP_NAME, values={"certificate-extra-sans": config_value})
+
+    juju.wait(
+        lambda status: are_apps_active_and_agents_idle(status, APP_NAME, unit_count=NUM_UNITS),
+        timeout=100,
+    )
+
+    # this will download the client cert from application.units[0]
+    download_client_certificate_from_unit(juju, APP_NAME)
+    client_cert_sans = subprocess.getoutput(
+        f"openssl x509 -noout -ext subjectAltName -in {TLS_CERT_FILE}"
+    )
+    unit_name = next(iter(juju.status().get_units(APP_NAME)))
+    expected_sans = config_value.replace("{unit}", unit_name.split("/")[-1])
+    assert expected_sans in client_cert_sans, (
+        f"expected sans {expected_sans} not found in certificate sans {client_cert_sans}"
+    )
+
+    logger.info("Resetting configuration for extra-sans")
+    juju.config(app=APP_NAME, reset="certificate-extra-sans")
+
+    juju.wait(
+        lambda status: are_apps_active_and_agents_idle(status, APP_NAME, unit_count=NUM_UNITS)
+    )
+
+    download_client_certificate_from_unit(juju, APP_NAME)
+    client_cert_sans = subprocess.getoutput(
+        f"openssl x509 -noout -ext subjectAltName -in {TLS_CERT_FILE}"
+    )
+    assert expected_sans not in client_cert_sans, (
+        f"sans value {expected_sans} found in certificate sans {client_cert_sans}"
+    )

tests/integration/tls/test_certificate_rotation.py

@@ -218,7 +218,7 @@ async def test_ca_rotation_by_expiration(juju: jubilant.Juju) -> None:
     juju.remove_relation(f"{APP_NAME}:client-certificates", f"{TLS_NAME}:certificates")
     juju.wait(
         lambda status: are_agents_idle(status, APP_NAME, idle_period=30, unit_count=NUM_UNITS),
-        timeout=100,
+        timeout=600,
     )
 
     logger.info("Adjust CA and certificate validity on TLS provider")

tests/spread/k8s/test_certificate_options.py/task.yaml

@@ -0,0 +1,7 @@
+summary: test_certificate_options.py
+environment:
+  TEST_MODULE: test_certificate_options.py
+execute: |
+  tox run -e integration -- "tests/integration/tls/$TEST_MODULE" --substrate k8s --alluredir="$SPREAD_TASK/allure-results"
+artifacts:
+  - allure-results

tests/spread/vm/test_certificate_options.py/task.yaml

@@ -0,0 +1,7 @@
+summary: test_certificate_options.py
+environment:
+  TEST_MODULE: test_certificate_options.py
+execute: |
+  tox run -e integration -- "tests/integration/tls/$TEST_MODULE" --substrate vm --alluredir="$SPREAD_TASK/allure-results"
+artifacts:
+  - allure-results