Fix tls.py abstraction

Author: carlcsaposs-canonical

kubernetes/metadata.yaml

@@ -47,6 +47,8 @@ requires:
     optional: true
     limit: 1
 peers:
+  tls:
+    interface: tls
   cos:
     interface: cos
   refresh-v-three:

kubernetes/src/abstract_charm.py

@@ -137,6 +137,14 @@ def is_externally_accessible(self, *, event) -> typing.Optional[bool]:
         Only defined in vm charm to return True/False. In k8s charm, returns None.
         """
 
+    @abc.abstractmethod
+    def tls_sans_ip(self, *, event) -> typing.Optional[typing.List[str]]:
+        """TLS IP subject alternative names"""
+
+    @abc.abstractmethod
+    def tls_sans_dns(self, *, event) -> typing.Optional[typing.List[str]]:
+        """TLS DNS subject alternative names"""
+
     @abc.abstractmethod
     def _status(self, *, event) -> typing.Optional[ops.StatusBase]:
         """Status of the charm."""

kubernetes/src/charm.py

@@ -411,6 +411,34 @@ def _read_write_endpoints(self, *, event) -> str:
     def _read_only_endpoints(self, *, event) -> str:
         return self._get_hosts_ports("ro")
 
+    def tls_sans_ip(self, *, event) -> typing.Optional[typing.List[str]]:
+        _, extra_ips = self.get_all_k8s_node_hostnames_and_ips()
+        return [
+            str(self.model.get_binding("juju-info").network.bind_address),
+            "127.0.0.1",
+            *extra_ips,
+        ]
+
+    def tls_sans_dns(self, *, event) -> typing.Optional[typing.List[str]]:
+        service_name = self.service_name
+        unit_name = self.unit.name.replace("/", "-")
+        extra_hosts, _ = self.get_all_k8s_node_hostnames_and_ips()
+        return [
+            socket.getfqdn(),
+            service_name,
+            f"{service_name}.{self.model_service_domain}",
+            unit_name,
+            f"{unit_name}.{self.app.name}-endpoints",
+            f"{unit_name}.{self.app.name}-endpoints.{self.model_service_domain}",
+            self.app.name,
+            f"{self.app.name}.{self.app.name}-endpoints",
+            f"{self.app.name}.{self.app.name}-endpoints.{self.model_service_domain}"
+            f"{self.app.name}-endpoints",
+            f"{self.app.name}-endpoints.{self.model_service_domain}",
+            f"{self.app.name}.{self.model_service_domain}",
+            *extra_hosts,
+        ]
+
     def get_all_k8s_node_hostnames_and_ips(
         self,
     ) -> typing.Tuple[typing.List[str], typing.List[str]]:

kubernetes/src/relations/tls.py

@@ -1,4 +1,4 @@
-# Copyright 2023 Canonical Ltd.
+# Copyright 2024 Canonical Ltd.
 # See LICENSE file for licensing details.
 
 """Relation to TLS certificate provider"""
@@ -17,11 +17,11 @@
 import relations.secrets
 
 if typing.TYPE_CHECKING:
-    import charm
+    import abstract_charm
 
 logger = logging.getLogger(__name__)
 
-_PEER_RELATION_ENDPOINT_NAME = "mysql-router-peers"
+_PEER_RELATION_ENDPOINT_NAME = "tls"
 
 _TLS_REQUESTED_CSR = "tls-requested-csr"
 _TLS_ACTIVE_CSR = "tls-active-csr"
@@ -48,7 +48,7 @@ def _generate_private_key() -> str:
 class _Relation:
     """Relation to TLS certificate provider"""
 
-    _charm: "charm.KubernetesRouterCharm"
+    _charm: "abstract_charm.MySQLRouterCharm"
     _interface: tls_certificates.TLSCertificatesRequiresV2
     _secrets: relations.secrets.RelationSecrets
 
@@ -110,56 +110,35 @@ def save_certificate(self, event: tls_certificates.CertificateAvailableEvent) ->
         logger.debug(f"Saved TLS certificate {event=}")
         self._charm.reconcile(event=None)
 
-    def _generate_csr(self, key: bytes) -> bytes:
+    def _generate_csr(self, *, event, key: bytes) -> bytes:
         """Generate certificate signing request (CSR)."""
-        service_name = self._charm.service_name
-        unit_name = self._charm.unit.name.replace("/", "-")
-        extra_hosts, extra_ips = self._charm.get_all_k8s_node_hostnames_and_ips()
         return tls_certificates.generate_csr(
             private_key=key,
             # X.509 CommonName has a limit of 64 characters
             # (https://github.com/pyca/cryptography/issues/10553)
             subject=socket.getfqdn()[:64],
             organization=self._charm.app.name,
-            sans_dns=[
-                socket.getfqdn(),
-                service_name,
-                f"{service_name}.{self._charm.model_service_domain}",
-                unit_name,
-                f"{unit_name}.{self._charm.app.name}-endpoints",
-                f"{unit_name}.{self._charm.app.name}-endpoints.{self._charm.model_service_domain}",
-                self._charm.app.name,
-                f"{self._charm.app.name}.{self._charm.app.name}-endpoints",
-                f"{self._charm.app.name}.{self._charm.app.name}-endpoints.{self._charm.model_service_domain}"
-                f"{self._charm.app.name}-endpoints",
-                f"{self._charm.app.name}-endpoints.{self._charm.model_service_domain}",
-                f"{self._charm.app.name}.{self._charm.model_service_domain}",
-                *extra_hosts,
-            ],
-            sans_ip=[
-                str(self._charm.model.get_binding("juju-info").network.bind_address),
-                "127.0.0.1",
-                *extra_ips,
-            ],
+            sans_ip=self._charm.tls_sans_ip(event=event),
+            sans_dns=self._charm.tls_sans_dns(event=event),
         )
 
-    def request_certificate_creation(self):
+    def request_certificate_creation(self, *, event):
         """Request new TLS certificate from related provider charm."""
         logger.debug("Requesting TLS certificate creation")
-        csr = self._generate_csr(self.key.encode("utf-8"))
+        csr = self._generate_csr(event=event, key=self.key.encode("utf-8"))
         self._interface.request_certificate_creation(certificate_signing_request=csr)
         self._secrets.set_value(
             relations.secrets.UNIT_SCOPE, _TLS_REQUESTED_CSR, csr.decode("utf-8")
         )
         logger.debug("Requested TLS certificate creation")
 
-    def request_certificate_renewal(self):
+    def request_certificate_renewal(self, *, event):
         """Request TLS certificate renewal from related provider charm."""
         logger.debug("Requesting TLS certificate renewal")
         old_csr = self._secrets.get_value(relations.secrets.UNIT_SCOPE, _TLS_ACTIVE_CSR).encode(
             "utf-8"
         )
-        new_csr = self._generate_csr(self.key.encode("utf-8"))
+        new_csr = self._generate_csr(event=event, key=self.key.encode("utf-8"))
         self._interface.request_certificate_renewal(
             old_certificate_signing_request=old_csr, new_certificate_signing_request=new_csr
         )
@@ -174,13 +153,15 @@ class RelationEndpoint(ops.Object):
 
     NAME = "certificates"
 
-    def __init__(self, charm_: "charm.KubernetesRouterCharm") -> None:
+    def __init__(self, charm_: "abstract_charm.MySQLRouterCharm") -> None:
         super().__init__(charm_, self.NAME)
         self._charm = charm_
         self._interface = tls_certificates.TLSCertificatesRequiresV2(self._charm, self.NAME)
 
         self._secrets = relations.secrets.RelationSecrets(
-            charm_, self._interface.relationship_name, unit_secret_fields=[_TLS_PRIVATE_KEY]
+            charm_,
+            _PEER_RELATION_ENDPOINT_NAME,
+            unit_secret_fields=[_TLS_PRIVATE_KEY],
         )
 
         self.framework.observe(
@@ -269,7 +250,7 @@ def _on_set_tls_private_key(self, event: ops.ActionEvent) -> None:
             logger.debug("No TLS certificate relation active. Skipped certificate request")
         else:
             try:
-                self._relation.request_certificate_creation()
+                self._relation.request_certificate_creation(event=event)
             except Exception as e:
                 event.fail(f"Failed to request certificate: {e}")
                 logger.exception(
@@ -278,9 +259,9 @@ def _on_set_tls_private_key(self, event: ops.ActionEvent) -> None:
                 raise
         logger.debug("Handled set TLS private key action")
 
-    def _on_tls_relation_created(self, _) -> None:
+    def _on_tls_relation_created(self, event) -> None:
         """Request certificate when TLS relation created."""
-        self._relation.request_certificate_creation()
+        self._relation.request_certificate_creation(event=event)
 
     def _on_tls_relation_broken(self, _) -> None:
         """Delete TLS certificate."""
@@ -300,4 +281,4 @@ def _on_certificate_expiring(self, event: tls_certificates.CertificateExpiringEv
             logger.warning("Unknown certificate expiring")
             return
 
-        self._relation.request_certificate_renewal()
+        self._relation.request_certificate_renewal(event=event)

machines/src/abstract_charm.py

@@ -137,6 +137,14 @@ def is_externally_accessible(self, *, event) -> typing.Optional[bool]:
         Only defined in vm charm to return True/False. In k8s charm, returns None.
         """
 
+    @abc.abstractmethod
+    def tls_sans_ip(self, *, event) -> typing.Optional[typing.List[str]]:
+        """TLS IP subject alternative names"""
+
+    @abc.abstractmethod
+    def tls_sans_dns(self, *, event) -> typing.Optional[typing.List[str]]:
+        """TLS DNS subject alternative names"""
+
     @abc.abstractmethod
     def _status(self, *, event) -> typing.Optional[ops.StatusBase]:
         """Status of the charm."""

machines/src/charm.py

@@ -127,6 +127,15 @@ def _status(self, *, event) -> typing.Optional[ops.StatusBase]:
     def _logrotate(self) -> machine_logrotate.LogRotate:
         return machine_logrotate.LogRotate(container_=self._container)
 
+    def tls_sans_ip(self, *, event) -> typing.Optional[typing.List[str]]:
+        sans_ip = ["127.0.0.1"]  # needed for the HTTP server when related with COS
+        if self.is_externally_accessible(event=event):
+            sans_ip.append(self.host_address)
+        return sans_ip
+
+    def tls_sans_dns(self, *, event) -> typing.Optional[typing.List[str]]:
+        return None
+
     @property
     def host_address(self) -> str:
         """The host address for the machine."""

machines/src/relations/tls.py

@@ -44,8 +44,7 @@ def _generate_private_key() -> str:
     return tls_certificates.generate_private_key().decode("utf-8")
 
 
-# TODO python3.10 min version: Add `(kw_only=True)`
-@dataclasses.dataclass
+@dataclasses.dataclass(kw_only=True)
 class _Relation:
     """Relation to TLS certificate provider"""
 
@@ -113,15 +112,14 @@ def save_certificate(self, event: tls_certificates.CertificateAvailableEvent) ->
 
     def _generate_csr(self, *, event, key: bytes) -> bytes:
         """Generate certificate signing request (CSR)."""
-        sans_ip = ["127.0.0.1"]  # needed for the HTTP server when related with COS
-        if self._charm.is_externally_accessible(event=event):
-            sans_ip.append(self._charm.host_address)
-
         return tls_certificates.generate_csr(
             private_key=key,
-            subject=socket.getfqdn(),
+            # X.509 CommonName has a limit of 64 characters
+            # (https://github.com/pyca/cryptography/issues/10553)
+            subject=socket.getfqdn()[:64],
             organization=self._charm.app.name,
-            sans_ip=sans_ip,
+            sans_ip=self._charm.tls_sans_ip(event=event),
+            sans_dns=self._charm.tls_sans_dns(event=event),
         )
 
     def request_certificate_creation(self, *, event):