feat: add secrets backend relation

lib/charms/vault_k8s/v0/vault_kv.py

@@ -0,0 +1,369 @@
+#!/usr/bin/env python3
+# Copyright 2023 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+"""Contains the VaultKVProvides class."""
+
+import json
+import logging
+from typing import Any, Dict, Optional
+
+import ops
+
+logger = logging.getLogger(__name__)
+
+
+# The unique Charmhub library identifier, never change it
+LIBID = "to_fill"
+
+# Increment this major API version when introducing breaking changes
+LIBAPI = 0
+
+# Increment this PATCH version before using `charmcraft publish-lib` or reset
+# to 0 if you are raising the major API version
+LIBPATCH = 1
+
+
+class HasVaultKvClientsEvent(ops.EventBase):
+    """Has VaultKvClients Event."""
+
+    pass
+
+
+class ReadyVaultKvClientsEvent(ops.EventBase):
+    """Ready VaultKvClients Event."""
+
+    def __init__(
+        self,
+        handle: ops.Handle,
+        relation_id: int,
+        relation_name: str,
+        secret_backend: str,
+    ):
+        super().__init__(handle)
+        self.relation_id = relation_id
+        self.relation_name = relation_name
+        self.secret_backend = secret_backend
+
+    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,
+        }
+
+    def restore(self, snapshot: Dict[str, Any]):
+        """Restore the value state from a given snapshot."""
+        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
+
+
+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)
+
+
+class VaultKvProvides(ops.Object):
+    """Class to be instanciated by the providing side of the relation."""
+
+    on = VaultKvProviderEvents()
+    _stored = ops.StoredState()
+
+    def __init__(
+        self,
+        charm: ops.CharmBase,
+        relation_name: str,
+    ) -> None:
+        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."""
+        if event.app is None:
+            logger.debug("No remote application yet")
+            return
+
+        secret_backend = event.relation.data[event.app].get("secret_backend")
+
+        if secret_backend is not None:
+            self.on.ready_vault_kv_clients.emit(
+                event.relation.id,
+                event.relation.name,
+                secret_backend,
+            )
+
+    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():
+            return
+
+        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."""
+        if not self.charm.unit.is_leader():
+            return
+
+        relation.data[self.charm.app]["kv_mountpoint"] = kv_mountpoint
+
+    def set_unit_credentials(
+        self, relation: ops.Relation, name: str, role_id: str, role_secret_id: str
+    ):
+        """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}
+
+        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."""
+        credentials = self.get_credentials(relation)
+        unit_credentials = credentials.get(name, {})
+        return unit_credentials.get("role_secret_id")
+
+    def get_credentials(self, relation: ops.Relation) -> dict:
+        """Get the unit credentials from the relation."""
+        return json.loads(relation.data[self.charm.app].get("credentials", "{}"))
+
+
+class VaultKvConnectedEvent(ops.EventBase):
+    """VaultKvConnectedEvent Event."""
+
+    pass
+
+
+class VaultKvReadyEvent(ops.EventBase):
+    """VaultKvReadyEvent Event."""
+
+    def __init__(
+        self,
+        handle: ops.Handle,
+        relation_id: int,
+        relation_name: str,
+        vault_url: str,
+        kv_mountpoint: str,
+        role_id: str,
+        role_secret_id: 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
+
+    def snapshot(self) -> dict:
+        """Return snapshot data that should be persisted."""
+        return {
+            "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,
+        }
+
+    def restore(self, snapshot: Dict[str, Any]):
+        """Restore the value state from a given snapshot."""
+        super().restore(snapshot)
+        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"]
+
+
+class VaultKvGoneAwayEvent(ops.EventBase):
+    """VaultKvGoneAwayEvent Event."""
+
+    pass
+
+
+class VaultKvRequireEvents(ops.ObjectEvents):
+    """List of events that the Vault Kv requirer charm can leverage."""
+
+    connected = ops.EventSource(VaultKvConnectedEvent)
+    ready = ops.EventSource(VaultKvReadyEvent)
+    gone_away = ops.EventSource(VaultKvGoneAwayEvent)
+
+
+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,
+        egress_subnet: str,
+    ) -> None:
+        super().__init__(charm, relation_name)
+        self.charm = charm
+        self.relation_name = relation_name
+        self.secret_backend = secret_backend
+        self.egress_subnet = egress_subnet
+        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_broken,
+            self._on_vault_kv_relation_broken,
+        )
+
+        self._unit_name = self.charm.unit.name.replace("/", "-")
+        self._update_unit_egress_subnet()
+
+    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.
+
+        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]
+            unit_egress_subnet = unit_databag.get("egress_subnet", self.egress_subnet)
+            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.
+
+        Set the secret backend in the application databag if we are the leader.
+        Always update the egress_subnet in the unit databag.
+        """
+        self.on.connected.emit()
+        if self.charm.unit.is_leader():
+            event.relation.data[self.charm.app]["secret_backend"] = self.secret_backend
+        self._update_unit_egress_subnet(force=True)
+
+    def _on_vault_kv_relation_changed(self, event: ops.RelationChangedEvent):
+        """Handle relation changed."""
+        if event.app is None:
+            logger.debug("No remote application yet")
+            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)):
+            self.on.ready.emit(
+                event.relation.id,
+                event.relation.name,
+                vault_url,
+                kv_mountpoint,
+                role_id,
+                role_secret_id,
+            )
+
+    def _on_vault_kv_relation_broken(self, event: ops.RelationBrokenEvent):
+        """Handle relation broken."""
+        self.on.gone_away.emit()
+
+    def vault_url(self, relation: ops.Relation) -> Optional[str]:
+        """Return the vault_url from the relation."""
+        if relation.app is None:
+            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."""
+        if relation.app is None:
+            return None
+        return relation.data[relation.app].get("kv_mountpoint")
+
+    def unit_credentials(self, relation: ops.Relation) -> Optional[dict]:
+        """Return the unit credentials from the relation."""
+        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")

metadata.yaml

@@ -35,6 +35,8 @@ storage:
 provides:
   certificates:
     interface: tls-certificates
+  secrets:
+    interface: vault-kv
 
 peers:
   peers: