feat(tls): add substrate-aware CharmState TLS SAN/CN accessors

Layer the certificate-SAN and common-name policy onto CharmState, on top of the raw peer-databag accessors from the previous branch, so the substrate-specific certificate identity is reviewable as a unit with its own tests before any manager or handler consumes it.

K8s must regain the parity the migration had dropped: common_hosts has to advertise the primary/replicas Service FQDNs and the resolved pod FQDN, and the operator-cert common name has to be the endpoints FQDN (wildcarded past the 64-char CN limit) rather than the VM-style host/address; the peer SAN set must exclude the ip key the original K8s charm never emitted. VM behaviour is left host/address-derived as before. The CharmState charm parameter is also widened to ops.CharmBase so the state object no longer depends on the concrete charm type.

These accessors are additive and only read state, so the existing charm keeps constructing unchanged.

Signed-off-by: Marcelo Henrique Neppel <marcelo.neppel@canonical.com>

Author: marceloneppel

single_kernel_postgresql/core/state.py

@@ -8,11 +8,12 @@
 import socket
 from contextlib import suppress
 from functools import cached_property
-from typing import TYPE_CHECKING, Any, get_args
+from typing import Any, get_args
 
 from data_platform_helpers.advanced_statuses import StatusesState, StatusObject
 from data_platform_helpers.advanced_statuses.types import Scope as AdvancedStatusesScope
 from ops import (
+    CharmBase,
     ConfigData,
     JujuVersion,
     ModelError,
@@ -43,16 +44,13 @@
 from single_kernel_postgresql.utils.secret import translate_field_to_secret_key
 from single_kernel_postgresql.utils.status import format_status
 
-if TYPE_CHECKING:
-    from single_kernel_postgresql.charms.abstract_charm import AbstractPostgreSQLCharm
-
 
 class CharmState(Object):
     """The global PostgreSQL Charm State."""
 
     def __init__(
         self,
-        charm: "AbstractPostgreSQLCharm",
+        charm: CharmBase,
         substrate: Substrates,
     ) -> None:
         """Initialize the CharmState object."""
@@ -245,13 +243,64 @@ def host(self) -> str:
     @property
     def common_hosts(self) -> set[str]:
         """Common hosts to be used in TLS certificate SANs."""
-        return {self.host, self.fqdn} if self.fqdn else {self.host}
+        hosts = {self.host, self.fqdn} if self.fqdn else {self.host}
+        if self.substrate == Substrates.K8S:
+            namespace = self.model.name
+            hosts |= {
+                f"{self.model.app.name}-primary.{namespace}.svc.cluster.local",
+                f"{self.model.app.name}-replicas.{namespace}.svc.cluster.local",
+                # the original K8s charm also included the resolved per-pod FQDN.
+                socket.getfqdn(),
+            }
+        return hosts
 
     @property
     def peer_common_name(self) -> str:
         """Return the common name for the internally generated peer certificate."""
+        if self.substrate == Substrates.K8S:
+            return self._k8s_cert_common_name
         return self.peer.database_peers_address or self.host
 
+    @property
+    def client_addresses(self) -> set[str]:
+        """Client-facing addresses for the operator client certificate SANs."""
+        addrs: set[str] = set()
+        if addr := self.peer.database_address:
+            addrs.add(addr)
+        return addrs
+
+    @property
+    def client_common_name(self) -> str:
+        """Common name for the operator client certificate."""
+        if self.substrate == Substrates.K8S:
+            return self._k8s_cert_common_name
+        return self.peer.database_address or self.host
+
+    @property
+    def _k8s_cert_common_name(self) -> str:
+        """K8s operator-cert CN: the unit endpoints FQDN, wildcarded if too long.
+
+        Matches the original K8s charm: ``<app>-<unit_id>.<app>-endpoints``, collapsing to
+        ``*.<app>-endpoints`` when the full FQDN exceeds the 64-char CN limit. The
+        migration had switched this to the VM-style ``database_address/peers_address or
+        host``; restore the endpoints-FQDN CN for K8s parity.
+        """
+        full = self._get_hostname_from_unit(unit_name_to_pod_name(self.model.unit.name))
+        if len(full) > 64:
+            return f"*.{self.model.app.name}-endpoints"
+        return full
+
+    @property
+    def peer_addresses(self) -> set[str]:
+        """Peer addresses for the operator peer certificate SANs (substrate-aware).
+
+        K8s excludes the ``ip`` databag key (original K8s charm never added an ``ip`` SAN);
+        VM keeps it (unchanged from pre-migration behavior).
+        """
+        if self.substrate == Substrates.K8S:
+            return self.peer.peer_addresses_no_ip
+        return self.peer.peer_addresses
+
     @property
     def listen_ips(self) -> list[str]:
         """Return the IPs to listen on.

tests/unit/test_tls_client_addrs.py

@@ -0,0 +1,50 @@
+# Copyright 2026 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+"""Tests for TLS client address and common-name state accessors."""
+
+
+def _set_unit_db(harness, key, value):
+    rel_id = harness.model.get_relation("database-peers").id
+    harness.update_relation_data(rel_id, harness.charm.unit.name, {key: value})
+
+
+def test_database_address_reads_unit_databag(harness):
+    _set_unit_db(harness, "database-address", "10.1.2.3")
+    assert harness.charm.state.peer.database_address == "10.1.2.3"
+
+
+def test_database_address_none_when_unset(harness):
+    assert harness.charm.state.peer.database_address is None
+
+
+def test_client_addresses_set(harness):
+    _set_unit_db(harness, "database-address", "10.1.2.3")
+    assert harness.charm.state.client_addresses == {"10.1.2.3"}
+
+
+def test_client_addresses_empty_when_unset(harness):
+    assert harness.charm.state.client_addresses == set()
+
+
+def test_client_common_name_prefers_database_address(substrate, harness):
+    _set_unit_db(harness, "database-address", "10.1.2.3")
+    state = harness.charm.state
+    if substrate == "vm":
+        # VM: CN follows the database-address databag value.
+        assert state.client_common_name == "10.1.2.3"
+    else:
+        # K8s: CN is the endpoints FQDN, not the databag address.
+        app = state.model.app.name
+        assert state.client_common_name == f"{app}-0.{app}-endpoints"
+
+
+def test_client_common_name_falls_back_to_host(substrate, harness):
+    state = harness.charm.state
+    if substrate == "vm":
+        # VM: falls back to host when no databag address is set.
+        assert state.client_common_name == state.host
+    else:
+        # K8s: no fallback — always the endpoints FQDN.
+        app = state.model.app.name
+        assert state.client_common_name == f"{app}-0.{app}-endpoints"

tests/unit/test_tls_state.py

@@ -0,0 +1,138 @@
+# Copyright 2026 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+"""Tests for TLS state accessors on PostgreSQLPeer.
+
+Operator cert/key are fetched live from the requirer and are NOT held in peer
+state; only the peer CA rotation slots (current-ca / old-ca) and the internal
+peer material live here.
+"""
+
+import socket
+from unittest.mock import patch
+
+from single_kernel_postgresql.config.enums import Substrates  # noqa: F401  (substrate fixture)
+
+
+def _set_unit_db(harness, key, value):
+    rel_id = harness.model.get_relation("database-peers").id
+    harness.update_relation_data(rel_id, harness.charm.unit.name, {key: value})
+
+
+def test_common_hosts_k8s_includes_service_endpoints(substrate, harness):
+    """On K8s the cert-SAN source must carry the primary/replicas Service DNS."""
+    state = harness.charm.state
+    app = state.model.app.name
+    namespace = state.model.name
+    hosts = state.common_hosts
+
+    if substrate == "k8s":
+        assert f"{app}-primary.{namespace}.svc.cluster.local" in hosts
+        assert f"{app}-replicas.{namespace}.svc.cluster.local" in hosts
+        # host/fqdn still present
+        assert state.host in hosts
+        assert state.fqdn in hosts
+        # the resolved per-pod FQDN is also included (parity with the original charm)
+        assert socket.getfqdn() in hosts
+    else:
+        # VM: only host/fqdn — no k8s service DNS leaks in
+        assert not any(h.endswith(".svc.cluster.local") for h in hosts)
+        assert hosts == {state.host, state.fqdn}
+
+
+def test_ca_rotation_slots_roundtrip(harness):
+    peer = harness.charm.state.peer
+    peer.current_ca = "CA-1"
+    peer.old_ca = "CA-0"
+
+    assert peer.current_ca == "CA-1"
+    assert peer.old_ca == "CA-0"
+
+
+def test_unset_ca_slots_are_none(harness):
+    peer = harness.charm.state.peer
+    assert peer.current_ca is None
+    assert peer.old_ca is None
+
+
+# -- G1: substrate-aware cert common name (parity with the original charm) ----
+
+
+def test_cert_common_name_is_endpoints_fqdn_on_k8s(substrate, harness):
+    """K8s operator-cert CN must be `<app>-<unit_id>.<app>-endpoints` (original charm parity)."""
+    state = harness.charm.state
+    app = state.model.app.name
+    expected = f"{app}-0.{app}-endpoints"
+
+    if substrate == "k8s":
+        assert state.client_common_name == expected
+        assert state.peer_common_name == expected
+    else:
+        # VM unchanged: host-derived, not the endpoints FQDN.
+        assert state.client_common_name != expected
+        assert state.peer_common_name != expected
+
+
+def test_cert_common_name_wildcards_when_too_long_on_k8s(substrate, harness):
+    """K8s CN collapses to `*.<app>-endpoints` when the endpoints FQDN exceeds 64 chars."""
+    if substrate != "k8s":
+        return  # wildcard rule is K8s-only
+
+    state = harness.charm.state
+    app = state.model.app.name
+    # Force the endpoints FQDN past the 64-char CN limit; the suffix stays app-derived.
+    long_fqdn = "x" * 80
+    with patch.object(state, "_get_hostname_from_unit", return_value=long_fqdn):
+        assert state._k8s_cert_common_name == f"*.{app}-endpoints"
+
+
+def test_cert_common_name_vm_unchanged(substrate, harness):
+    """VM CN stays host-derived: client reads database-address, peer reads database-peers-address."""
+    if substrate != "vm":
+        return
+
+    _set_unit_db(harness, "database-address", "10.1.2.3")
+    _set_unit_db(harness, "database-peers-address", "10.4.5.6")
+    state = harness.charm.state
+    assert state.client_common_name == "10.1.2.3"
+    assert state.peer_common_name == "10.4.5.6"
+
+
+# -- G2: substrate-aware peer_addresses (no `ip` SAN on K8s) -----------------
+
+
+def test_peer_addresses_excludes_ip_on_k8s(substrate, harness):
+    """K8s peer SAN set must omit the `ip` databag key (original K8s charm never added it)."""
+    _set_unit_db(harness, "ip", "10.0.0.1")
+    _set_unit_db(harness, "private-address", "10.0.0.2")
+    _set_unit_db(harness, "database-peers-address", "10.0.0.3")
+
+    addrs = harness.charm.state.peer_addresses
+
+    if substrate == "k8s":
+        assert "10.0.0.1" not in addrs  # `ip` excluded
+        assert "10.0.0.2" in addrs  # private-address kept
+        assert "10.0.0.3" in addrs  # database-peers-address kept
+    else:
+        assert "10.0.0.1" in addrs  # VM keeps `ip`
+
+
+def test_peer_addresses_includes_ip_on_vm(substrate, harness):
+    """VM peer SAN set keeps `ip` (unchanged from pre-migration behavior)."""
+    if substrate != "vm":
+        return
+
+    _set_unit_db(harness, "ip", "10.0.0.1")
+    addrs = harness.charm.state.peer_addresses
+    assert "10.0.0.1" in addrs  # `ip` retained on VM (the G2 regression was K8s-only)
+
+
+def test_charmstate_accepts_plain_charmbase():
+    """CharmState must accept any ops.CharmBase, not only AbstractPostgreSQLCharm."""
+    import inspect
+
+    import ops
+    from single_kernel_postgresql.core.state import CharmState
+
+    ann = inspect.signature(CharmState.__init__).parameters["charm"].annotation
+    assert ann is ops.CharmBase