Merge ~r00ta/maas:MAASENG-2908-2-3.5 into maas:3.5

[Jacopo Rota]

fix: generate a root maas CA certificate to sign the cluster certificate to be used by temporal mTLS.

(cherry-pick from 34d4731ac18cd72fe1f56300f5c7231fdc7773b9)

[Jacopo Rota (r00ta)]

self approving backport

review: Approve

[MAAS Lander (maas-lander)]

UNIT TESTS
-b MAASENG-2908-2-3.5 lp:~r00ta/maas/+git/maas into -b 3.5 lp:~maas-committers/maas

STATUS: SUCCESS
COMMIT: a8782120a90542c5c75090779f1c24b7df85688e

review: Approve

diff --git a/src/maasserver/rpc/tests/test_regionservice.py b/src/maasserver/rpc/tests/test_regionservice.py
index 926940a..901f78a 100644
--- a/src/maasserver/rpc/tests/test_regionservice.py
+++ b/src/maasserver/rpc/tests/test_regionservice.py
@@ -281,7 +281,8 @@ class TestRegionServer(MAASTransactionServerTestCase):
         def setup_cluster_certificates():
             secret_manger = SecretManager()
             secret_manger.set_composite_secret(
-                "cluster-certificate", {"key": "key", "cert": "cert"}
+                "cluster-certificate",
+                {"key": "key", "cert": "cert", "cacerts": "cacerts"},
             )
 
         yield deferToDatabase(transactional(setup_cluster_certificates))
@@ -367,7 +368,7 @@ class TestRegionServer(MAASTransactionServerTestCase):
             json.loads(
                 fernet_decrypt_psk(response["encrypted_cluster_certificate"])
             ),
-            {"key": "key", "cert": "cert"},
+            {"key": "key", "cert": "cert", "cacerts": "cacerts"},
         )
 
     @wait_for_reactor
diff --git a/src/maasserver/secrets.py b/src/maasserver/secrets.py
index 10334ec..1c7edd0 100644
--- a/src/maasserver/secrets.py
+++ b/src/maasserver/secrets.py
@@ -59,6 +59,7 @@ GLOBAL_SECRETS = frozenset(
         "macaroon-key",
         "omapi-key",
         "rpc-shared",
+        "maas-ca-certificate",
         "cluster-certificate",
         "tls",
         "vcenter-password",
diff --git a/src/maasserver/start_up.py b/src/maasserver/start_up.py
index 41f3a0b..f2a1808 100644
--- a/src/maasserver/start_up.py
+++ b/src/maasserver/start_up.py
@@ -27,7 +27,10 @@ from maasserver.models.config import ensure_uuid_in_config
 from maasserver.models.domain import dns_kms_setting_changed
 from maasserver.secrets import SecretManager, SecretNotFound
 from maasserver.utils import synchronised
-from maasserver.utils.certificates import generate_self_signed_v3_certificate
+from maasserver.utils.certificates import (
+    generate_ca_certificate,
+    generate_signed_certificate,
+)
 from maasserver.utils.orm import (
     get_psycopg2_exception,
     transactional,
@@ -128,34 +131,59 @@ def _cleanup_expired_discovered_ip_addresses() -> None:
     ).delete()
 
 
+def _get_certificate_from_database(
+    secret_manager: SecretManager, secret_name: str
+) -> Certificate | None:
+    try:
+        raw_certificate = secret_manager.get_composite_secret(secret_name)
+        return Certificate.from_pem(
+            raw_certificate["key"],
+            raw_certificate["cert"],
+            ca_certs_material=raw_certificate.get("cacerts", ""),
+        )
+    except SecretNotFound:
+        return None
+
+
 def _create_cluster_certificate_if_necessary(
     client: VaultClient | None = None,
 ) -> None:
     # Use the vault if configured.
     secret_manager = SecretManager(client)
-    # The PK/certificate are not on the disk yet
+
+    # The PK/certificate for the maas CA are not in the db yet
+    maas_ca = _get_certificate_from_database(
+        secret_manager, "maas-ca-certificate"
+    )
+    if not maas_ca:
+        maas_ca = generate_ca_certificate("maas-ca")
+        secrets = {
+            "key": maas_ca.private_key_pem(),
+            "cert": maas_ca.certificate_pem(),
+        }
+        secret_manager.set_composite_secret("maas-ca-certificate", secrets)
+
+    # The PK/certificate for the cluster are not in the db yet
+    cluster_certificate = _get_certificate_from_database(
+        secret_manager, "cluster-certificate"
+    )
+    if not cluster_certificate:
+        cluster_certificate = generate_signed_certificate(
+            maas_ca, "maas-cluster", b"DNS:maas"
+        )
+        secrets = {
+            "key": cluster_certificate.private_key_pem(),
+            "cert": cluster_certificate.certificate_pem(),
+            "cacerts": cluster_certificate.ca_certificates_pem(),
+        }
+        secret_manager.set_composite_secret("cluster-certificate", secrets)
+
+    # If the certificates are not on the disk yet store them.
     if not get_maas_cluster_cert_paths():
-        # Create the PK/certificate if they don't exist yet. Store the files on the disk as well.
-        try:
-            raw_certificate = secret_manager.get_composite_secret(
-                "cluster-certificate"
-            )
-            certificate = Certificate.from_pem(
-                raw_certificate["key"],
-                raw_certificate["cert"],
-            )
-        except SecretNotFound:
-            # If there are no PK/certificate at all, generate them.
-            certificate = generate_self_signed_v3_certificate("maas", b"DNS:*")
-            secrets = {
-                "key": certificate.private_key_pem(),
-                "cert": certificate.certificate_pem(),
-            }
-            secret_manager.set_composite_secret("cluster-certificate", secrets)
-        # Store the PK and the certificate to the disk
         store_maas_cluster_cert_tuple(
-            private_key=certificate.private_key_pem().encode(),
-            certificate=certificate.certificate_pem().encode(),
+            private_key=cluster_certificate.private_key_pem().encode(),
+            certificate=cluster_certificate.certificate_pem().encode(),
+            cacerts=cluster_certificate.ca_certificates_pem().encode(),
         )
 
 
diff --git a/src/maasserver/tests/test_start_up.py b/src/maasserver/tests/test_start_up.py
index eb33ca2..125b848 100644
--- a/src/maasserver/tests/test_start_up.py
+++ b/src/maasserver/tests/test_start_up.py
@@ -28,7 +28,10 @@ from maasserver.testing.testcase import (
     MAASTransactionServerTestCase,
 )
 from maasserver.testing.vault import FakeVaultClient
-from maasserver.utils.certificates import generate_self_signed_v3_certificate
+from maasserver.utils.certificates import (
+    generate_ca_certificate,
+    generate_signed_certificate,
+)
 from maasserver.utils.orm import post_commit_hooks
 from maasserver.vault import UnknownSecretPath, VaultError
 from maastesting import get_testing_timeout
@@ -524,11 +527,28 @@ class TestCreateClusterCertificate(MAASServerTestCase):
         self.assertRaises(
             SecretNotFound,
             secret_manager.get_composite_secret,
+            "maas-ca-certificate",
+        )
+        self.assertRaises(
+            SecretNotFound,
+            secret_manager.get_composite_secret,
             "cluster-certificate",
         )
 
         _create_cluster_certificate_if_necessary()
         self.assertIsNotNone(get_maas_cluster_cert_paths())
+
+        maasca_secret = secret_manager.get_composite_secret(
+            "maas-ca-certificate"
+        )
+        maasca = Certificate.from_pem(
+            maasca_secret["key"],
+            maasca_secret["cert"],
+        )
+
+        self.assertIsNotNone(maasca.private_key_pem())
+        self.assertIsNotNone(maasca.certificate_pem())
+
         certificate_secret = secret_manager.get_composite_secret(
             "cluster-certificate"
         )
@@ -539,6 +559,7 @@ class TestCreateClusterCertificate(MAASServerTestCase):
 
         self.assertIsNotNone(certificate.private_key_pem())
         self.assertIsNotNone(certificate.certificate_pem())
+        self.assertIsNotNone(certificate.ca_certificates_pem())
 
     def test_fetch_certificates_from_db(self):
         # No certificates on the disk
@@ -546,10 +567,18 @@ class TestCreateClusterCertificate(MAASServerTestCase):
 
         # store the certificates on the database
         secret_manager = SecretManager()
-        certificate = generate_self_signed_v3_certificate("maas")
+        maasca = generate_ca_certificate("maasca")
+        secrets = {
+            "key": maasca.private_key_pem(),
+            "cert": maasca.certificate_pem(),
+        }
+        secret_manager.set_composite_secret("maas-ca-certificate", secrets)
+
+        cluster_certificate = generate_signed_certificate(maasca, "cluster")
         secrets = {
-            "key": certificate.private_key_pem(),
-            "cert": certificate.certificate_pem(),
+            "key": cluster_certificate.private_key_pem(),
+            "cert": cluster_certificate.certificate_pem(),
+            "cacerts": cluster_certificate.ca_certificates_pem(),
         }
         secret_manager.set_composite_secret("cluster-certificate", secrets)
 
diff --git a/src/maasserver/utils/certificates.py b/src/maasserver/utils/certificates.py
index 9eea26b..f63657f 100644
--- a/src/maasserver/utils/certificates.py
+++ b/src/maasserver/utils/certificates.py
@@ -1,5 +1,5 @@
 from maasserver.models import Config
-from provisioningserver.certificates import Certificate
+from provisioningserver.certificates import Certificate, CertificateRequest
 from provisioningserver.utils.env import MAAS_UUID
 
 
@@ -38,24 +38,58 @@ def generate_certificate(cn) -> Certificate:
     )
 
 
-def generate_self_signed_v3_certificate(
-    cn: str, subject_alternative_name: bytes | None = None
+def generate_ca_certificate(cn: str) -> Certificate:
+    """
+    Generate an X509 MAAS CA certificate with an RSA private key.
+
+    Set Organization (O) and Organizational Unit (OU) fields to identify
+    that a certificate was created from this MAAS deployment.
+
+    Parameters:
+        cn (str): Common Name (CN) for the subject of the certificate.
+
+    Returns:
+        Certificate: The generated X509 MAAS CA certificate with an RSA private key.
+
+    Raises:
+        AssertionError: If the MAAS_UUID is not configured.
+    """
+    maas_uuid = MAAS_UUID.get()
+    assert maas_uuid is not None, "MAAS_UUID not configured, ensure it is set."
+    return Certificate.generate_ca_certificate(
+        cn,
+        organization_name="MAAS",
+        organizational_unit_name=maas_uuid,
+    )
+
+
+def generate_signed_certificate(
+    ca: Certificate, cn: str, subject_alternative_name: bytes | None = None
 ) -> Certificate:
-    """Generate an X509 V3 self signed certificate with an RSA private key.
+    """
+    Generate an X509 V3 certificate with an RSA private key signed with the root CA certificate provided.
 
-    Set O and OU so that we can identify that a certificate was
-    created from this MAAS deployment.
+    Parameters:
+        ca (Certificate): The Certificate Authority (CA) certificate used to sign the generated certificate.
+        cn (str): Common Name (CN) for the subject of the certificate.
+        subject_alternative_name (bytes | None): Subject Alternative Name (SAN) for the certificate,
+                                                  can be None if not needed.
+
+    Returns:
+        Certificate: The generated X509 V3 certificate signed with the root CA's private key.
+
+    Raises:
+        AssertionError: If the MAAS_UUID is not configured.
     """
-    # Set O and OU so that we can identify that a certificate was
-    # created from this MAAS deployment.
     maas_uuid = MAAS_UUID.get()
     assert maas_uuid is not None, "MAAS_UUID not configured, ensure it is set."
-    return Certificate.generate_self_signed_v3(
+    certificate_request = CertificateRequest.generate(
         cn,
         organization_name="MAAS",
         organizational_unit_name=maas_uuid,
         subject_alternative_name=subject_alternative_name,
     )
+    return ca.sign_certificate_request(certificate_request)
 
 
 def certificate_generated_by_this_maas(certificate: Certificate):
diff --git a/src/maasserver/utils/tests/test_certificates.py b/src/maasserver/utils/tests/test_certificates.py
index 1514bde..d84113d 100644
--- a/src/maasserver/utils/tests/test_certificates.py
+++ b/src/maasserver/utils/tests/test_certificates.py
@@ -1,3 +1,4 @@
+from unittest.mock import Mock
 from uuid import uuid4
 
 from OpenSSL import crypto
@@ -8,8 +9,8 @@ from maasserver.testing.testcase import MAASServerTestCase
 import maasserver.utils.certificates as certificates
 from maasserver.utils.certificates import (
     certificate_generated_by_this_maas,
+    generate_ca_certificate,
     generate_certificate,
-    generate_self_signed_v3_certificate,
     get_maas_client_cn,
 )
 from provisioningserver.certificates import Certificate
@@ -48,20 +49,32 @@ class TestGenerateCertificate(MAASServerTestCase):
         )
 
 
-class TestGenerateSelfSignedCertificate(MAASServerTestCase):
-    def test_generate_self_signed_certificate(self):
+class TestGenerateCACertificate(MAASServerTestCase):
+    def test_generate_ca_certificate(self):
         mock_cert = self.patch_autospec(certificates, "Certificate")
         maas_uuid = str(uuid4())
         self.useFixture(MAASUUIDFixture(maas_uuid))
-        generate_self_signed_v3_certificate("maas", b"DNS:*")
-        mock_cert.generate_self_signed_v3.assert_called_once_with(
+        generate_ca_certificate("maas")
+        mock_cert.generate_ca_certificate.assert_called_once_with(
             "maas",
             organization_name="MAAS",
             organizational_unit_name=maas_uuid,
-            subject_alternative_name=b"DNS:*",
         )
 
 
+class TestGenerateSignedCertificate(MAASServerTestCase):
+    def test_generate_ca_certificate(self):
+        mock_cert_request = self.patch_autospec(
+            certificates, "CertificateRequest"
+        )
+        maas_uuid = str(uuid4())
+        self.useFixture(MAASUUIDFixture(maas_uuid))
+        ca = Mock()
+        certificates.generate_signed_certificate(ca, "maas")
+        mock_cert_request.generate.assert_called_once()
+        ca.sign_certificate_request.assert_called_once()
+
+
 class TestCertificateGeneratedByThisMAAS(MAASServerTestCase):
     def setUp(self):
         super().setUp()
diff --git a/src/provisioningserver/certificates.py b/src/provisioningserver/certificates.py
index f7bdf7c..ddd6559 100644
--- a/src/provisioningserver/certificates.py
+++ b/src/provisioningserver/certificates.py
@@ -31,6 +31,44 @@ class CertificateError(Exception):
     """Error handling certificates and keys."""
 
 
+class CertificateRequest(NamedTuple):
+    key: crypto.PKey
+    csr: crypto.X509Req
+
+    @classmethod
+    def generate(
+        cls,
+        cn: str,
+        organization_name: Optional[str] = None,
+        organizational_unit_name: Optional[str] = None,
+        key_bits: int = 4096,
+        subject_alternative_name: bytes | None = None,
+    ):
+        key = crypto.PKey()
+        key.generate_key(crypto.TYPE_RSA, key_bits)
+
+        csr = crypto.X509Req()
+        csr.get_subject().CN = cn[:64]
+        if organization_name:
+            csr.get_subject().organizationName = organization_name[:64]
+        if organizational_unit_name:
+            csr.get_subject().organizationalUnitName = (
+                organizational_unit_name[:64]
+            )
+        csr.set_pubkey(key)
+
+        if subject_alternative_name:
+            csr.add_extensions(
+                [
+                    crypto.X509Extension(
+                        b"subjectAltName", False, subject_alternative_name
+                    )
+                ]
+            )
+        csr.sign(key, "sha512")
+        return cls(key, csr)
+
+
 class Certificate(NamedTuple):
     """A self-signed X509 certificate with an associated key."""
 
@@ -72,12 +110,10 @@ class Certificate(NamedTuple):
         cls,
         key: crypto.PKey,
         version: crypto.x509.Version,
-        cn: str,
         validity: timedelta,
     ) -> crypto.X509:
         cert = crypto.X509()
         cert.set_version(version.value)
-        cert.get_subject().CN = cn[:64]
         cert.set_serial_number(random.randint(0, (1 << 128) - 1))
         cert.gmtime_adj_notBefore(0)
         cert.gmtime_adj_notAfter(int(validity.total_seconds()))
@@ -106,8 +142,9 @@ class Certificate(NamedTuple):
         key.generate_key(crypto.TYPE_RSA, key_bits)
 
         cert = cls._build_base_certificate(
-            key, crypto.x509.Version.v1, cn, validity
+            key, crypto.x509.Version.v1, validity
         )
+        cert.get_subject().CN = cn[:64]
         if organization_name:
             cert.get_issuer().organizationName = organization_name[:64]
         if organizational_unit_name:
@@ -119,26 +156,26 @@ class Certificate(NamedTuple):
         return cls(key, cert, ())
 
     @classmethod
-    def generate_self_signed_v3(
+    def generate_ca_certificate(
         cls,
         cn: str,
         organization_name: Optional[str] = None,
         organizational_unit_name: Optional[str] = None,
         key_bits: int = 4096,
         validity: timedelta = timedelta(days=3650),
-        subject_alternative_name: bytes | None = None,
     ) -> "Certificate":
-        """Low-level method for generating a self-signed certificate X509 v3 certificate.
+        """Low-level method for generating a root X509 certificate.
 
         This should only be used in test and in cases where you don't have
-        access to the database. Use maasserver.utils.certificate.generate_self_signed_v3_certificate() instead.
+        access to the database. Use maasserver.utils.certificate.generate_ca_certificate() instead.
         """
         key = crypto.PKey()
         key.generate_key(crypto.TYPE_RSA, key_bits)
 
         cert = cls._build_base_certificate(
-            key, crypto.x509.Version.v3, cn, validity
+            key, crypto.x509.Version.v3, validity
         )
+        cert.get_subject().CN = cn[:64]
         if organization_name:
             cert.get_issuer().organizationName = organization_name[:64]
         if organizational_unit_name:
@@ -151,17 +188,38 @@ class Certificate(NamedTuple):
         cert.add_extensions(
             [crypto.X509Extension(b"basicConstraints", True, b"CA:TRUE")]
         )
-        if subject_alternative_name:
-            cert.add_extensions(
-                [
-                    crypto.X509Extension(
-                        b"subjectAltName", False, subject_alternative_name
-                    )
-                ]
-            )
         cert.sign(key, "sha512")
         return cls(key, cert, ())
 
+    def sign_certificate_request(
+        self,
+        certificate_request: CertificateRequest,
+        validity: timedelta = timedelta(days=3650),
+    ) -> "Certificate":
+        """
+        Sign a certificate request with the CA's private key.
+
+        This method signs the provided certificate request with the Certificate Authority (CA)'s
+        private key and returns the signed certificate.
+
+        Parameters:
+            certificate_request (CertificateRequest): The certificate request to sign.
+            validity (timedelta): The validity period for the signed certificate. Default is 10 years.
+
+        Returns:
+            Certificate: The signed certificate.
+        """
+        cert = Certificate._build_base_certificate(
+            certificate_request.key, crypto.x509.Version.v3, validity
+        )
+        cert.set_issuer(self.cert.get_subject())
+        cert.set_subject(certificate_request.csr.get_subject())
+        cert.add_extensions(certificate_request.csr.get_extensions())
+        cert.sign(self.key, "sha512")
+        return Certificate(
+            key=certificate_request.key, cert=cert, ca_certs=(self.cert,)
+        )
+
     def cn(self) -> str:
         """Return the certificate CN."""
         return self.cert.get_subject().CN
@@ -256,19 +314,24 @@ def _get_cluster_certificates_path() -> Path:
     return Path(maas_root) / "certificates"
 
 
-def get_maas_cluster_cert_paths() -> tuple[str, str] | None:
+def get_maas_cluster_cert_paths() -> tuple[str, str, str] | None:
     """Return a 2-tuple with certificate and private key paths for the cluster certificates."""
 
     cert_dir = _get_cluster_certificates_path()
     private_key = cert_dir / "cluster.key"
     certificate = cert_dir / "cluster.pem"
-    if not private_key.exists() or not certificate.exists():
+    cacerts = cert_dir / "cacerts.pem"
+    if (
+        not private_key.exists()
+        or not certificate.exists()
+        or not cacerts.exists()
+    ):
         return None
-    return str(certificate), str(private_key)
+    return str(certificate), str(private_key), str(cacerts)
 
 
 def store_maas_cluster_cert_tuple(
-    private_key: bytes, certificate: bytes
+    private_key: bytes, certificate: bytes, cacerts: bytes
 ) -> None:
     """
     Stores the private key and the certificate on the disk.
@@ -289,6 +352,13 @@ def store_maas_cluster_cert_tuple(
         mode=0o644,
     )
 
+    atomic_write(
+        cacerts,
+        cert_dir / "cacerts.pem",
+        overwrite=True,
+        mode=0o644,
+    )
+
 
 def get_maas_cert_tuple():
     """Return a 2-tuple with certificate and private key paths.
diff --git a/src/provisioningserver/rpc/clusterservice.py b/src/provisioningserver/rpc/clusterservice.py
index ba8970b..35f8d9a 100644
--- a/src/provisioningserver/rpc/clusterservice.py
+++ b/src/provisioningserver/rpc/clusterservice.py
@@ -976,10 +976,12 @@ class ClusterClient(Cluster):
                 certificate = Certificate.from_pem(
                     decoded_secret["key"],
                     decoded_secret["cert"],
+                    ca_certs_material=decoded_secret["cacerts"],
                 )
                 store_maas_cluster_cert_tuple(
                     private_key=certificate.private_key_pem().encode(),
                     certificate=certificate.certificate_pem().encode(),
+                    cacerts=certificate.ca_certificates_pem().encode(),
                 )
 
             # If the region supports beacons, full registration of rack
diff --git a/src/provisioningserver/rpc/tests/test_clusterservice.py b/src/provisioningserver/rpc/tests/test_clusterservice.py
index d95a023..08d09b4 100644
--- a/src/provisioningserver/rpc/tests/test_clusterservice.py
+++ b/src/provisioningserver/rpc/tests/test_clusterservice.py
@@ -40,6 +40,7 @@ from maastesting.twisted import (
 from provisioningserver import concurrency
 from provisioningserver.certificates import (
     Certificate,
+    CertificateRequest,
     get_maas_cluster_cert_paths,
 )
 from provisioningserver.dhcp.testing.config import (
@@ -1322,7 +1323,10 @@ class TestClusterClientClusterCertificatesAreStored(TestClusterClientBase):
 
         client = self.make_running_client()
 
-        certificate = Certificate.generate_self_signed_v3("maas")
+        maasca = Certificate.generate_ca_certificate("maas")
+        certificate_request = CertificateRequest.generate("request")
+        certificate = maasca.sign_certificate_request(certificate_request)
+
         callRemote = self.patch_autospec(client, "callRemote")
         callRemote.side_effect = always_succeed_with(
             {
@@ -1332,6 +1336,7 @@ class TestClusterClientClusterCertificatesAreStored(TestClusterClientBase):
                         {
                             "cert": certificate.certificate_pem(),
                             "key": certificate.private_key_pem(),
+                            "cacerts": certificate.ca_certificates_pem(),
                         }
                     )
                 ),
@@ -1345,6 +1350,7 @@ class TestClusterClientClusterCertificatesAreStored(TestClusterClientBase):
             (
                 f"{certs_dir}/cluster.pem",
                 f"{certs_dir}/cluster.key",
+                f"{certs_dir}/cacerts.pem",
             ),
         )
 
diff --git a/src/provisioningserver/tests/test_certificates.py b/src/provisioningserver/tests/test_certificates.py
index 348aa2d..4e12897 100644
--- a/src/provisioningserver/tests/test_certificates.py
+++ b/src/provisioningserver/tests/test_certificates.py
@@ -11,6 +11,7 @@ from maastesting.testcase import MAASTestCase
 from provisioningserver.certificates import (
     Certificate,
     CertificateError,
+    CertificateRequest,
     get_maas_cert_tuple,
     get_maas_cluster_cert_paths,
     store_maas_cluster_cert_tuple,
@@ -18,6 +19,26 @@ from provisioningserver.certificates import (
 from provisioningserver.testing.certificates import get_sample_cert
 
 
+class TestCertificateRequest(MAASTestCase):
+    def test_certificate_request(self):
+        request = CertificateRequest.generate(
+            "maas", "organization", "organization_unit", 2048, b"DNS:*"
+        )
+        self.assertIsInstance(request.csr, crypto.X509Req)
+        self.assertIsInstance(request.key, crypto.PKey)
+
+        self.assertEqual(request.csr.get_subject().CN, "maas")
+        self.assertEqual(request.csr.get_subject().O, "organization")
+        self.assertEqual(request.csr.get_subject().OU, "organization_unit")
+
+        self.assertEqual(len(request.csr.get_extensions()), 1)
+        subject_alt_name_extension = request.csr.get_extensions()[0]
+        self.assertEqual(
+            subject_alt_name_extension.get_short_name(), b"subjectAltName"
+        )
+        self.assertEqual(subject_alt_name_extension.get_critical(), False)
+
+
 class TestCertificate(MAASTestCase):
     def setUp(self):
         super().setUp()
@@ -212,13 +233,12 @@ class TestCertificate(MAASTestCase):
             cert.not_before(),
         )
 
-    def test_generate_self_signed_v3(self):
-        cert = Certificate.generate_self_signed_v3(
+    def test_generate_ca_certificate(self):
+        cert = Certificate.generate_ca_certificate(
             "maas",
             organization_name="test",
             organizational_unit_name="unit",
             validity=timedelta(days=100),
-            subject_alternative_name=b"DNS:*",
         )
 
         self.assertIsInstance(cert.cert, crypto.X509)
@@ -246,7 +266,7 @@ class TestCertificate(MAASTestCase):
         self.assertEqual(
             x509certificate.get_version(), crypto.x509.Version.v3.value
         )
-        self.assertEqual(x509certificate.get_extension_count(), 2)
+        self.assertEqual(x509certificate.get_extension_count(), 1)
 
         # Extensions are kept in order
         basic_constraints_extension = x509certificate.get_extension(0)
@@ -255,11 +275,52 @@ class TestCertificate(MAASTestCase):
         )
         self.assertEqual(basic_constraints_extension.get_critical(), True)
 
-        subject_alt_name_extension = x509certificate.get_extension(1)
+    def test_sign_certificate(self):
+        certificate_request = CertificateRequest.generate(
+            "maas", "organization", "organization_unit", 2048, b"DNS:*"
+        )
+        ca = Certificate.generate_ca_certificate(
+            "maas",
+            organization_name="test",
+            organizational_unit_name="unit",
+            validity=timedelta(days=100),
+        )
+
+        signed_certificate = ca.sign_certificate_request(certificate_request)
+
+        self.assertIsInstance(signed_certificate.cert, crypto.X509)
+        self.assertIsInstance(signed_certificate.key, crypto.PKey)
+        self.assertEqual(len(signed_certificate.ca_certs), 1)
+        self.assertEqual(signed_certificate.cert.get_subject().CN, "maas")
+        self.assertEqual(signed_certificate.key, certificate_request.key)
+        self.assertLessEqual(
+            datetime.utcnow() + timedelta(days=-1),
+            signed_certificate.not_before(),
+        )
+        self.assertGreater(
+            signed_certificate.expiration(),
+            signed_certificate.not_before(),
+        )
         self.assertEqual(
-            subject_alt_name_extension.get_short_name(), b"subjectAltName"
+            len(certificate_request.csr.get_extensions()),
+            signed_certificate.cert.get_extension_count(),
         )
-        self.assertEqual(subject_alt_name_extension.get_critical(), False)
+        self.assertEqual(
+            certificate_request.csr.get_extensions()[0].get_short_name(),
+            signed_certificate.cert.get_extension(0).get_short_name(),
+        )
+        self.assertEqual(
+            signed_certificate.cert.get_issuer(), ca.cert.get_subject()
+        )
+
+        # check that the signed certificate is valid
+        store = crypto.X509Store()
+        store.add_cert(ca.cert)
+        context = crypto.X509StoreContext(store, signed_certificate.cert)
+        try:
+            context.verify_certificate()
+        except crypto.X509StoreContextError:
+            self.fail("Failed to verify the signed certificate.")
 
 
 class TestClusterCertificates(MAASTestCase):
@@ -279,11 +340,13 @@ class TestClusterCertificates(MAASTestCase):
         certs_dir.mkdir(parents=True)
         (certs_dir / "cluster.pem").touch()
         (certs_dir / "cluster.key").touch()
+        (certs_dir / "cacerts.pem").touch()
         self.assertEqual(
             get_maas_cluster_cert_paths(),
             (
                 f"{certs_dir}/cluster.pem",
                 f"{certs_dir}/cluster.key",
+                f"{certs_dir}/cacerts.pem",
             ),
         )
 
@@ -291,10 +354,17 @@ class TestClusterCertificates(MAASTestCase):
         certs_dir = self.tempdir / "certificates"
         certs_dir.mkdir(parents=True)
         self.useFixture(EnvironmentVariable("MAAS_ROOT", str(self.tempdir)))
-        store_maas_cluster_cert_tuple(b"private_key", b"certificate")
-        certificate_path, private_key_path = get_maas_cluster_cert_paths()
+        store_maas_cluster_cert_tuple(
+            b"private_key", b"certificate", b"cacerts"
+        )
+        (
+            certificate_path,
+            private_key_path,
+            cacerts_path,
+        ) = get_maas_cluster_cert_paths()
         self.assertEqual(Path(certificate_path).lstat().st_mode & 0o777, 0o644)
         self.assertEqual(Path(private_key_path).lstat().st_mode & 0o777, 0o600)
+        self.assertEqual(Path(cacerts_path).lstat().st_mode & 0o777, 0o644)
 
 
 class TestGetMAASCertTuple(MAASTestCase):