feat(tls): TLS state + workload file primitives (1/4)

single_kernel_postgresql/config/literals.py

@@ -52,6 +52,10 @@
 ## TLS Paths
 TLS_CA_BUNDLE_FILE = "peer_ca_bundle.pem"
 
+# TLS relation names
+TLS_CLIENT_RELATION = "client-certificates"
+TLS_PEER_RELATION = "peer-certificates"
+
 # Scopes
 SCOPES = Literal["app", "unit"]
 APP_SCOPE = "app"

single_kernel_postgresql/core/peer_relation.py

@@ -100,6 +100,24 @@ def internal_key(self, value: str) -> None:
         """Set internal private key in the peer relation."""
         self.set_secret("internal-key", value)
 
+    @property
+    def current_ca(self) -> str | None:
+        """Current peer CA (unit secret); part of the peer CA bundle."""
+        return self.get_secret("current-ca")
+
+    @current_ca.setter
+    def current_ca(self, value: str) -> None:
+        self.set_secret("current-ca", value)
+
+    @property
+    def old_ca(self) -> str | None:
+        """Previous peer CA (unit secret); retained for the rotation window."""
+        return self.get_secret("old-ca")
+
+    @old_ca.setter
+    def old_ca(self, value: str) -> None:
+        self.set_secret("old-ca", value)
+
     @property
     def ip(self) -> str | None:
         """Get the unit's IP address from the peer relation data."""
@@ -144,6 +162,13 @@ def database_peers_address(self) -> str | None:
             return None
         return self.relation.data[self.unit].get("database-peers-address", None)
 
+    @property
+    def database_address(self) -> str | None:
+        """Get the client-facing database endpoint address."""
+        if not self.relation:
+            return None
+        return self.relation.data[self.unit].get("database-address", None)
+
     @property
     def replication_address(self) -> str | None:
         """Get the address to be used for replication communication."""
@@ -207,6 +232,27 @@ def data(self) -> MutableMapping[str, str]:
             return {}
         return self.relation.data[self.unit]
 
+    @property
+    def peer_addresses_no_ip(self) -> set[str]:
+        """Peer addresses excluding the ``ip`` databag key (original K8s charm behavior).
+
+        The K8s charm never wrote ``ip`` into the operator peer-cert SANs; it relied on
+        ``database-peers-address`` + ``replication-address`` + ``replication-offer-address``
+        + ``private-address``. The VM charm additionally included ``ip``. This property
+        exposes the K8s-shaped set so :class:`CharmState` can pick the right one per
+        substrate without the peer object needing to know the substrate.
+        """
+        peer_addrs: set[str] = set()
+        if addr := self.database_peers_address:
+            peer_addrs.add(addr)
+        if addr := self.replication_address:
+            peer_addrs.add(addr)
+        if addr := self.replication_offer_address:
+            peer_addrs.add(addr)
+        if addr := self.private_address:
+            peer_addrs.add(addr)
+        return peer_addrs
+
 
 class PostgreSQLApplication(RelationState):
     """An PostgreSQL Application is the peer application state.

single_kernel_postgresql/workload/base.py

@@ -41,6 +41,27 @@ def root(self) -> PathProtocol:
         """Return the root path."""
         pass
 
+    @property
+    @abstractmethod
+    def user(self) -> str:
+        """The OS user that owns workload files (substrate-specific)."""
+        pass
+
+    @property
+    @abstractmethod
+    def group(self) -> str:
+        """The OS group that owns workload files (substrate-specific)."""
+        pass
+
+    @property
+    def tls_file_mode(self) -> int:
+        """File mode for TLS material written to disk.
+
+        Defaults to 0o600 (VM); K8s overrides to 0o400 to match the
+        pre-migration charm.
+        """
+        return 0o600
+
     @abstractmethod
     def install(self) -> None:
         """Install the workload."""
@@ -59,26 +80,35 @@ def workload_present(self) -> bool:
         pass
 
     def write_text(
-        self, content: str, path: pathops.PathProtocol, mode: int | None = None
+        self,
+        content: str,
+        path: pathops.PathProtocol,
+        mode: int | None = None,
+        user: str | None = None,
+        group: str | None = None,
     ) -> None:
         """Write content to a file on disk.
 
         Args:
             content (str): The content to be written.
             path (pathops.PathProtocol): The file path where the content should be written.
             mode (int, optional): The mode/permissions to use when writing the file.
+            user (str, optional): The user to own the file (forwarded to pathops for
+                substrate-correct chown: os.chown on VM, Pebble push on K8s).
+            group (str, optional): The group to own the file (forwarded to pathops).
 
         Raises:
             PostgreSQLFileOperationError: If there is an error during the file write operation.
         """
         try:
-            path.write_text(content, mode=mode)
+            path.write_text(content, mode=mode, user=user, group=group)
         except (
             FileNotFoundError,
             LookupError,
             NotADirectoryError,
             PermissionError,
             pathops.PebbleConnectionError,
+            PebbleError,
             ValueError,
         ) as e:
             raise PostgreSQLFileOperationError(e) from e

single_kernel_postgresql/workload/k8s.py

@@ -167,6 +167,21 @@ def temp_file(
                 except PostgreSQLFileOperationError as e:
                     logger.warning(f"Failed to delete temporary file {file_path}: {e}")
 
+    @property
+    def user(self) -> str:
+        """The OS user that owns workload files in the K8s container."""
+        return "postgres"
+
+    @property
+    def group(self) -> str:
+        """The OS group that owns workload files in the K8s container."""
+        return "postgres"
+
+    @property
+    def tls_file_mode(self) -> int:
+        """K8s pushes TLS material owner-read-only, matching the original charm."""
+        return 0o400
+
     @property
     def root(self) -> PathProtocol:
         """Return the root path for container filesystem.

single_kernel_postgresql/workload/paths/base.py

@@ -82,3 +82,9 @@ def patroni_logs(self) -> PathProtocol:
     def pgbackrest_conf(self) -> PathProtocol:
         """Path to the patroni logs."""
         pass
+
+    @property
+    @abstractmethod
+    def tls(self) -> PathProtocol:
+        """Directory where TLS files are written for this substrate."""
+        pass

single_kernel_postgresql/workload/paths/k8s.py

@@ -60,7 +60,22 @@ def postgresql_conf(self) -> PathProtocol:
 
     @property
     def patroni_conf(self) -> PathProtocol:
-        """Path to the patroni configuration file."""
+        """Path to the patroni configuration directory.
+
+        This is the data storage root (``/var/lib/pg/data``) where the Patroni
+        config subsystem renders ``patroni.yaml`` — matching #172's Patroni port.
+        (The TLS ``.pem`` files go to the ``tls`` path, which is the same dir.)
+        """
+        return self.root / K8S_DATA_PATH
+
+    @property
+    def tls(self) -> PathProtocol:
+        """Directory where TLS files are written on K8s (the data dir Patroni reads from).
+
+        This is the *unversioned* data storage root (``/var/lib/pg/data``), which is
+        where the charm-rendered patroni.yml references the ``.pem`` files
+        (``{storage_path}/*.pem``) — NOT the versioned ``data`` subdir (``.../16/main``).
+        """
         return self.root / K8S_DATA_PATH
 
     @property

single_kernel_postgresql/workload/paths/vm.py

@@ -74,6 +74,11 @@ def patroni_conf(self) -> PathProtocol:
         """Path to the patroni.yaml file."""
         return self.snap_current / PATRONI_CONF_PATH
 
+    @property
+    def tls(self) -> PathProtocol:
+        """Directory where TLS files are written on VM (same as patroni_conf)."""
+        return self.patroni_conf
+
     @property
     def patroni_logs(self) -> PathProtocol:
         """Path to the patroni logs."""

single_kernel_postgresql/workload/vm.py

@@ -195,6 +195,16 @@ def temp_file(
                 except OSError as e:
                     raise e
 
+    @property
+    def user(self) -> str:
+        """The OS user that owns workload files on VM."""
+        return "_daemon_"
+
+    @property
+    def group(self) -> str:
+        """The OS group that owns workload files on VM."""
+        return "_daemon_"
+
     @property
     def paths(self) -> BasePaths:
         """Return Workload's paths."""

tests/unit/conftest.py

@@ -0,0 +1,37 @@
+# Copyright 2026 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+from unittest.mock import patch
+
+import pytest
+from ops.testing import Harness
+from single_kernel_postgresql.charms import k8s_charm, vm_charm
+from single_kernel_postgresql.config.literals import PEER_RELATION
+
+
+@pytest.fixture
+def harness(substrate, test_charm_path):
+    """A begun Harness for the substrate's test charm, with the peer relation added."""
+    with open(test_charm_path + "/metadata.yaml") as meta_file:
+        meta = meta_file.read()
+    with open(test_charm_path + "/actions.yaml") as actions_file:
+        actions = actions_file.read()
+    if substrate == "vm":
+        harness = Harness(vm_charm.PostgreSQLVMCharm, meta=meta, actions=actions)
+    else:
+        harness = Harness(k8s_charm.PostgreSQLK8sCharm, meta=meta, actions=actions)
+    peer_rel_id = harness.add_relation(PEER_RELATION, "postgresql-single-kernel")
+    harness.add_relation_unit(peer_rel_id, "postgresql-single-kernel/0")
+    # Set before begin(): Model.name (K8s namespace) is read by substrate-aware
+    # state accessors (e.g. common_hosts Service FQDNs).
+    harness.set_model_name("test-model")
+    # The workload's versioned paths (K8sPaths) read the major version via
+    # get_postgresql_version(), which reads refresh_versions.toml from cwd — absent
+    # in the unit env. Patch it, mirroring tests/unit/test_postgresql.py.
+    with patch(
+        "single_kernel_postgresql.workload.base.BaseWorkload.get_postgresql_version",
+        return_value="16.0",
+    ):
+        harness.begin()
+        yield harness
+    harness.cleanup()