fix: Update intermediate CA and PKI role when common_name is changed. (…

…#377)

Co-authored-by: Guillaume Belanger <[email protected]>

charmcraft.yaml

@@ -166,3 +166,4 @@ config:
       description: |
         The common name that will be used by Vault as an intermediate CA. This will only be used when the charm is 
         configured to use a Vault PKI backend through the `vault-pki` relation.
+        The charm will only issue certificates for the subdomains under the `common_name` specified.

lib/charms/vault_k8s/v0/vault_client.py

@@ -26,7 +26,7 @@
 
 # Increment this PATCH version before using `charmcraft publish-lib` or reset
 # to 0 if you are raising the major API version
-LIBPATCH = 11
+LIBPATCH = 12
 
 
 RAFT_STATE_ENDPOINT = "v1/sys/storage/raft/autopilot/state"
@@ -288,11 +288,6 @@ def enable_secrets_engine(self, backend_type: SecretsBackend, path: str) -> None
             else:
                 raise VaultClientError(e) from e
 
-    def is_intermediate_ca_set(self, mount: str, certificate: str) -> bool:
-        """Check if the intermediate CA is set for the PKI backend."""
-        intermediate_ca = self._client.secrets.pki.read_ca_certificate(mount_point=mount)
-        return intermediate_ca == certificate
-
     def get_intermediate_ca(self, mount: str) -> str:
         """Get the intermediate CA for the PKI backend."""
         return self._client.secrets.pki.read_ca_certificate(mount_point=mount)
@@ -353,12 +348,7 @@ def sign_pki_certificate_signing_request(
             logger.warning("Error while signing PKI certificate: %s", e)
             return None
 
-    def is_pki_ca_certificate_set(self, mount: str, certificate: str) -> bool:
-        """Check if the CA certificate is set for the PKI backend."""
-        existing_certificate = self._client.secrets.pki.read_ca_certificate(mount_point=mount)
-        return existing_certificate == certificate
-
-    def create_pki_charm_role(self, role: str, allowed_domains: str, mount: str) -> None:
+    def create_or_update_pki_charm_role(self, role: str, allowed_domains: str, mount: str) -> None:
         """Create a role for the PKI backend."""
         self._client.secrets.pki.create_or_update_role(
             name=role,
@@ -416,3 +406,13 @@ def get_num_raft_peers(self) -> int:
         """Return the number of raft peers."""
         raft_config = self._client.sys.read_raft_config()
         return len(raft_config["data"]["config"]["servers"])
+
+    def is_common_name_allowed_in_pki_role(self, role: str, mount: str, common_name: str) -> bool:
+        """Return whether the provided common name is in the list of domains allowed by the specified PKI role."""
+        try:
+            return common_name in self._client.secrets.pki.read_role(
+                name=role, mount_point=mount
+            ).get("data", {}).get("allowed_domains", [])
+        except InvalidPath:
+            logger.error("Role does not exist on the specified path.")
+            return False

src/charm.py

@@ -349,22 +349,27 @@ def _configure_pki_secrets_engine(self) -> None:
             return
         common_name = self._get_config_common_name()
         vault.enable_secrets_engine(SecretsBackend.PKI, PKI_MOUNT)
-        if not self._is_intermediate_ca_set(vault, common_name):
+        if not self._is_intermediate_ca_common_name_valid(vault, common_name):
             csr = vault.generate_pki_intermediate_ca_csr(mount=PKI_MOUNT, common_name=common_name)
             self.tls_certificates_pki.request_certificate_creation(
                 certificate_signing_request=csr.encode(),
                 is_ca=True,
             )
             self._set_pki_csr_secret(csr)
 
-    def _is_intermediate_ca_set(self, vault: Vault, common_name: str) -> bool:
-        """Check if the intermediate CA is set in the PKI secrets engine."""
+    def _is_intermediate_ca_common_name_valid(self, vault: Vault, common_name: str) -> bool:
+        """Check if the intermediate CA is set with the valid common name."""
         intermediate_ca = vault.get_intermediate_ca(mount=PKI_MOUNT)
         if not intermediate_ca:
             return False
         intermediate_ca_common_name = get_common_name_from_certificate(intermediate_ca)
         return intermediate_ca_common_name == common_name
 
+    def _is_intermediate_ca_set(self, vault: Vault, certificate: str) -> bool:
+        """Check if the intermediate CA is set in the PKI secrets engine."""
+        intermediate_ca = vault.get_intermediate_ca(mount=PKI_MOUNT)
+        return certificate == intermediate_ca
+
     def _add_ca_certificate_to_pki_secrets_engine(self) -> None:
         """Add the CA certificate to the PKI secrets engine."""
         if not self.unit.is_leader():
@@ -382,10 +387,13 @@ def _add_ca_certificate_to_pki_secrets_engine(self) -> None:
         if not certificate:
             logger.debug("No certificate available")
             return
-        if not vault.is_intermediate_ca_set(mount=PKI_MOUNT, certificate=certificate):
+        if (not self._is_intermediate_ca_common_name_valid(vault, common_name) or
+            not self._is_intermediate_ca_set(vault, certificate)):
             vault.set_pki_intermediate_ca_certificate(certificate=certificate, mount=PKI_MOUNT)
-        if not vault.is_pki_role_created(role=PKI_ROLE, mount=PKI_MOUNT):
-            vault.create_pki_charm_role(
+        if not vault.is_common_name_allowed_in_pki_role(
+            role=PKI_ROLE, mount=PKI_MOUNT, common_name=common_name
+        ):
+            vault.create_or_update_pki_charm_role(
                 allowed_domains=common_name,
                 mount=PKI_MOUNT,
                 role=PKI_ROLE,

tests/unit/test_charm.py

@@ -1505,6 +1505,55 @@ def test_given_vault_is_available_when_tls_certificates_pki_relation_joined_then
             certificate_signing_request=csr.encode(), is_ca=True
         )
 
+    @patch("ops.model.Container.restart", new=Mock)
+    @patch("charm.get_common_name_from_certificate", new=Mock)
+    @patch(f"{TLS_CERTIFICATES_LIB_PATH}.TLSCertificatesRequiresV3.request_certificate_creation")
+    def test_given_vault_pki_configured_when_common_name_is_changed_then_new_certificate_request_is_made(
+        self,
+        patch_request_certificate_creation,
+    ):
+        csr = "some csr content"
+        self.harness.update_config({"common_name": "vault"})
+        self.mock_vault.configure_mock(
+            **{
+                "is_initialized.return_value": True,
+                "is_api_available.return_value": True,
+                "get_intermediate_ca.return_value": "vault",
+                "generate_pki_intermediate_ca_csr.return_value": csr,
+                "is_sealed.return_value": False,
+            },
+        )
+        self.harness.set_leader(is_leader=True)
+        self.harness.add_storage(storage_name="config", attach=True)
+        self.harness.set_can_connect(container=self.container_name, val=True)
+        self._set_peer_relation()
+        self._set_approle_secret(
+            role_id="root token content",
+            secret_id="whatever secret id",
+        )
+
+        relation_id = self.harness.add_relation(
+            relation_name=TLS_CERTIFICATES_PKI_RELATION_NAME, remote_app="tls-provider"
+        )
+        self.harness.add_relation_unit(relation_id, "tls-provider/0")
+
+        self.mock_vault.enable_secrets_engine.assert_called_with(SecretsBackend.PKI, "charm-pki")
+        self.mock_vault.generate_pki_intermediate_ca_csr.assert_called_with(
+            mount="charm-pki", common_name="vault"
+        )
+        patch_request_certificate_creation.assert_called_with(
+            certificate_signing_request=csr.encode(), is_ca=True
+        )
+        self.harness.update_config({"common_name": "new_common_name"})
+        self.mock_vault.generate_pki_intermediate_ca_csr.assert_called_with(
+            mount="charm-pki", common_name="new_common_name"
+        )
+        patch_request_certificate_creation.assert_called_with(
+            certificate_signing_request=csr.encode(), is_ca=True
+        )
+
+
+    @patch("charm.get_common_name_from_certificate", new=Mock)
     @patch(f"{TLS_CERTIFICATES_LIB_PATH}.TLSCertificatesRequiresV3.get_assigned_certificates")
     def test_given_vault_is_available_when_pki_certificate_is_available_then_certificate_added_to_vault_pki(
         self,
@@ -1514,11 +1563,11 @@ def test_given_vault_is_available_when_pki_certificate_is_available_then_certifi
             **{
                 "is_initialized.return_value": True,
                 "is_api_available.return_value": True,
-                "is_intermediate_ca_set.return_value": False,
                 "is_pki_role_created.return_value": False,
+                "get_intermediate_ca.return_value": "vault",
+                "is_common_name_allowed_in_pki_role.return_value": False,
             },
         )
-
         csr = "some csr content"
         certificate = "some certificate"
         ca = "some ca"
@@ -1562,10 +1611,87 @@ def test_given_vault_is_available_when_pki_certificate_is_available_then_certifi
             certificate=certificate,
             mount="charm-pki",
         )
-        self.mock_vault.create_pki_charm_role.assert_called_with(
+        self.mock_vault.create_or_update_pki_charm_role.assert_called_with(
+            allowed_domains="vault", mount="charm-pki", role="charm"
+        )
+
+    @patch("ops.model.Container.restart", new=Mock)
+    @patch("charm.get_common_name_from_certificate", new=Mock)
+    @patch(f"{TLS_CERTIFICATES_LIB_PATH}.TLSCertificatesRequiresV3.get_assigned_certificates")
+    def test_given_vault_pki_configured_when_common_name_is_changed_then_new_certificate_added_to_vault_pki(
+        self,
+        patch_get_assigned_certificates,
+    ):
+        csr = "some csr content"
+        self.harness.update_config({"common_name": "vault"})
+        self.mock_vault.configure_mock(
+            **{
+                "is_initialized.return_value": True,
+                "is_api_available.return_value": True,
+                "get_intermediate_ca.return_value": "vault",
+                "generate_pki_intermediate_ca_csr.return_value": csr,
+                "is_sealed.return_value": False,
+                "is_common_name_allowed_in_pki_role.return_value": False,
+            },
+        )
+        self.harness.set_leader(is_leader=True)
+        self.harness.add_storage(storage_name="config", attach=True)
+        self.harness.set_can_connect(container=self.container_name, val=True)
+        peer_relation_id = self._set_peer_relation()
+        self._set_approle_secret(
+            role_id="root token content",
+            secret_id="whatever secret id",
+        )
+        self._set_csr_secret_in_peer_relation(relation_id=peer_relation_id, csr=csr)
+        certificate = "some certificate"
+        ca = "some ca"
+        chain = [ca]
+        event = CertificateAvailableEvent(
+            handle=Mock(),
+            certificate=certificate,
+            certificate_signing_request=csr,
+            ca=ca,
+            chain=chain,
+        )
+        relation_id = self.harness.add_relation(
+            relation_name=TLS_CERTIFICATES_PKI_RELATION_NAME, remote_app="tls-provider"
+        )
+
+        patch_get_assigned_certificates.return_value = [
+            ProviderCertificate(
+                relation_id=relation_id,
+                application_name="tls-provider",
+                csr=csr,
+                certificate=certificate,
+                ca=ca,
+                chain=chain,
+                revoked=False,
+                expiry_time=datetime.now(timezone.utc),
+            )
+        ]
+
+        self.harness.charm._on_tls_certificate_pki_certificate_available(event=event)
+
+        self.mock_vault.set_pki_intermediate_ca_certificate.assert_called_with(
+            certificate=certificate,
+            mount="charm-pki",
+        )
+        self.mock_vault.create_or_update_pki_charm_role.assert_called_with(
             allowed_domains="vault", mount="charm-pki", role="charm"
         )
 
+        self.harness.update_config({"common_name": "new_common_name"})
+
+        self.harness.charm._on_tls_certificate_pki_certificate_available(event=event)
+
+        self.mock_vault.set_pki_intermediate_ca_certificate.assert_called_with(
+            certificate=certificate,
+            mount="charm-pki",
+        )
+        self.mock_vault.create_or_update_pki_charm_role.assert_called_with(
+            allowed_domains="new_common_name", mount="charm-pki", role="charm"
+        )
+
     @patch(f"{TLS_CERTIFICATES_LIB_PATH}.TLSCertificatesProvidesV3.set_relation_certificate")
     @patch("charm.get_common_name_from_csr")
     def test_given_vault_available_when_vault_pki_certificate_creation_request_then_certificate_is_provided(