feat(tls): substrate-aware CharmState SAN/CN accessors (2/4)

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