[Fix] Store vault kv secret ids in peer relationship

Getting a secret by label only returns the short form of a secret id
(without the model's UUID), which is fine when the two applications are
inside the same model, but leads to error when in different models.

Creating a secret always returns the long form of a secret id,
therefore, storing the secret id in the peer relationship with key:
label, value: secret id, to ensure long form of secret id.

lib/charms/vault_k8s/v0/vault_kv.py

@@ -106,7 +106,7 @@ def _on_update_status(self, event):
 
 import json
 import logging
-from typing import Any, Dict, Optional
+from typing import Any, Dict, List, Optional, Union
 
 import ops
 
@@ -230,17 +230,22 @@ def set_unit_credentials(self, relation: ops.Relation, nonce: str, secret: ops.S
             )
             return
         credentials[nonce] = secret.id
-
         relation.data[self.charm.app]["credentials"] = json.dumps(credentials, sort_keys=True)
 
-    def get_unit_credentials(self, relation: ops.Relation, nonce: str) -> Optional[str]:
-        """Get the unit credentials from the relation.
+    def remove_unit_credentials(self, relation: ops.Relation, nonce: Union[str, List[str]]):
+        """Remove nonce(s) from the relation."""
+        if not self.charm.unit.is_leader():
+            return
+
+        if isinstance(nonce, str):
+            nonce = [nonce]
 
-        Return None if the unit credentials are not set.
-        Return a juju secret id if the unit credentials are set.
-        """
         credentials = self.get_credentials(relation)
-        return credentials.get(nonce)
+
+        for n in nonce:
+            credentials.pop(n, None)
+
+        relation.data[self.charm.app]["credentials"] = json.dumps(credentials, sort_keys=True)
 
     def get_credentials(self, relation: ops.Relation) -> dict:
         """Get the unit credentials from the relation."""
@@ -415,11 +420,7 @@ def request_credentials(self, relation: ops.Relation, egress_subnet: str) -> Non
 
         A change in egress_subnet can happen when the pod is rescheduled to a different
         node by the underlying substrate without a change from Juju.
-
-        Only update the egress_subnet if it is already set.
         """
-        if not self.charm.unit.is_leader():
-            return
         self._set_unit_egress_subnet(relation, egress_subnet)
         self._set_unit_nonce(relation, self.nonce)
 

src/charm.py

@@ -51,7 +51,7 @@
 TLS_CA_FILE_PATH = "/vault/certs/ca.pem"
 PEER_RELATION_NAME = "vault-peers"
 KV_RELATION_NAME = "vault-kv"
-KV_SECRET_PREFIX = "kv_creds_"
+KV_SECRET_PREFIX = "kv-creds-"
 
 
 class PeerSecretError(Exception):
@@ -281,6 +281,7 @@ def _on_new_vault_kv_client_attached(self, event: NewVaultKvClientAttachedEvent)
             self.unit.status = WaitingStatus("Waiting for peer relation")
             event.defer()
             return
+
         try:
             root_token, _ = self._get_initialization_secret_from_peer_relation()
         except PeerSecretError:
@@ -298,6 +299,7 @@ def _on_new_vault_kv_client_attached(self, event: NewVaultKvClientAttachedEvent)
             self.unit.status = WaitingStatus("Waiting for vault certificate to be available")
             event.defer()
             return
+
         relation = self.model.get_relation(event.relation_name, event.relation_id)
 
         if relation is None or relation.app is None:
@@ -320,6 +322,7 @@ def _on_new_vault_kv_client_attached(self, event: NewVaultKvClientAttachedEvent)
             self.vault_kv.set_vault_url(relation, vault_url)
         self.vault_kv.set_ca_certificate(relation, ca_certificate)
 
+        nonces = []
         for unit in relation.units:
             egress_subnet = relation.data[unit].get("egress_subnet")
             nonce = relation.data[unit].get("nonce")
@@ -329,20 +332,35 @@ def _on_new_vault_kv_client_attached(self, event: NewVaultKvClientAttachedEvent)
                     unit.name,
                 )
                 continue
-            policy_name = role_name = mount + "-" + unit.name.replace("/", "-")
-            vault.configure_kv_policy(policy_name, mount)
-            role_id = vault.configure_approle(role_name, [egress_subnet], [policy_name])
-            unit_secret = self.vault_kv.get_unit_credentials(relation, nonce)
-            secret = self._create_or_update_kv_secret(
-                vault,
-                relation,
-                role_id,
-                role_name,
-                egress_subnet,
-                unit_secret,
-            )
-            if secret is not None:
-                self.vault_kv.set_unit_credentials(relation, nonce, secret)
+            nonces.append(nonce)
+            self._ensure_unit_credentials(vault, relation, unit.name, mount, nonce, egress_subnet)
+
+        # Remove any stale nonce
+        credential_nonces = self.vault_kv.get_credentials(relation).keys()
+        stale_nonces = list(set(credential_nonces) - set(nonces))
+        self.vault_kv.remove_unit_credentials(relation, stale_nonces)
+
+    def _ensure_unit_credentials(
+        self,
+        vault: Vault,
+        relation: Relation,
+        unit_name: str,
+        mount: str,
+        nonce: str,
+        egress_subnet: str,
+    ):
+        """Ensures a unit has credentials to access the vault-kv mount."""
+        policy_name = role_name = mount + "-" + unit_name.replace("/", "-")
+        vault.configure_kv_policy(policy_name, mount)
+        role_id = vault.configure_approle(role_name, [egress_subnet], [policy_name])
+        secret = self._create_or_update_kv_secret(
+            vault,
+            relation,
+            role_id,
+            role_name,
+            egress_subnet,
+        )
+        self.vault_kv.set_unit_credentials(relation, nonce, secret)
 
     def _create_or_update_kv_secret(
         self,
@@ -351,30 +369,86 @@ def _create_or_update_kv_secret(
         role_id: str,
         role_name: str,
         egress_subnet: str,
-        secret_id: Optional[str],
-    ) -> Optional[Secret]:
-        """Create or update a KV secret for a unit."""
+    ) -> Secret:
+        """Create or update a KV secret for a unit.
+
+        Fetch secret id from peer relation, if it exists, update the secret,
+        otherwise create it.
+        """
         label = KV_SECRET_PREFIX + role_name
-        try:
-            secret = self.model.get_secret(id=secret_id, label=label)
-            credentials = secret.get_content()
-            role_secret_id_data = vault.read_role_secret(role_name, credentials["role-secret-id"])
-            # if unit subnet is already in cidr_list, skip
-            if egress_subnet in role_secret_id_data["cidr_list"]:
-                return None
-            credentials["role-secret-id"] = vault.generate_role_secret_id(
-                role_name, [egress_subnet]
+        secret_id = self._get_vault_kv_secret_in_peer_relation(label)
+        if secret_id is None:
+            return self._create_kv_secret(
+                vault, relation, role_id, role_name, egress_subnet, label
             )
-            secret.set_content(credentials)
-        except SecretNotFoundError:
-            role_secret_id = vault.generate_role_secret_id(role_name, [egress_subnet])
-            secret = self.app.add_secret(
-                {"role-id": role_id, "role-secret-id": role_secret_id},
-                label=label,
+        else:
+            return self._update_kv_secret(
+                vault, relation, role_name, egress_subnet, label, secret_id
             )
+
+    def _create_kv_secret(
+        self,
+        vault: Vault,
+        relation: Relation,
+        role_id: str,
+        role_name: str,
+        egress_subnet: str,
+        label: str,
+    ) -> Secret:
+        """Create a vault kv secret, store its id in the peer relation and return it."""
+        role_secret_id = vault.generate_role_secret_id(role_name, [egress_subnet])
+        secret = self.app.add_secret(
+            {"role-id": role_id, "role-secret-id": role_secret_id},
+            label=label,
+        )
+        if secret.id is None:
+            raise RuntimeError(f"Unexpected error, just created secret {label!r} has no id")
+        self._set_vault_kv_secret_in_peer_relation(label, secret.id)
         secret.grant(relation)
         return secret
 
+    def _update_kv_secret(
+        self,
+        vault: Vault,
+        relation: Relation,
+        role_name: str,
+        egress_subnet: str,
+        label: str,
+        secret_id: str,
+    ) -> Secret:
+        """Update a vault kv secret if the unit subnet is not in the cidr list."""
+        secret = self.model.get_secret(id=secret_id, label=label)
+        secret.grant(relation)
+        credentials = secret.get_content()
+        role_secret_id_data = vault.read_role_secret(role_name, credentials["role-secret-id"])
+        # if unit subnet is already in cidr_list, skip
+        if egress_subnet in role_secret_id_data["cidr_list"]:
+            return secret
+        credentials["role-secret-id"] = vault.generate_role_secret_id(role_name, [egress_subnet])
+        secret.set_content(credentials)
+        return secret
+
+    def _get_vault_kv_secrets_in_peer_relation(self) -> Dict[str, str]:
+        """Return the vault kv secrets from the peer relation."""
+        if not self._is_peer_relation_created():
+            raise RuntimeError("Peer relation not created")
+        relation = self.model.get_relation(PEER_RELATION_NAME)
+        secrets = json.loads(relation.data[self.app].get("vault-kv-secrets", "{}"))  # type: ignore[union-attr]  # noqa: E501
+        return secrets
+
+    def _get_vault_kv_secret_in_peer_relation(self, label: str) -> Optional[str]:
+        """Return the vault kv secret id associated to input label from peer relation."""
+        return self._get_vault_kv_secrets_in_peer_relation().get(label)
+
+    def _set_vault_kv_secret_in_peer_relation(self, label: str, secret_id: str):
+        """Set the vault kv secret in the peer relation."""
+        if not self._is_peer_relation_created():
+            raise RuntimeError("Peer relation not created")
+        secrets = self._get_vault_kv_secrets_in_peer_relation()
+        secrets[label] = secret_id
+        relation = self.model.get_relation(PEER_RELATION_NAME)
+        relation.data[self.app].update({"vault-kv-secrets": json.dumps(secrets, sort_keys=True)})  # type: ignore[union-attr]  # noqa: E501
+
     def _delete_vault_data(self) -> None:
         """Delete Vault's data."""
         try: