Improve integration test TLS certificate handling

newsfragments/612.internal.md

@@ -0,0 +1 @@
+CI: improve testing of TLS certificates with intermediates.

tests/integration/artifacts/certs.py

@@ -17,7 +17,7 @@
 from cryptography.hazmat.primitives import hashes, serialization
 from cryptography.hazmat.primitives.asymmetric import rsa
 from cryptography.hazmat.primitives.asymmetric.rsa import RSAPrivateKey
-from cryptography.hazmat.primitives.serialization import load_pem_private_key, pkcs12
+from cryptography.hazmat.primitives.serialization import load_pem_private_key
 from cryptography.x509 import Certificate
 from cryptography.x509.oid import NameOID
 from platformdirs import user_cache_dir
@@ -29,25 +29,20 @@ class CertKey:
     cert: Certificate
     key: RSAPrivateKey
 
-    def cert_bundle_as_pfx(self, password: bytes | None = None) -> bytes:
-        if password is None:
-            password = b""
-
-        return pkcs12.serialize_key_and_certificates(
-            name=b"certificate",
-            key=self.key,
-            cert=self.cert,
-            cas=None,
-            encryption_algorithm=serialization.BestAvailableEncryption(password)
-            if password
-            else serialization.NoEncryption(),
-        )
+    def get_root_ca(self) -> CertKey:
+        if self.ca is None:
+            return self
+        return self.ca.get_root_ca()
 
     def cert_bundle_as_pem(self):
         bundle = []
         bundle.append(self.cert.public_bytes(encoding=serialization.Encoding.PEM).decode("utf-8"))
-        if self.ca is not None:
-            bundle.append(self.ca.cert_bundle_as_pem())
+        ca = self.ca
+        while ca is not None:
+            # We only append this CA cert if it isn't the root
+            if ca.ca is not None:
+                bundle.append(self.ca.cert_as_pem())
+            ca = ca.ca
         return "".join(bundle)
 
     def cert_as_pem(self):
@@ -76,11 +71,10 @@ def to_json_mapping(self) -> dict[str, Any]:
         }
 
 
-def get_ca(name, root_ca=None) -> CertKey:
+def get_ca(name, issuing_ca=None) -> CertKey:
     ca_filename = Path(user_cache_dir("pytest-ess", "element")) / Path(name.lower().replace(" ", "-"))
     cert_path = ca_filename.with_suffix(".crt")
     key_path = ca_filename.with_suffix(".key")
-    bundle_path = (ca_filename.parent / (ca_filename.name + "-bundle")).with_suffix(".pem")
     if not ca_filename.parent.exists():
         os.makedirs(ca_filename.parent, exist_ok=True)
     certkey = None
@@ -93,19 +87,25 @@ def get_ca(name, root_ca=None) -> CertKey:
         with open(cert_path, "rb") as pem_in:
             cert = x509.load_pem_x509_certificate(pem_in.read(), default_backend())
         if cert.not_valid_after_utc > pytz.UTC.localize(datetime.datetime.now()):
-            certkey = CertKey(ca=root_ca, cert=cert, key=private_key)
+            certkey = CertKey(ca=issuing_ca, cert=cert, key=private_key)
+
     if not certkey:
-        certkey = generate_ca(name, root_ca)
+        certkey = generate_ca(name, issuing_ca)
         with open(key_path, "wb") as pem_out:
             pem_out.write(certkey.key_as_pem().encode("utf-8"))
         with open(cert_path, "wb") as pem_out:
             pem_out.write(certkey.cert_as_pem().encode("utf-8"))
-        with open(bundle_path, "wb") as pem_out:
-            pem_out.write(certkey.cert_bundle_as_pem().encode("utf-8"))
+
+    # Remove unused bundle - given we should only need to trust the root CA, that the tests will construct the
+    # bundle appropriate for ingresses, and that a bundle of CA certs wasn't super useful this file was unneeded
+    bundle_path = (ca_filename.parent / (ca_filename.name + "-bundle")).with_suffix(".pem")
+    if bundle_path.exists():
+        bundle_path.unlink()
+
     return certkey
 
 
-def generate_ca(name, root_ca=None) -> CertKey:
+def generate_ca(name, issuing_ca=None) -> CertKey:
     two_days = datetime.timedelta(2, 0, 0)
     three_months = datetime.timedelta(90, 0, 0)
     private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048, backend=default_backend())
@@ -120,8 +120,8 @@ def generate_ca(name, root_ca=None) -> CertKey:
             ]
         )
     )
-    if root_ca:
-        builder = builder.issuer_name(root_ca.cert.subject)
+    if issuing_ca:
+        builder = builder.issuer_name(issuing_ca.cert.subject)
     else:
         builder = builder.issuer_name(
             x509.Name(
@@ -145,12 +145,12 @@ def generate_ca(name, root_ca=None) -> CertKey:
         critical=True,
     )
     builder = builder.add_extension(x509.SubjectKeyIdentifier.from_public_key(public_key), critical=False)
-    if root_ca:
+    if issuing_ca:
         builder = builder.add_extension(
-            x509.AuthorityKeyIdentifier.from_issuer_public_key(root_ca.cert.public_key()), critical=False
+            x509.AuthorityKeyIdentifier.from_issuer_public_key(issuing_ca.cert.public_key()), critical=False
         )
-        certificate = builder.sign(root_ca.key, hashes.SHA256(), default_backend())
-        ca = CertKey(ca=root_ca, cert=certificate, key=private_key)
+        certificate = builder.sign(issuing_ca.key, hashes.SHA256(), default_backend())
+        ca = CertKey(ca=issuing_ca, cert=certificate, key=private_key)
     else:
         builder = builder.add_extension(x509.AuthorityKeyIdentifier.from_issuer_public_key(public_key), critical=False)
         certificate = builder.sign(private_key, hashes.SHA256(), default_backend())

tests/integration/fixtures/__init__.py

@@ -2,7 +2,7 @@
 #
 # SPDX-License-Identifier: AGPL-3.0-only
 
-from .ca import ca, ssl_context
+from .ca import delegated_ca, root_ca, ssl_context
 from .cluster import cluster, ess_namespace, helm_client, ingress, kube_client, prometheus_operator_crds, registry
 from .data import ESSData, generated_data
 from .helm import helm_prerequisites, ingress_ready, matrix_stack, secrets_generated
@@ -11,20 +11,21 @@
 
 __all__ = [
     "build_matrix_tools",
-    "ca",
     "cluster",
+    "delegated_ca",
     "ess_namespace",
     "ESSData",
     "generated_data",
     "helm_client",
     "helm_prerequisites",
-    "ingress",
     "ingress_ready",
+    "ingress",
     "kube_client",
     "loaded_matrix_tools",
     "matrix_stack",
     "prometheus_operator_crds",
     "registry",
+    "root_ca",
     "secrets_generated",
     "ssl_context",
     "users",

tests/integration/fixtures/ca.py

@@ -1,4 +1,4 @@
-# Copyright 2024 New Vector Ltd
+# Copyright 2024-2025 New Vector Ltd
 #
 # SPDX-License-Identifier: AGPL-3.0-only
 
@@ -9,15 +9,18 @@
 from ..artifacts import get_ca
 
 
-@pytest.fixture(autouse=True, scope="session")
-async def ca():
-    root_ca = get_ca("ESS CA")
-    delegated_ca = get_ca("ESS CA Delegated", root_ca)
-    return delegated_ca
+@pytest.fixture(scope="session")
+async def root_ca():
+    return get_ca("ESS CA")
+
+
+@pytest.fixture(scope="session")
+async def delegated_ca(root_ca):
+    return get_ca("ESS CA Delegated", root_ca)
 
 
 @pytest.fixture(scope="session")
-async def ssl_context(ca):
+async def ssl_context(root_ca):
     context = ssl.create_default_context()
-    context.load_verify_locations(cadata=ca.cert_bundle_as_pem())
+    context.load_verify_locations(cadata=root_ca.cert_as_pem())
     return context

tests/integration/fixtures/data.py

@@ -24,7 +24,8 @@ def random_string(choice, size):
 @dataclass(frozen=True)
 class ESSData:
     secrets_random: str
-    ca: CertKey
+    # Just for persisting across sessions, shouldn't be directly accessed
+    _root_ca: CertKey
 
     # Only here because we need to refer to it, in the tests, after the Secret has been constructed
     mas_oidc_client_secret: str
@@ -46,26 +47,26 @@ def from_dict(cls, kv):
         return ESSData(
             secrets_random=kv["secrets_random"],
             mas_oidc_client_secret=kv["mas_oidc_client_secret"],
-            ca=CertKey.from_dict(kv["ca"]),
+            _root_ca=CertKey.from_dict(kv["ca"]),
         )
 
     def to_json_mapping(self) -> dict:
         return {
             "secrets_random": self.secrets_random,
-            "ca": self.ca.to_json_mapping(),
+            "ca": self._root_ca.to_json_mapping(),
             "mas_oidc_client_secret": self.mas_oidc_client_secret,
         }
 
 
 @pytest.fixture(scope="session")
-async def generated_data(pytestconfig, ca):
+async def generated_data(pytestconfig, root_ca):
     serialized_data = pytestconfig.cache.get("ess-helm/generated-data", None)
     if serialized_data:
         data = ESSData.from_dict(serialized_data)
     else:
         data = ESSData(
             secrets_random=random_string(string.ascii_lowercase + string.digits, 8),
-            ca=ca,
+            _root_ca=root_ca,
             mas_oidc_client_secret=secrets.token_urlsafe(36),
         )
         pytestconfig.cache.set("ess-helm/generated-data", data.to_json_mapping())

tests/integration/fixtures/helm.py

@@ -15,14 +15,19 @@
 from lightkube.resources.core_v1 import Namespace, Secret, Service
 from lightkube.resources.networking_v1 import Ingress
 
+from ..artifacts.certs import CertKey, generate_cert
 from ..lib.helpers import kubernetes_docker_secret, kubernetes_tls_secret, wait_for_endpoint_ready
 from ..lib.utils import DockerAuth, docker_config_json, value_file_has
 from .data import ESSData
 
 
 @pytest.fixture(scope="session")
 async def helm_prerequisites(
-    kube_client: AsyncClient, helm_client: pyhelm3.Client, ca, ess_namespace: Namespace, generated_data: ESSData
+    kube_client: AsyncClient,
+    helm_client: pyhelm3.Client,
+    delegated_ca: CertKey,
+    ess_namespace: Namespace,
+    generated_data: ESSData,
 ):
     resources = []
     setups: list[Awaitable] = []
@@ -50,9 +55,7 @@ async def helm_prerequisites(
             kubernetes_tls_secret(
                 f"{generated_data.release_name}-matrix-rtc-tls",
                 generated_data.ess_namespace,
-                ca,
-                [f"mrtc.{generated_data.server_name}"],
-                bundled=True,
+                generate_cert(delegated_ca, [f"mrtc.{generated_data.server_name}"]),
             )
         )
 
@@ -61,9 +64,7 @@ async def helm_prerequisites(
             kubernetes_tls_secret(
                 f"{generated_data.release_name}-element-web-tls",
                 generated_data.ess_namespace,
-                ca,
-                [f"element.{generated_data.server_name}"],
-                bundled=True,
+                generate_cert(delegated_ca, [f"element.{generated_data.server_name}"]),
             )
         )
 
@@ -76,9 +77,7 @@ async def helm_prerequisites(
             kubernetes_tls_secret(
                 f"{generated_data.release_name}-mas-web-tls",
                 generated_data.ess_namespace,
-                ca,
-                [f"mas.{generated_data.server_name}"],
-                bundled=True,
+                generate_cert(delegated_ca, [f"mas.{generated_data.server_name}"]),
             )
         )
         resources.append(
@@ -108,9 +107,7 @@ async def helm_prerequisites(
             kubernetes_tls_secret(
                 f"{generated_data.release_name}-synapse-web-tls",
                 generated_data.ess_namespace,
-                ca,
-                [f"synapse.{generated_data.server_name}"],
-                bundled=True,
+                generate_cert(delegated_ca, [f"synapse.{generated_data.server_name}"]),
             )
         )
         resources.append(
@@ -134,9 +131,7 @@ async def helm_prerequisites(
             kubernetes_tls_secret(
                 f"{generated_data.release_name}-well-known-web-tls",
                 generated_data.ess_namespace,
-                ca,
-                [generated_data.server_name],
-                bundled=True,
+                generate_cert(delegated_ca, [generated_data.server_name]),
             )
         )
 

tests/integration/lib/helpers.py

@@ -17,7 +17,7 @@
 from lightkube.models.meta_v1 import ObjectMeta
 from lightkube.resources.core_v1 import ConfigMap, Endpoints, Namespace, Pod, Secret
 
-from ..artifacts import CertKey, generate_cert
+from ..artifacts import CertKey
 from .utils import merge
 
 
@@ -34,14 +34,14 @@ def kubernetes_docker_secret(name: str, namespace: str, docker_config_json: str)
     return secret
 
 
-def kubernetes_tls_secret(name: str, namespace: str, ca: CertKey, dns_names: list[str], bundled=False) -> Secret:
-    certificate = generate_cert(ca, dns_names)
+def kubernetes_tls_secret(name: str, namespace: str, certificate: CertKey) -> Secret:
     secret = Secret(
         type="kubernetes.io/tls",
         metadata=ObjectMeta(name=name, namespace=namespace, labels={"app.kubernetes.io/managed-by": "pytest"}),
         stringData={
-            "tls.crt": certificate.cert_bundle_as_pem() if bundled else certificate.cert_as_pem(),
+            "tls.crt": certificate.cert_bundle_as_pem(),
             "tls.key": certificate.key_as_pem(),
+            "ca.crt": certificate.get_root_ca().cert_as_pem(),
         },
     )
     return secret