[DPE-9315] add user-provided private key

config.yaml

@@ -16,3 +16,8 @@ options:
       - charmed-sentinel-operator: administrative user for operating valkey sentinel
       - charmed-sentinel-valkey: used by sentinel to authenticate against valkey
       - charmed-sentinel-peers: used by sentinel for inter-sentinel authentication
+
+  tls-client-private-key:
+    type: secret
+    description: |
+      A Juju secret URI of a secret containing the private key for client TLS certificates.

pyproject.toml

@@ -48,6 +48,7 @@ pytest = "*"
 pytest-asyncio = "*"
 allure-pytest = "*"
 allure-pytest-default-results = "^0.1.2"
+charmlibs-interfaces-tls-certificates = ">=1.6.0"
 data-platform-helpers = ">=0.1.7"
 jubilant = "^1.6.0"
 python-dateutil = "*"

src/core/models.py

@@ -13,6 +13,7 @@
     PrivateKey,
 )
 from charms.data_platform_libs.v1.data_interfaces import (
+    ExtraSecretStr,
     OpsOtherPeerUnitRepositoryInterface,
     OpsPeerRepositoryInterface,
     OpsPeerUnitRepositoryInterface,
@@ -55,6 +56,7 @@ class PeerAppModel(PeerModel):
     start_member: str = Field(default="")
     internal_ca_certificate: InternalCertificatesSecret = Field(default="")
     internal_ca_private_key: InternalCertificatesSecret = Field(default="")
+    tls_client_private_key: ExtraSecretStr = Field(default=None)
 
 
 class PeerUnitModel(PeerModel):
@@ -235,3 +237,11 @@ def internal_ca_private_key(self) -> PrivateKey | None:
             return None
 
         return PrivateKey.from_string(self.model.internal_ca_private_key)
+
+    @property
+    def tls_client_private_key(self) -> PrivateKey | None:
+        """Retrieve the private key for client TLS."""
+        if self.model and (private_key := self.model.tls_client_private_key):
+            return PrivateKey(raw=private_key)
+
+        return None

src/events/tls.py

@@ -24,6 +24,7 @@
     CLIENT_PORT,
     CLIENT_TLS_RELATION_NAME,
     PEER_RELATION,
+    TLS_CLIENT_PRIVATE_KEY_CONFIG,
     TLSCARotationState,
     TLSState,
 )
@@ -57,7 +58,7 @@ def __init__(self, charm: "ValkeyCharm"):
                     sans_dns=self.charm.tls_manager.build_sans_dns(),
                 ),
             ],
-            private_key=None,
+            private_key=self.charm.tls_manager.get_client_tls_private_key(),
             refresh_events=[self.refresh_tls_certificates_event],
         )
 
@@ -78,6 +79,8 @@ def __init__(self, charm: "ValkeyCharm"):
             self.charm.on[PEER_RELATION].relation_changed, self._on_peer_relation_changed
         )
         self.framework.observe(self.charm.on.update_status, self._on_update_status)
+        self.framework.observe(self.charm.on.secret_changed, self._on_secret_changed)
+        self.framework.observe(self.charm.on.config_changed, self._on_config_changed)
 
     def _on_peer_relation_created(self, event: ops.RelationCreatedEvent) -> None:
         """Set up self-signed certificates for peer TLS by default."""
@@ -282,6 +285,39 @@ def _on_update_status(self, event: ops.UpdateStatusEvent) -> None:
             logger.debug("Trigger peer relation change to orchestrate certificate/CA rotation")
             self.charm.on[PEER_RELATION].relation_changed.emit(self.charm.state.peer_relation)
 
+    def _on_secret_changed(self, event: ops.SecretChangedEvent) -> None:
+        """Handle TLS related secret changes."""
+        if not (secret_id := self.charm.config.get(TLS_CLIENT_PRIVATE_KEY_CONFIG)):
+            return
+
+        if secret_id != event.secret.id:
+            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.")
+            return
+
+        if self.charm.unit.is_leader():
+            self.charm.state.cluster.update({"tls_client_private_key": private_key.raw})
+
+        if self.charm.state.client_tls_relation:
+            self.refresh_tls_certificates_event.emit()
+
+    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.")
+            return
+
+        if self.charm.unit.is_leader():
+            self.charm.state.cluster.update({"tls_client_private_key": private_key.raw})
+
+        if self.charm.state.client_tls_relation:
+            self.refresh_tls_certificates_event.emit()
+
     def _enable_client_tls(self) -> None:
         """Check preconditions and enable TLS if possible."""
         if not all(server.model.client_cert_ready for server in self.charm.state.servers):

src/literals.py

@@ -41,6 +41,7 @@
 INTERNAL_USERS_PASSWORD_CONFIG = "system-users"
 INTERNAL_USERS_SECRET_LABEL_SUFFIX = "internal_users_secret"
 INTERNET_CERTS_SECRET_LABEL_SUFFIX = "internal_certificates_secret"
+TLS_CLIENT_PRIVATE_KEY_CONFIG = "tls-client-private-key"
 
 DATA_STORAGE = "data"
 

src/managers/tls.py

@@ -4,7 +4,10 @@
 
 """Manager for handling TLS-related operations."""
 
+import base64
+import binascii
 import logging
+import re
 from datetime import timedelta
 
 from charmlibs.interfaces.tls_certificates import (
@@ -17,11 +20,12 @@
 from data_platform_helpers.advanced_statuses.models import StatusObject
 from data_platform_helpers.advanced_statuses.protocol import ManagerStatusProtocol
 from data_platform_helpers.advanced_statuses.types import Scope
+from ops import ModelError, SecretNotFoundError
 
 from common.exceptions import ValkeyWorkloadCommandError
 from core.base_workload import WorkloadBase
 from core.cluster_state import ClusterState
-from literals import TLSCARotationState, TLSState
+from literals import TLS_CLIENT_PRIVATE_KEY_CONFIG, TLSCARotationState, TLSState
 from statuses import CharmStatuses, TLSStatuses
 
 logger = logging.getLogger(__name__)
@@ -117,6 +121,60 @@ def build_sans_dns(self) -> frozenset[str]:
 
         return frozenset(sans_dns)
 
+    def read_and_validate_private_key(self, private_key_secret_id: str) -> PrivateKey | None:
+        """Read and validate a private key provided via Juju secret.
+
+        Args:
+            private_key_secret_id (str): The Juju secret ID for the secret
+                that stores the private key.
+
+        Returns:
+            PrivateKey: The private key.
+        """
+        try:
+            secret_content = self.state.get_secret_from_id(private_key_secret_id).get(
+                "private-key"
+            )
+        except (ModelError, SecretNotFoundError) as e:
+            logger.error(e)
+            return None
+
+        if secret_content is None:
+            logger.error(f"Secret {private_key_secret_id} does not contain a private key.")
+            return None
+
+        try:
+            private_key = (
+                secret_content
+                if re.match(r"(-+(BEGIN|END) [A-Z ]+-+)", secret_content)
+                else base64.b64decode(secret_content).decode("utf-8").strip()
+            )
+        except (UnicodeDecodeError, binascii.Error) as e:
+            logger.error(e)
+            return None
+        try:
+            private_key = PrivateKey(raw=private_key)
+        except ValueError as e:
+            logger.error(e)
+            return None
+        if not private_key.is_valid():
+            logger.error("Invalid private key format.")
+            return None
+
+        return private_key
+
+    def get_client_tls_private_key(self) -> PrivateKey | None:
+        """Get the private key provided by users, if available."""
+        if secret_id := self.state.config.get(TLS_CLIENT_PRIVATE_KEY_CONFIG):
+            if private_key := self.read_and_validate_private_key(secret_id):
+                return private_key
+
+            # in case the configured secret is invalid
+            return self.state.cluster.tls_client_private_key
+
+        # in case no user supplied private key configured
+        return None
+
     def _generate_private_key(self) -> PrivateKey:
         """Generate a private key for use in peer TLS."""
         return PrivateKey.generate()
@@ -246,7 +304,7 @@ def start_ca_rotation_if_required(
         )
         return True
 
-    def get_statuses(self, scope: Scope, recompute: bool = False) -> list[StatusObject]:
+    def get_statuses(self, scope: Scope, recompute: bool = False) -> list[StatusObject]:  # noqa: C901
         """Compute the TLS statuses."""
         status_list: list[StatusObject] = []
 
@@ -264,6 +322,14 @@ def get_statuses(self, scope: Scope, recompute: bool = False) -> list[StatusObje
         ):
             status_list.append(TLSStatuses.DISABLING_CLIENT_TLS_FAILED.value)
 
+        if self.state.cluster.tls_client_private_key and not self.state.client_tls_relation:
+            status_list.append(TLSStatuses.PRIVATE_KEY_BUT_NO_TLS.value)
+
+        if (
+            private_key_id := self.state.config.get(TLS_CLIENT_PRIVATE_KEY_CONFIG)
+        ) and self.read_and_validate_private_key(str(private_key_id)) is None:
+            status_list.append(TLSStatuses.PRIVATE_KEY_INVALID.value)
+
         if self.state.unit_server.tls_client_state == TLSState.TO_NO_TLS:
             status_list.append(TLSStatuses.DISABLING_CLIENT_TLS.value)
 

src/statuses.py

@@ -116,3 +116,10 @@ class TLSStatuses(Enum):
     CA_ROTATION_UPDATED = StatusObject(
         status="maintenance", message="TLS CA rotation: certificates updated"
     )
+    PRIVATE_KEY_BUT_NO_TLS = StatusObject(
+        status="blocked", message="Private Key provided, but client TLS not enabled"
+    )
+    PRIVATE_KEY_INVALID = StatusObject(
+        status="blocked",
+        message="The private key provided is not valid. Please provide a valid private key",
+    )

tests/integration/tls/test_certificate_rotation.py

@@ -226,7 +226,7 @@ async def test_ca_rotation_by_expiration(juju: jubilant.Juju) -> None:
     juju.config(app=TLS_NAME, values=tls_config)
     juju.wait(
         lambda status: are_agents_idle(status, TLS_NAME, idle_period=30),
-        timeout=100,
+        timeout=600,
     )
 
     logger.info("Enabling client TLS again")