diff --git a/lib/charms/vault_k8s/v0/vault_kv.py b/lib/charms/vault_k8s/v0/vault_kv.py
index cc7076a9..872f6642 100644
--- a/lib/charms/vault_k8s/v0/vault_kv.py
+++ b/lib/charms/vault_k8s/v0/vault_kv.py
@@ -2,7 +2,78 @@
 # Copyright 2023 Canonical Ltd.
 # See LICENSE file for licensing details.
 
-"""Contains the VaultKVProvides class."""
+"""Library for the vault-kv relation.
+
+This library contains the Requires and Provides classes for handling the vault-kv
+interface.
+
+## Getting Started
+From a charm directory, fetch the library using `charmcraft`:
+
+```shell
+charmcraft fetch-lib charms.vault_k8s.v0.vault_kv
+```
+
+### Requirer charm
+The requirer charm is the charm requiring a secret value store. In this example, the requirer charm
+is requiring a secret value store.
+
+from charms.vault_k8s.v0 import vault_kv
+from ops.charm import CharmBase, InstallEvent
+from ops.main import main
+from ops.model import ActiveStatus, BlockedStatus
+
+
+class ExampleRequirerCharm(CharmBase):
+    def __init__(self, *args):
+        super().__init__(*args)
+        self.interface = vault_kv.VaultKvRequires(
+            self,
+            "secrets",
+            "banana",
+            "10.20.20.4/32",
+        )
+
+        self.framework.observe(self.interface.on.ready, self._on_ready)
+        self.framework.observe(self.interface.on.gone_away, self._on_gone_away)
+
+    def _on_install(self, event: InstallEvent):
+        self.unit.status = BlockedStatus("Waiting for vault-kv relation")
+
+    def _on_ready(self, event: vault_kv.VaultKvReadyEvent):
+        relation = self.model.get_relation(event.relation_name, event.relation_id)
+        if relation is None:
+            return
+        vault_url = self.interface.vault_url(relation)
+        mount = self.interface.mount(relation)
+
+        unit_credentials = self.interface.unit_credentials(relation)
+        # unit_credentials is a juju secret id
+        secret = self.model.get_secret(id=unit_credentials)
+        secret_content = secret.get_content(refresh=True)
+        role_id = secret_content["role-id"]
+        role_secret_id = secret_content["role-secret-id"]
+
+        self._configure(vault_url, mount, role_id, role_secret_id)
+
+        self.unit.status = ActiveStatus()
+
+    def _on_gone_away(self, event: vault_kv.VaultKvGoneAwayEvent):
+        self.unit.status = BlockedStatus("Waiting for vault-kv relation")
+
+    def _configure(self, vault_url: str, mount: str, role_id: str, role_secret_id: str):
+        pass
+
+
+if __name__ == "__main__":
+    main(ExampleRequirerCharm)
+
+You can integrate both charms by running:
+
+```bash
+juju integrate <vault provider charm> <vault requirer charm>
+```
+"""
 
 import json
 import logging
@@ -24,13 +95,7 @@
 LIBPATCH = 1
 
 
-class HasVaultKvClientsEvent(ops.EventBase):
-    """Has VaultKvClients Event."""
-
-    pass
-
-
-class ReadyVaultKvClientsEvent(ops.EventBase):
+class NewVaultKvClientAttachedEvent(ops.EventBase):
     """Ready VaultKvClients Event."""
 
     def __init__(
@@ -38,19 +103,19 @@ def __init__(
         handle: ops.Handle,
         relation_id: int,
         relation_name: str,
-        secret_backend: str,
+        mount_suffix: str,
     ):
         super().__init__(handle)
         self.relation_id = relation_id
         self.relation_name = relation_name
-        self.secret_backend = secret_backend
+        self.mount_suffix = mount_suffix
 
     def snapshot(self) -> dict:
         """Return snapshot data that should be persisted."""
         return {
             "relation_id": self.relation_id,
             "relation_name": self.relation_name,
-            "secret_backend": self.secret_backend,
+            "mount_suffix": self.mount_suffix,
         }
 
     def restore(self, snapshot: Dict[str, Any]):
@@ -58,35 +123,19 @@ def restore(self, snapshot: Dict[str, Any]):
         super().restore(snapshot)
         self.relation_id = snapshot["relation_id"]
         self.relation_name = snapshot["relation_name"]
-        self.secret_backend = snapshot["secret_backend"]
-
-
-class DepartedVaultKvClientsEvent(ops.EventBase):
-    """Departed VaultKvClients Event."""
-
-    pass
-
-
-class GoneAwayVaultKvClientsEvent(ops.EventBase):
-    """GoneAway VaultKvClients Event."""
-
-    pass
+        self.mount_suffix = snapshot["mount_suffix"]
 
 
 class VaultKvProviderEvents(ops.ObjectEvents):
     """List of events that the Vault Kv provider charm can leverage."""
 
-    has_vault_kv_clients = ops.EventSource(HasVaultKvClientsEvent)
-    ready_vault_kv_clients = ops.EventSource(ReadyVaultKvClientsEvent)
-    departed_vault_kv_clients = ops.EventSource(DepartedVaultKvClientsEvent)
-    gone_away_vault_kv_clients = ops.EventSource(GoneAwayVaultKvClientsEvent)
+    new_vault_kv_client_attached = ops.EventSource(NewVaultKvClientAttachedEvent)
 
 
 class VaultKvProvides(ops.Object):
     """Class to be instanciated by the providing side of the relation."""
 
     on = VaultKvProviderEvents()
-    _stored = ops.StoredState()
 
     def __init__(
         self,
@@ -96,30 +145,10 @@ def __init__(
         super().__init__(charm, relation_name)
         self.charm = charm
         self.relation_name = relation_name
-        self.framework.observe(
-            self.charm.on[relation_name].relation_joined,
-            self._on_vault_kv_relation_joined,
-        )
         self.framework.observe(
             self.charm.on[relation_name].relation_changed,
             self._on_vault_kv_relation_changed,
         )
-        self.framework.observe(
-            self.charm.on[relation_name].relation_departed,
-            self._on_vault_kv_relation_departed,
-        )
-        self.framework.observe(
-            self.charm.on[relation_name].relation_broken,
-            self._on_vault_kv_relation_broken,
-        )
-
-    def _on_vault_kv_relation_joined(self, event: ops.RelationJoinedEvent):
-        """Handle client joined relation.
-
-        Args:
-            event: The event that triggered the handler.
-        """
-        self.on.has_vault_kv_clients.emit()
 
     def _on_vault_kv_relation_changed(self, event: ops.RelationChangedEvent):
         """Handle client changed relation."""
@@ -127,23 +156,15 @@ def _on_vault_kv_relation_changed(self, event: ops.RelationChangedEvent):
             logger.debug("No remote application yet")
             return
 
-        secret_backend = event.relation.data[event.app].get("secret_backend")
+        mount_suffix = event.relation.data[event.app].get("mount_suffix")
 
-        if secret_backend is not None:
-            self.on.ready_vault_kv_clients.emit(
+        if mount_suffix is not None:
+            self.on.new_vault_kv_client_attached.emit(
                 event.relation.id,
                 event.relation.name,
-                secret_backend,
+                mount_suffix,
             )
 
-    def _on_vault_kv_relation_departed(self, event: ops.RelationDepartedEvent):
-        """Handle client departed relation."""
-        self.on.departed_vault_kv_clients.emit()
-
-    def _on_vault_kv_relation_broken(self, event: ops.RelationBrokenEvent):
-        """Handle client broken relation."""
-        self.on.gone_away_vault_kv_clients.emit()
-
     def set_vault_url(self, relation: ops.Relation, vault_url: str):
         """Set the vault_url on the relation."""
         if not self.charm.unit.is_leader():
@@ -151,30 +172,34 @@ def set_vault_url(self, relation: ops.Relation, vault_url: str):
 
         relation.data[self.charm.app]["vault_url"] = vault_url
 
-    def set_kv_mountpoint(self, relation: ops.Relation, kv_mountpoint: str):
-        """Set the kv_mountpoint on the relation."""
+    def set_mount(self, relation: ops.Relation, mount: str):
+        """Set the mount on the relation."""
         if not self.charm.unit.is_leader():
             return
 
-        relation.data[self.charm.app]["kv_mountpoint"] = kv_mountpoint
+        relation.data[self.charm.app]["mount"] = mount
 
-    def set_unit_credentials(
-        self, relation: ops.Relation, name: str, role_id: str, role_secret_id: str
-    ):
+    def set_unit_credentials(self, relation: ops.Relation, unit_name: str, secret: ops.Secret):
         """Set the unit credentials on the relation."""
         if not self.charm.unit.is_leader():
             return
 
         credentials = self.get_credentials(relation)
-        credentials[name] = {"role_id": role_id, "role_secret_id": role_secret_id}
+        if secret.id is None:
+            logger.debug("Secret id is None, not updating the relation")
+            return
+        credentials[unit_name] = secret.id
 
         relation.data[self.charm.app]["credentials"] = json.dumps(credentials, sort_keys=True)
 
-    def get_role_secret_id(self, relation: ops.Relation, name: str) -> Optional[str]:
-        """Get the role_secret_id from the relation."""
+    def get_unit_credentials(self, relation: ops.Relation, unit_name: str) -> Optional[str]:
+        """Get the unit credentials from the relation.
+
+        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)
-        unit_credentials = credentials.get(name, {})
-        return unit_credentials.get("role_secret_id")
+        return credentials.get(unit_name)
 
     def get_credentials(self, relation: ops.Relation) -> dict:
         """Get the unit credentials from the relation."""
@@ -196,17 +221,15 @@ def __init__(
         relation_id: int,
         relation_name: str,
         vault_url: str,
-        kv_mountpoint: str,
-        role_id: str,
-        role_secret_id: str,
+        mount: str,
+        credentials_secret: str,
     ):
         super().__init__(handle)
         self.relation_id = relation_id
         self.relation_name = relation_name
         self.vault_url = vault_url
-        self.kv_mountpoint = kv_mountpoint
-        self.role_id = role_id
-        self.role_secret_id = role_secret_id
+        self.mount = mount
+        self.credentials_secret = credentials_secret
 
     def snapshot(self) -> dict:
         """Return snapshot data that should be persisted."""
@@ -214,9 +237,8 @@ def snapshot(self) -> dict:
             "relation_id": self.relation_id,
             "relation_name": self.relation_name,
             "vault_url": self.vault_url,
-            "kv_mountpoint": self.kv_mountpoint,
-            "role_id": self.role_id,
-            "role_secret_id": self.role_secret_id,
+            "mount": self.mount,
+            "credentials_secret": self.credentials_secret,
         }
 
     def restore(self, snapshot: Dict[str, Any]):
@@ -225,9 +247,8 @@ def restore(self, snapshot: Dict[str, Any]):
         self.relation_id = snapshot["relation_id"]
         self.relation_name = snapshot["relation_name"]
         self.vault_url = snapshot["vault_url"]
-        self.kv_mountpoint = snapshot["kv_mountpoint"]
-        self.role_id = snapshot["role_id"]
-        self.role_secret_id = snapshot["role_secret_id"]
+        self.mount = snapshot["mount"]
+        self.credentials_secret = snapshot["credentials_secret"]
 
 
 class VaultKvGoneAwayEvent(ops.EventBase):
@@ -248,19 +269,18 @@ class VaultKvRequires(ops.Object):
     """Class to be instanciated by the requiring side of the relation."""
 
     on = VaultKvRequireEvents()
-    _stored = ops.StoredState()
 
     def __init__(
         self,
         charm: ops.CharmBase,
         relation_name: str,
-        secret_backend: str,
+        mount_suffix: str,
         egress_subnet: str,
     ) -> None:
         super().__init__(charm, relation_name)
         self.charm = charm
         self.relation_name = relation_name
-        self.secret_backend = secret_backend
+        self.mount_suffix = mount_suffix
         self.egress_subnet = egress_subnet
         self.framework.observe(
             self.charm.on[relation_name].relation_joined,
@@ -281,11 +301,11 @@ def __init__(
     def _update_unit_egress_subnet(self, force: bool = False):
         """Update egress_subnet for every instance of the relation.
 
-        Secret ids are generated based on the egress_subnet, so if the egress_subnet changes
-        a new secret id must be generated.
+        Generated secret ids are tied to the unit egress_subnet, so if the egress_subnet
+        changes a new secret id must be generated.
 
-        A change in egress_subnet can happend when the pod is rescheduled to a different node by
-        the underlying substrate without a change from Juju.
+        A change in egress_subnet can happend when the pod is rescheduled to a different
+        node by the underlying substrate without a change from Juju.
         """
         for relation in self.model.relations[self.relation_name]:
             unit_databag = relation.data[self.charm.unit]
@@ -293,9 +313,6 @@ def _update_unit_egress_subnet(self, force: bool = False):
             if force or unit_egress_subnet != self.egress_subnet:
                 unit_databag["egress_subnet"] = self.egress_subnet
 
-    def _on_vault_kv_relation_created(self, event: ops.RelationCreatedEvent):
-        pass
-
     def _on_vault_kv_relation_joined(self, event: ops.RelationJoinedEvent):
         """Handle relation joined.
 
@@ -304,7 +321,7 @@ def _on_vault_kv_relation_joined(self, event: ops.RelationJoinedEvent):
         """
         self.on.connected.emit()
         if self.charm.unit.is_leader():
-            event.relation.data[self.charm.app]["secret_backend"] = self.secret_backend
+            event.relation.data[self.charm.app]["mount_suffix"] = self.mount_suffix
         self._update_unit_egress_subnet(force=True)
 
     def _on_vault_kv_relation_changed(self, event: ops.RelationChangedEvent):
@@ -314,20 +331,15 @@ def _on_vault_kv_relation_changed(self, event: ops.RelationChangedEvent):
             return
 
         vault_url = self.vault_url(event.relation)
-        kv_mountpoint = self.kv_mountpoint(event.relation)
-        unit_credentials = self.unit_credentials(event.relation)
-        if unit_credentials is None:
-            return
-        role_id = unit_credentials.get("role_id")
-        role_secret_id = unit_credentials.get("role_secret_id")
-        if all((vault_url, kv_mountpoint, role_id, role_secret_id)):
+        mount = self.mount(event.relation)
+        unit_credentials_secret = self.unit_credentials(event.relation)
+        if all((vault_url, mount, unit_credentials_secret)):
             self.on.ready.emit(
                 event.relation.id,
                 event.relation.name,
                 vault_url,
-                kv_mountpoint,
-                role_id,
-                role_secret_id,
+                mount,
+                unit_credentials_secret,
             )
 
     def _on_vault_kv_relation_broken(self, event: ops.RelationBrokenEvent):
@@ -340,30 +352,19 @@ def vault_url(self, relation: ops.Relation) -> Optional[str]:
             return None
         return relation.data[relation.app].get("vault_url")
 
-    def kv_mountpoint(self, relation: ops.Relation) -> Optional[str]:
-        """Return the kv_mountpoint from the relation."""
+    def mount(self, relation: ops.Relation) -> Optional[str]:
+        """Return the mount from the relation."""
         if relation.app is None:
             return None
-        return relation.data[relation.app].get("kv_mountpoint")
+        return relation.data[relation.app].get("mount")
 
-    def unit_credentials(self, relation: ops.Relation) -> Optional[dict]:
-        """Return the unit credentials from the relation."""
+    def unit_credentials(self, relation: ops.Relation) -> Optional[str]:
+        """Return the unit credentials from the relation.
+
+        Unit credentials are stored in the relation data as a Juju secret id.
+        """
         if relation.app is None:
             return None
         return json.loads(relation.data[relation.app].get("credentials", "{}")).get(
             self._unit_name
         )
-
-    def role_id(self, relation: ops.Relation) -> Optional[str]:
-        """Return the role_id from the relation."""
-        credentials = self.unit_credentials(relation)
-        if credentials is None:
-            return None
-        return credentials.get("role_id")
-
-    def role_secret_id(self, relation: ops.Relation) -> Optional[str]:
-        """Return the role_secret_id from the relation."""
-        credentials = self.unit_credentials(relation)
-        if credentials is None:
-            return None
-        return credentials.get("role_secret_id")
diff --git a/src/charm.py b/src/charm.py
index 8f94e8dc..d327c9b0 100755
--- a/src/charm.py
+++ b/src/charm.py
@@ -9,7 +9,7 @@
 
 import json
 import logging
-from typing import Optional
+from typing import List, Optional
 
 from charms.observability_libs.v1.kubernetes_service_patch import (
     KubernetesServicePatch,
@@ -19,18 +19,20 @@
     CertificateCreationRequestEvent,
     TLSCertificatesProvidesV2,
 )
-from charms.vault_k8s.v0.vault_kv import (
-    GoneAwayVaultKvClientsEvent,
-    ReadyVaultKvClientsEvent,
-    VaultKvProvides,
-)
-from ops.charm import ActionEvent, CharmBase, ConfigChangedEvent
+from charms.vault_k8s.v0.vault_kv import NewVaultKvClientAttachedEvent, VaultKvProvides
+from ops.charm import ActionEvent, CharmBase, ConfigChangedEvent, SecretRotateEvent
 from ops.framework import StoredState
 from ops.main import main
-from ops.model import ActiveStatus, BlockedStatus, MaintenanceStatus, Relation
+from ops.model import (
+    ActiveStatus,
+    BlockedStatus,
+    MaintenanceStatus,
+    Relation,
+    SecretRotate,
+)
 from ops.pebble import Layer
 
-from vault import SECRET_BACKEND_SHARED_HCL, Vault
+from vault import KV_MOUNT_HCL, Vault
 
 logger = logging.getLogger(__name__)
 
@@ -66,8 +68,9 @@ def __init__(self, *args):
         self.framework.observe(self.on.config_changed, self._on_config_changed)
         self.framework.observe(self.on.authorise_charm_action, self._on_authorise_charm_action)
         self.framework.observe(
-            self.vault_kv.on.ready_vault_kv_clients, self._on_ready_vault_kv_clients
+            self.vault_kv.on.new_vault_kv_client_attached, self._on_new_vault_kv_client_attached
         )
+        self.framework.observe(self.on.secret_rotate, self._on_secret_rotate)
         self.service_patcher = KubernetesServicePatch(
             charm=self,
             ports=[ServicePort(name="vault", port=self.VAULT_PORT)],
@@ -225,30 +228,29 @@ def _on_authorise_charm_action(self, event: ActionEvent) -> None:
             self._stored.secret_id = secret_id
             self.unit.status = ActiveStatus()
 
-    def _on_ready_vault_kv_clients(self, event: ReadyVaultKvClientsEvent) -> None:
-        """Configure secret backend for related application.
-
-        Args:
-            event: ReadyVaultKvClientsEvent
-
-        Returns:
-            None
-        """
+    def _on_new_vault_kv_client_attached(self, event: NewVaultKvClientAttachedEvent) -> None:
+        """Configure mount for related application."""
         if not self.unit.is_leader():
             return
 
-        if not event.secret_backend.startswith("charm-"):
-            logger.debug("Skipping secret backend configuration for non-charm application")
-            return
-
         relation = self.model.get_relation(event.relation_name, event.relation_id)
 
-        if relation is None:
-            logger.debug("Relation not found")
+        if relation is None or relation.app is None:
+            logger.debug("Relation or remote application is None, skipping")
             return
 
-        self.vault.configure_secret_backend(event.secret_backend)
-        self.vault_kv.set_kv_mountpoint(relation, event.secret_backend)
+        if "/" in event.mount_suffix:
+            logger.info(
+                "Skipping configuring access for %r, mount suffix %r contains '/'",
+                relation.app.name,
+                event.mount_suffix,
+            )
+            return
+
+        mount = "charm-" + relation.app.name + "-" + event.mount_suffix
+
+        self.vault.configure_kv_mount(mount)
+        self.vault_kv.set_mount(relation, mount)
         vault_url = self._api_address(relation)
         if vault_url is not None:
             self.vault_kv.set_vault_url(relation, vault_url)
@@ -257,33 +259,76 @@ def _on_ready_vault_kv_clients(self, event: ReadyVaultKvClientsEvent) -> None:
             egress_subnet = relation.data[unit].get("egress_subnet")
             if egress_subnet is None:
                 logger.debug(
-                    f"Skipping configuring access for unit {unit.name!r}, egress_subnet missing"
+                    "Skipping configuring access for unit %r, egress_subnet missing", unit.name
                 )
                 continue
             unit_name = unit.name.replace("/", "-")
             policy_name = role_name = "charm-" + unit_name
 
-            self.vault.configure_policy(
-                policy_name, SECRET_BACKEND_SHARED_HCL.format(backend=event.secret_backend)
-            )
+            self.vault.configure_policy(policy_name, KV_MOUNT_HCL.format(mount=mount))
             approle_id = self.vault.configure_approle(role_name, [egress_subnet], [policy_name])
 
-            role_secret_id = self.vault_kv.get_role_secret_id(relation, unit_name)
+            unit_credentials = self.vault_kv.get_unit_credentials(relation, unit_name)
 
-            if role_secret_id is not None:
-                role_secret_id_data = self.vault.read_role_secret_id(role_name, role_secret_id)
+            if unit_credentials is not None:
+                secret = self.model.get_secret(id=unit_credentials)
+                credentials = secret.get_content()
+                role_secret_id_data = self.vault.read_role_secret_id(
+                    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"]:
                     continue
-
+                credentials["role-secret-id"] = self.vault.generate_role_secret_id(
+                    role_name, [egress_subnet]
+                )
+                secret.set_content(credentials)
+            else:
+                role_secret_id = self.vault.generate_role_secret_id(role_name, [egress_subnet])
+                secret = self.app.add_secret(
+                    {"role-id": approle_id, "role-secret-id": role_secret_id},
+                    label=mount + "/" + unit_name,
+                    rotate=SecretRotate.MONTHLY,
+                )
+                secret.grant(relation, unit=unit)
+                self.vault_kv.set_unit_credentials(relation, unit_name, secret)
+
+    def _on_secret_rotate(self, event: SecretRotateEvent):
+        """Handle secret rotate event."""
+        secret = event.secret
+        if secret.label is not None and secret.label.startswith("charm-"):
+            mount, unit_name = secret.label.split("/", 1)
+            relations: List[Relation] = []
+            for relation in self.model.relations[KV_RELATION]:
+                if relation.data[self.app].get("mount") == mount:
+                    relations.append(relation)
+            if len(relations) != 1:
+                logger.warning(
+                    "Either no relation or too many relations found for mount %r, skipping", mount
+                )
+                return
+            relation = relations[0]
+            unit = None
+            for unit_iter in relation.units:
+                if unit_name == unit_iter.name.replace("/", "-"):
+                    unit = unit_iter
+                    break
+            if unit is None:
+                # Unit related to this secret might be gone, skip
+                # Remove secret ?
+                logger.debug("Unit %r not found", unit_name)
+                return
+            egress_subnet = relation.data[unit].get("egress_subnet")
+            if egress_subnet is None:
+                logger.debug(
+                    "Skipping configuring access for unit %r, egress_subnet missing", unit
+                )
+                return
+            role_name = policy_name = "charm-" + unit_name
+            approle_id = self.vault.configure_approle(role_name, [egress_subnet], [policy_name])
             role_secret_id = self.vault.generate_role_secret_id(role_name, [egress_subnet])
-
-            self.vault_kv.set_unit_credentials(relation, unit_name, approle_id, role_secret_id)
-
-    def _on_goneaway_vault_kv_clients(self, event: GoneAwayVaultKvClientsEvent):
-        # clean up
-        pass
+            secret.set_content({"role-id": approle_id, "role-secret-id": role_secret_id})
 
 
-if __name__ == "__main__":  # pragma: no cover
+if __name__ == "__main__":  # pragma: no covergg
     main(VaultCharm)
diff --git a/src/vault.py b/src/vault.py
index 30ab5abf..0d95a0ed 100644
--- a/src/vault.py
+++ b/src/vault.py
@@ -20,11 +20,11 @@
 CHARM_PKI_MOUNT_POINT = "charm-pki-local"
 CHARM_PKI_ROLE = "local"
 
-SECRET_BACKEND_SHARED_HCL = """
-path "{backend}/*" {{
+KV_MOUNT_HCL = """
+path "{mount}/*" {{
   capabilities = ["create", "read", "update", "delete", "list"]
 }}
-path "sys/internal/ui/mounts/{backend}" {{
+path "sys/internal/ui/mounts/{mount}" {{
   capabilities = ["read"]
 }}
 """
@@ -263,14 +263,13 @@ def _issue_certificate(self, **config) -> dict:
         logger.info(f"Issued certificate with role {CHARM_PKI_ROLE} for config: {config}")
         return response["data"]
 
-    def configure_secret_backend(self, name: str):
-        """Ensure a KV backend is enabled."""
+    def configure_kv_mount(self, name: str):
+        """Ensure a KV mount is enabled."""
         if "{}/".format(name) not in self._client.sys.list_mounted_secrets_engines():
             self._client.sys.enable_secrets_engine(
-                backend_type="kv",
+                backend_type="kv-v2",
                 description="Charm created KV backend",
                 path=name,
-                options={"version": 1},
             )
 
     def configure_policy(self, name: str, hcl: str):
diff --git a/tests/unit/test_charm.py b/tests/unit/test_charm.py
index 6c879d55..96a111e9 100644
--- a/tests/unit/test_charm.py
+++ b/tests/unit/test_charm.py
@@ -177,7 +177,7 @@ def test_given_unit_is_leader_when_on_authorise_charm_action_then_status_is_acti
 
         self.assertEqual(self.harness.charm.unit.status, ActiveStatus())
 
-    def setup_secrets_relation(self, nb_units: int = 1) -> tuple:
+    def setup_vault_kv_relation(self, nb_units: int = 1) -> tuple:
         app_name = "app-a"
         unit_name = app_name + "/0"
         relation_name = "secrets"
@@ -205,65 +205,202 @@ def setup_secrets_relation(self, nb_units: int = 1) -> tuple:
     @patch("vault.Vault.generate_role_secret_id")
     @patch("vault.Vault.configure_approle")
     @patch("vault.Vault.configure_policy")
-    @patch("vault.Vault.configure_secret_backend")
-    def test_ready_vault_kv_clients(
+    @patch("vault.Vault.configure_kv_mount")
+    def test_given_unit_is_leader_when_secret_kv_is_complete_then_provider_side_is_filled(
         self,
-        configure_secret_backend,
-        configure_policy,
+        _,
+        __,
+        configure_approle,
+        generate_role_secret_id,
+    ):
+        (
+            app_name,
+            host_ip,
+            _,
+            rel_id,
+            units,
+        ) = self.setup_vault_kv_relation(nb_units=3)
+
+        configure_approle.return_value = "12345678"
+        generate_role_secret_id.return_value = "11111111"
+
+        mount_suffix = "dummy"
+        self.harness.update_relation_data(rel_id, app_name, {"mount_suffix": mount_suffix})
+        mount = "charm-" + app_name + "-" + mount_suffix
+
+        relation_data = self.harness.get_relation_data(rel_id, self.harness.charm.app.name)
+        assert relation_data["vault_url"] == f"http://{host_ip}:{self.harness.charm.VAULT_PORT}"
+        assert relation_data["mount"] == mount
+        for secret_id in json.loads(relation_data["credentials"]).values():
+            secret = self.harness.model.get_secret(id=secret_id)
+            secret_content = secret.get_content()
+            assert configure_approle.return_value == secret_content["role-id"]
+            assert generate_role_secret_id.return_value == secret_content["role-secret-id"]
+
+    @patch("vault.Vault.generate_role_secret_id")
+    @patch("vault.Vault.configure_approle")
+    @patch("vault.Vault.configure_policy")
+    @patch("vault.Vault.configure_kv_mount")
+    def test_given_unit_is_not_leader_when_secret_kv_is_complete_then_no_data_is_updated(
+        self,
+        _,
+        __,
         configure_approle,
         generate_role_secret_id,
     ):
-        from charms.vault_k8s.v0.vault_kv import ReadyVaultKvClientsEvent
+        (
+            app_name,
+            _,
+            _,
+            rel_id,
+            _,
+        ) = self.setup_vault_kv_relation(nb_units=3)
+        self.harness.set_leader(False)
 
-        from vault import SECRET_BACKEND_SHARED_HCL
+        configure_approle.return_value = "12345678"
+        generate_role_secret_id.return_value = "11111111"
+
+        mount_suffix = "dummy"
+        self.harness.update_relation_data(rel_id, app_name, {"mount_suffix": mount_suffix})
+
+        relation_data = self.harness.get_relation_data(rel_id, self.harness.charm.app.name)
+        assert "vault_url" not in relation_data
+        assert "mount" not in relation_data
+        assert "credentials" not in relation_data
 
+    @patch("vault.Vault.read_role_secret_id")
+    @patch("vault.Vault.generate_role_secret_id")
+    @patch("vault.Vault.configure_approle")
+    @patch("vault.Vault.configure_policy")
+    @patch("vault.Vault.configure_kv_mount")
+    def test_given_unit_is_leader_when_related_unit_egress_is_updated_then_secret_content_is_updated(  # noqa: E501
+        self,
+        _,
+        __,
+        configure_approle,
+        generate_role_secret_id,
+        read_role_secret_id,
+    ):
         (
             app_name,
-            host_ip,
-            relation_name,
+            _,
+            _,
             rel_id,
             units,
-        ) = self.setup_secrets_relation(nb_units=3)
+        ) = self.setup_vault_kv_relation(nb_units=3)
 
         configure_approle.return_value = "12345678"
-        generate_role_secret_id.return_value = "12345690"
-
-        event = Mock(ReadyVaultKvClientsEvent)
-        event.secret_backend = "charm-" + app_name
-        event.relation_name = relation_name
-        event.relation_id = rel_id
-        self.harness.charm._on_ready_vault_kv_clients(event)
-
-        configure_secret_backend_calls = [call(event.secret_backend)]
-        configure_policy_calls = []
-        configure_approle_calls = []
-        generate_role_secret_id_calls = []
-
-        for unit, egress_subnet in units.items():
-            unit_name = unit.replace("/", "-")
-            policy_name = role_name = "charm-" + unit_name
-            configure_policy_calls.append(
-                call(policy_name, SECRET_BACKEND_SHARED_HCL.format(backend=event.secret_backend))
-            )
-            configure_approle_calls.append(call(role_name, [egress_subnet], [policy_name]))
-            generate_role_secret_id_calls.append(call(role_name, [egress_subnet]))
-
-        configure_secret_backend.assert_has_calls(configure_secret_backend_calls)
-        configure_policy.assert_has_calls(configure_policy_calls, any_order=True)
-        configure_approle.assert_has_calls(configure_approle_calls, any_order=True)
-        generate_role_secret_id.assert_has_calls(generate_role_secret_id_calls, any_order=True)
-
-        assert self.harness.get_relation_data(rel_id, self.harness.charm.app.name) == {
-            "vault_url": f"http://{host_ip}:{self.harness.charm.VAULT_PORT}",
-            "kv_mountpoint": event.secret_backend,
-            "credentials": json.dumps(
-                {
-                    unit_name.replace("/", "-"): {
-                        "role_id": configure_approle.return_value,
-                        "role_secret_id": generate_role_secret_id.return_value,
-                    }
-                    for unit_name in units
-                },
-                sort_keys=True,
-            ),
-        }
+        generate_role_secret_id.return_value = "11111111"
+
+        mount_suffix = "dummy"
+        self.harness.update_relation_data(rel_id, app_name, {"mount_suffix": mount_suffix})
+        # choose an unit to update
+        unit = next(iter(units.keys()))
+        unit_name = unit.replace("/", "-")
+
+        # Mock read to actually return a comparable cidr_list
+        def mock_role_secret_id(role_name, _):
+            unit_slash = "/".join(role_name[6:].rsplit("-", 1))
+            return {"cidr_list": [units[unit_slash]]}
+
+        read_role_secret_id.side_effect = mock_role_secret_id
+        # get current role secret id from unit's secert
+        app_relation_data = self.harness.get_relation_data(rel_id, self.harness.charm.app.name)
+        credentials = json.loads(app_relation_data["credentials"])
+        juju_secret_id = credentials[unit_name]
+        old_secret_id = self.harness.model.get_secret(id=juju_secret_id).get_content()[
+            "role-secret-id"
+        ]
+
+        generate_role_secret_id.return_value = "22222222"
+        # Update unit egress
+        self.harness.update_relation_data(rel_id, unit, {"egress_subnet": "10.20.20.240/32"})
+
+        read_role_secret_id.has_calls([call("charm-" + unit_name, old_secret_id)])
+
+        for cred_unit, juju_secret_id in credentials.items():
+            secret_content = self.harness.model.get_secret(id=juju_secret_id).get_content()
+            # Assert only the updated unit has a new secret id
+            if cred_unit == unit_name:
+                assert secret_content["role-secret-id"] == "22222222"
+            else:
+                assert secret_content["role-secret-id"] == "11111111"
+
+    @patch("vault.Vault.generate_role_secret_id")
+    @patch("vault.Vault.configure_approle")
+    @patch("vault.Vault.configure_policy")
+    @patch("vault.Vault.configure_kv_mount")
+    def test_given_kv_secret_credentials_when_secret_rotated_then_secret_content_is_updated(  # noqa: E501
+        self,
+        _,
+        __,
+        configure_approle,
+        generate_role_secret_id,
+    ):
+        (
+            app_name,
+            _,
+            _,
+            rel_id,
+            units,
+        ) = self.setup_vault_kv_relation(nb_units=3)
+
+        configure_approle.return_value = "12345678"
+        generate_role_secret_id.return_value = "11111111"
+
+        mount_suffix = "dummy"
+        self.harness.update_relation_data(rel_id, app_name, {"mount_suffix": mount_suffix})
+        app_relation_data = self.harness.get_relation_data(rel_id, self.harness.charm.app.name)
+        credentials = json.loads(app_relation_data["credentials"])
+        unit = next(iter(units.keys()))
+        unit_name = unit.replace("/", "-")
+        juju_secret_id = credentials[unit_name]
+
+        generate_role_secret_id.return_value = "22222222"
+        self.harness.trigger_secret_rotation(juju_secret_id)
+
+        for cred_unit, juju_secret_id in credentials.items():
+            secret_content = self.harness.model.get_secret(id=juju_secret_id).get_content()
+            # Assert only the updated unit has a new secret id
+            print(unit_name, cred_unit)
+            if cred_unit == unit_name:
+                assert secret_content["role-secret-id"] == "22222222"
+            else:
+                assert secret_content["role-secret-id"] == "11111111"
+
+    @patch("vault.Vault.generate_role_secret_id")
+    @patch("vault.Vault.configure_approle")
+    @patch("vault.Vault.configure_policy")
+    @patch("vault.Vault.configure_kv_mount")
+    def test_given_any_non_vault_kv__secret_when_secret_rotated_then_vault_kv_secret_content_are_not_updated(  # noqa: E501
+        self,
+        _,
+        __,
+        configure_approle,
+        generate_role_secret_id,
+    ):
+        (
+            app_name,
+            _,
+            _,
+            rel_id,
+            _,
+        ) = self.setup_vault_kv_relation(nb_units=3)
+
+        configure_approle.return_value = "12345678"
+        generate_role_secret_id.return_value = "11111111"
+
+        mount_suffix = "dummy"
+        self.harness.update_relation_data(rel_id, app_name, {"mount_suffix": mount_suffix})
+        app_relation_data = self.harness.get_relation_data(rel_id, self.harness.charm.app.name)
+        credentials = json.loads(app_relation_data["credentials"])
+
+        secret = self.harness.charm.app.add_secret({"dummy": "secret"})
+        if secret.id is None:
+            # secret is just created, therefore it cannot be None
+            assert False
+        self.harness.trigger_secret_rotation(secret.id)
+
+        for juju_secret_id in credentials.values():
+            secret_content = self.harness.model.get_secret(id=juju_secret_id).get_content()
+            assert secret_content["role-secret-id"] == "11111111"
diff --git a/tests/unit/test_vault_kv.py b/tests/unit/test_vault_kv.py
new file mode 100644
index 00000000..0313cbd7
--- /dev/null
+++ b/tests/unit/test_vault_kv.py
@@ -0,0 +1,275 @@
+#!/usr/bin/env python3
+# Copyright 2023 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+import json
+import textwrap
+import unittest
+from unittest.mock import patch
+
+from charms.vault_k8s.v0 import vault_kv
+from ops import testing
+from ops.charm import CharmBase
+
+
+class VaultKvProviderCharm(CharmBase):
+    metadata_yaml = textwrap.dedent(
+        """
+        name: vault-kv-provider
+        containers:
+          vault:
+            resource: vault-image
+        provides:
+          secrets:
+            interface: vault-kv
+        """
+    )
+
+    def __init__(self, *args):
+        super().__init__(*args)
+        self.interface = vault_kv.VaultKvProvides(self, "secrets")
+        self.framework.observe(
+            self.interface.on.new_vault_kv_client_attached, self._on_new_vault_kv_client_attached
+        )
+
+    def _on_new_vault_kv_client_attached(self, event: vault_kv.NewVaultKvClientAttachedEvent):
+        pass
+
+
+class VaultKvRequirerCharm(CharmBase):
+    metadata_yaml = textwrap.dedent(
+        """
+        name: vault-kv-requirer
+        containers:
+          my-app:
+            resource: my-app-image
+        requires:
+          secrets:
+            interface: vault-kv
+        """
+    )
+
+    def __init__(self, *args):
+        super().__init__(*args)
+        self.interface = vault_kv.VaultKvRequires(self, "secrets", "dummy", "10.20.20.1/32")
+        self.framework.observe(self.interface.on.connected, self._on_connected)
+        self.framework.observe(self.interface.on.ready, self._on_ready)
+        self.framework.observe(self.interface.on.gone_away, self._on_gone_away)
+
+    def _on_connected(self, event: vault_kv.VaultKvConnectedEvent):
+        pass
+
+    def _on_ready(self, event: vault_kv.VaultKvReadyEvent):
+        pass
+
+    def _on_gone_away(self, event: vault_kv.VaultKvGoneAwayEvent):
+        pass
+
+
+class TestVaultKvProvides(unittest.TestCase):
+    def setUp(self):
+        self.harness = testing.Harness(
+            VaultKvProviderCharm, meta=VaultKvProviderCharm.metadata_yaml
+        )
+        self.addCleanup(self.harness.cleanup)
+        self.harness.begin()
+
+    def setup_relation(self, leader: bool = True) -> tuple:
+        remote_app = "vault-kv-requires"
+        remote_unit = remote_app + "/0"
+        rel_name = "secrets"
+        self.harness.set_leader(leader)
+        rel_id = self.harness.add_relation(rel_name, remote_app)
+        relation = self.harness.model.get_relation("secrets", rel_id)
+        assert relation
+        self.harness.add_relation_unit(rel_id, remote_unit)
+        return remote_app, remote_unit, relation, rel_id
+
+    @patch("test_vault_kv.VaultKvProviderCharm._on_new_vault_kv_client_attached")
+    def test_given_unit_joined_when_all_data_present_then_new_client_attached_fired(
+        self, _on_new_vault_kv_client_attached
+    ):
+        remote_app, _, _, rel_id = self.setup_relation()
+
+        suffix = "dummy"
+        self.harness.update_relation_data(rel_id, remote_app, {"mount_suffix": suffix})
+        args, _ = _on_new_vault_kv_client_attached.call_args
+        event = args[0]
+
+        assert isinstance(event, vault_kv.NewVaultKvClientAttachedEvent)
+        assert args[0].mount_suffix == suffix
+
+    @patch("test_vault_kv.VaultKvProviderCharm._on_new_vault_kv_client_attached")
+    def test_given_unit_joined_when_missing_data_then_new_client_attached_is_never_fired(
+        self, _on_new_vault_kv_client_attached
+    ):
+        self.setup_relation()
+        _on_new_vault_kv_client_attached.assert_not_called()
+
+    def test_given_unit_is_leader_when_setting_vault_url_then_relation_data_is_updated(
+        self,
+    ):
+        _, _, relation, rel_id = self.setup_relation()
+        vault_url = "https://vault.example.com"
+        self.harness.charm.interface.set_vault_url(relation, vault_url)
+
+        assert (
+            self.harness.get_relation_data(rel_id, self.harness.charm.app.name)["vault_url"]
+            == vault_url
+        )
+
+    def test_given_unit_is_not_leader_when_setting_vault_url_then_relation_data_is_not_updated(
+        self,
+    ):
+        _, _, relation, rel_id = self.setup_relation(leader=False)
+        vault_url = "https://vault.example.com"
+        self.harness.charm.interface.set_vault_url(relation, vault_url)
+
+        assert "vault_url" not in self.harness.get_relation_data(
+            rel_id, self.harness.charm.app.name
+        )
+
+    def test_given_unit_is_leader_when_setting_mount_then_relation_data_is_updated(
+        self,
+    ):
+        _, _, relation, rel_id = self.setup_relation()
+        mount = "charm-vault-kv-requires-dummy"
+        self.harness.charm.interface.set_mount(relation, mount)
+
+        assert (
+            self.harness.get_relation_data(rel_id, self.harness.charm.app.name)["mount"] == mount
+        )
+
+    def test_given_unit_is_not_leader_when_setting_mount_then_relation_data_is_not_updated(
+        self,
+    ):
+        _, _, relation, rel_id = self.setup_relation(leader=False)
+        mount = "charm-vault-kv-requires-dummy"
+        self.harness.charm.interface.set_mount(relation, mount)
+
+        assert "mount" not in self.harness.get_relation_data(rel_id, self.harness.charm.app.name)
+
+    def test_given_unit_is_leader_when_setting_credentials_then_relation_data_is_updated(
+        self,
+    ):
+        _, remote_unit, relation, rel_id = self.setup_relation()
+        unit_name = remote_unit.replace("/", "-")
+        secret = self.harness.charm.app.add_secret({"role-id": "111", "role-secret-id": "222"})
+        self.harness.charm.interface.set_unit_credentials(relation, unit_name, secret)
+
+        assert json.loads(
+            self.harness.get_relation_data(rel_id, self.harness.charm.app.name)["credentials"]
+        ) == {unit_name: secret.id}
+
+    def test_given_unit_is_not_leader_when_setting_credentials_then_relation_data_is_not_updated(
+        self,
+    ):
+        _, remote_unit, relation, rel_id = self.setup_relation(leader=False)
+        unit_name = remote_unit.replace("/", "-")
+        secret = self.harness.charm.app.add_secret({"role-id": "111", "role-secret-id": "222"})
+        self.harness.charm.interface.set_unit_credentials(relation, unit_name, secret)
+
+        assert "credentials" not in self.harness.get_relation_data(
+            rel_id, self.harness.charm.app.name
+        )
+
+    def test_given_secret_is_missing_id_when_setting_credentials_then_relation_data_is_not_updated(
+        self,
+    ):
+        """Secret._id is None when the secret has been looked up by label."""
+        _, remote_unit, relation, rel_id = self.setup_relation()
+        unit_name = remote_unit.replace("/", "-")
+        secret = self.harness.charm.app.add_secret({"role-id": "111", "role-secret-id": "222"})
+        secret._id = None
+        self.harness.charm.interface.set_unit_credentials(relation, unit_name, secret)
+
+        assert "credentials" not in self.harness.get_relation_data(
+            rel_id, self.harness.charm.app.name
+        )
+
+
+class TestVaultKvRequires(unittest.TestCase):
+    def setUp(self):
+        self.harness = testing.Harness(
+            VaultKvRequirerCharm, meta=VaultKvRequirerCharm.metadata_yaml
+        )
+        self.addCleanup(self.harness.cleanup)
+        self.harness.begin()
+
+    def setup_relation(self, leader: bool = True) -> tuple:
+        remote_app = "vault-kv-provides"
+        remote_unit = remote_app + "/0"
+        rel_name = "secrets"
+        self.harness.set_leader(leader)
+        rel_id = self.harness.add_relation(rel_name, remote_app)
+        relation = self.harness.model.get_relation("secrets", rel_id)
+        assert relation
+        self.harness.add_relation_unit(rel_id, remote_unit)
+        return remote_app, remote_unit, relation, rel_id
+
+    @patch("test_vault_kv.VaultKvRequirerCharm._on_connected")
+    def test_give_unit_leader_when_unit_joined_then_connected_event_fired_and_all_relation_data_is_updated(  # noqa: E501
+        self, _on_connected
+    ):
+        rel_id = self.setup_relation()[-1]
+
+        args, _ = _on_connected.call_args
+        event = args[0]
+
+        app_relation_data = self.harness.get_relation_data(rel_id, self.harness.charm.app.name)
+        unit_relation_data = self.harness.get_relation_data(rel_id, self.harness.charm.unit.name)
+        assert isinstance(event, vault_kv.VaultKvConnectedEvent)
+        assert app_relation_data["mount_suffix"] == self.harness.charm.interface.mount_suffix
+        assert unit_relation_data["egress_subnet"] == self.harness.charm.interface.egress_subnet
+
+    @patch("test_vault_kv.VaultKvRequirerCharm._on_connected")
+    def test_given_unit_joined_is_not_leader_when_relation_joined_then_connected_is_fired_and_mount_suffix_is_not_updated(  # noqa: E501
+        self, _on_connected
+    ):
+        rel_id = self.setup_relation(leader=False)[-1]
+
+        args, _ = _on_connected.call_args
+        event = args[0]
+
+        app_relation_data = self.harness.get_relation_data(rel_id, self.harness.charm.app.name)
+        unit_relation_data = self.harness.get_relation_data(rel_id, self.harness.charm.unit.name)
+        assert isinstance(event, vault_kv.VaultKvConnectedEvent)
+        assert "mount_suffix" not in app_relation_data
+        assert unit_relation_data["egress_subnet"] == self.harness.charm.interface.egress_subnet
+
+    @patch("test_vault_kv.VaultKvRequirerCharm._on_gone_away")
+    def test_given_all_units_departed_when_relation_broken_then_gone_away_event_fired(
+        self, _on_gone_away
+    ):
+        rel_id = self.setup_relation()[-1]
+        self.harness.remove_relation(rel_id)
+        args, _ = _on_gone_away.call_args
+        event = args[0]
+
+        assert isinstance(event, vault_kv.VaultKvGoneAwayEvent)
+
+    @patch("test_vault_kv.VaultKvRequirerCharm._on_ready")
+    def test_given_relation_changed_when_all_data_present_then_ready_event_fired(self, _on_ready):
+        remote_app, _, _, rel_id = self.setup_relation()
+
+        self.harness.update_relation_data(
+            rel_id,
+            remote_app,
+            {
+                "vault_url": "https://vault.example.com",
+                "mount": "charm-vault-kv-requires-dummy",
+                "credentials": json.dumps({"vault-kv-requirer-0": "dummy"}),
+            },
+        )
+
+        args, _ = _on_ready.call_args
+        event = args[0]
+
+        assert isinstance(event, vault_kv.VaultKvReadyEvent)
+
+    @patch("test_vault_kv.VaultKvRequirerCharm._on_ready")
+    def test_given_relation_changed_when_data_missing_then_ready_event_never_fired(
+        self, _on_ready
+    ):
+        self.setup_relation()
+        _on_ready.assert_not_called()
diff --git a/tox.ini b/tox.ini
index a9c13278..55207c8a 100644
--- a/tox.ini
+++ b/tox.ini
@@ -9,7 +9,8 @@ envlist = lint, static, unit
 src_path = {toxinidir}/src/
 unit_test_path = {toxinidir}/tests/unit/
 integration_test_path = {toxinidir}/tests/integration/
-all_path = {[vars]src_path} {[vars]unit_test_path} {[vars]integration_test_path}
+vault_libs_path = {toxinidir}/lib/charms/vault_k8s/
+all_path = {[vars]src_path} {[vars]unit_test_path} {[vars]integration_test_path} {[vars]vault_libs_path}
 
 [testenv]
 setenv =
@@ -70,7 +71,7 @@ deps =
     coverage[toml]
     -r{toxinidir}/requirements.txt
 commands =
-    coverage run --source={[vars]src_path} -m pytest {[vars]unit_test_path} -v --tb native -s {posargs}
+    coverage run --source={[vars]src_path},{[vars]vault_libs_path} -m pytest {[vars]unit_test_path} -v --tb native -s {posargs}
     coverage report
 
 [testenv:integration]