[DPE-2886] Use labels for internal secrets

.github/workflows/ci.yaml

@@ -148,11 +148,11 @@ jobs:
         run: |
           pipx install tox
           pipx install poetry
-      - name: Free disk space
+      - name: Free up disk space
         run: |
           echo "Free disk space before cleanup"
           df -T
-          # free space in the runner
+          # From https://github.com/actions/runner-images/issues/2840#issuecomment-790492173
           sudo rm -rf /usr/share/dotnet
           sudo rm -rf /opt/ghc
           sudo rm -rf /usr/local/share/boost

lib/charms/mysql/v0/mysql.py

@@ -78,6 +78,13 @@
 from typing import Any, Dict, Iterable, List, Optional, Tuple, Union
 
 import ops
+from charms.mysql.v0.mysql_secrets import (
+    APP_SCOPE,
+    UNIT_SCOPE,
+    Scopes,
+    SecretCache,
+    generate_secret_label,
+)
 from ops.charm import ActionEvent, CharmBase, RelationBrokenEvent
 from ops.model import Unit
 from tenacity import (
@@ -100,7 +107,6 @@
     PEER,
     ROOT_PASSWORD_KEY,
     ROOT_USERNAME,
-    SECRET_ID_KEY,
     SERVER_CONFIG_PASSWORD_KEY,
     SERVER_CONFIG_USERNAME,
 )
@@ -116,7 +122,7 @@
 
 # Increment this PATCH version before using `charmcraft publish-lib` or reset
 # to 0 if you are raising the major API version
-LIBPATCH = 52
+LIBPATCH = 53
 
 UNIT_TEARDOWN_LOCKNAME = "unit-teardown"
 UNIT_ADD_LOCKNAME = "unit-add"
@@ -377,7 +383,7 @@
     def __init__(self, *args):
         super().__init__(*args)
 
-        self.app_secrets, self.unit_secrets = None, None
+        self.secrets = SecretCache(self)
 
         self.framework.observe(self.on.get_cluster_status_action, self._get_cluster_status)
         self.framework.observe(self.on.get_password_action, self._on_get_password)
@@ -535,36 +541,30 @@
 
         return len(active_cos_relations) > 0
 
-    def _get_secret_from_juju(self, scope: str, key: str) -> Optional[str]:
-        """Retrieve and return the secret from the juju secret storage."""
-        if scope == "unit":
-            secret_id = self.unit_peer_data.get(SECRET_ID_KEY)
-
-            if not self.unit_secrets and not secret_id:
-                logger.debug("Getting a secret when no secrets added in juju")
-                return None
-
-            if not self.unit_secrets:
-                secret = self.model.get_secret(id=secret_id)
-                content = secret.get_content()
-                self.unit_secrets = content
-                logger.debug(f"Retrieved secret {key} for unit from juju")
+    def _scope_obj(self, scope: Scopes):
+        if scope == APP_SCOPE:
+            return self.app
+        if scope == UNIT_SCOPE:
+            return self.unit
 
-            return self.unit_secrets.get(key)
-
-        secret_id = self.app_peer_data.get(SECRET_ID_KEY)
+    def _peer_data(self, scope: Scopes) -> Dict:
+        """Return corresponding databag for app/unit."""
+        if self.peers is None:
+            return {}
+        return self.peers.data[self._scope_obj(scope)]
 
-        if not self.app_secrets and not secret_id:
-            logger.debug("Getting a secret when no secrets added in juju")
-            return None
+    def _get_secret_from_juju(self, scope: Scopes, key: str) -> Optional[str]:
+        """Retrieve and return the secret from the juju secret storage."""
+        label = generate_secret_label(self, scope)
+        secret = self.secrets.get(label)
 
-        if not self.app_secrets:
-            secret = self.model.get_secret(id=secret_id)
-            content = secret.get_content()
-            self.app_secrets = content
-            logger.debug(f"Retrieved secret {key} for app from juju")
+        if not secret:
+            logger.debug("Getting a secret when secret is not added in juju")
+            return
 
-        return self.app_secrets.get(key)
+        value = secret.get_content().get(key)
+        logger.debug(f"Retrieved secret {key} for unit from juju")
+        return value
 
     def _get_secret_from_databag(self, scope: str, key: str) -> Optional[str]:
         """Retrieve and return the secret from the peer relation databag."""
@@ -595,62 +595,42 @@
             scope, fallback_key
         )
 
-    def _set_secret_in_databag(self, scope: str, key: str, value: Optional[str]) -> None:
+    def _set_secret_in_databag(self, scope: Scopes, key: str, value: Optional[str]) -> None:
         """Set secret in the peer relation databag."""
         if not value:
-            if scope == "unit":
-                del self.unit_peer_data[key]
-            else:
-                del self.app_peer_data[key]
-            return
-
-        if scope == "unit":
-            self.unit_peer_data[key] = value
-            return
+            try:
+                self._peer_data(scope).pop(key)
+                return
+            except KeyError:
+                logger.error(f"Non-existing secret {scope}:{key} was attempted to be removed.")
+                return
 
-        self.app_peer_data[key] = value
+        self._peer_data(scope)[key] = value
 
-    def _set_secret_in_juju(self, scope: str, key: str, value: Optional[str]) -> None:
+    def _set_secret_in_juju(self, scope: Scopes, key: str, value: Optional[str]) -> None:
         """Set the secret in the juju secret storage."""
-        if scope == "unit":
-            secret_id = self.unit_peer_data.get(SECRET_ID_KEY)
-        else:
-            secret_id = self.app_peer_data.get(SECRET_ID_KEY)
-
-        if secret_id:
-            secret = self.model.get_secret(id=secret_id)
+        # Charm could have been upgraded since last run
+        # We make an attempt to remove potential traces from the databag
+        self._peer_data(scope).pop(key, None)
+
+        label = generate_secret_label(self, scope)
+        secret = self.secrets.get(label)
+        if not secret and value:
+            self.secrets.add(label, {key: value}, scope)
+            return
 
-            if scope == "unit":
-                content = self.unit_secrets or secret.get_content()
-            else:
-                content = self.app_secrets or secret.get_content()
+        content = secret.get_content() if secret else None
 
-            if not value:
-                del content[key]
+        if not value:
+            if content and key in content:
+                content.pop(key, None)
             else:
-                content[key] = value
-
-            secret.set_content(content)
-            logger.debug(f"Updated {scope} secret {secret_id} for {key}")
-        elif not value:
-            return
+                logger.error(f"Non-existing secret {scope}:{key} was attempted to be removed.")
+                return
         else:
-            content = {
-                key: value,
-            }
+            content.update({key: value})
 
-            if scope == "unit":
-                secret = self.unit.add_secret(content)
-                self.unit_peer_data[SECRET_ID_KEY] = secret.id
-            else:
-                secret = self.app.add_secret(content)
-                self.app_peer_data[SECRET_ID_KEY] = secret.id
-            logger.debug(f"Added {scope} secret {secret.id} for {key}")
-
-        if scope == "unit":
-            self.unit_secrets = content
-        else:
-            self.app_secrets = content
+        secret.set_content(content)
 
     def set_secret(
         self, scope: str, key: str, value: Optional[str], fallback_key: Optional[str] = None

lib/charms/mysql/v0/mysql_secrets.py

@@ -0,0 +1,139 @@
+"""Secrets related helper classes/functions."""
+# Copyright 2023 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+from typing import Dict, Literal, Optional
+
+from ops import Secret, SecretInfo
+from ops.charm import CharmBase
+from ops.model import SecretNotFoundError
+
+# The unique Charmhub library identifier, never change it
+LIBID = "ea38eb76a89148659453a3b992387b17"
+
+# 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
+
+APP_SCOPE = "app"
+UNIT_SCOPE = "unit"
+Scopes = Literal[APP_SCOPE, UNIT_SCOPE]
+
+
+class MySQLSecretsError(Exception):
+    """MySQL secrets related error."""
+
+
+class SecretAlreadyExistsError(MySQLSecretsError):
+    """A secret that we want to create already exists."""
+
+
+def generate_secret_label(charm: CharmBase, scope: Scopes) -> str:
+    """Generate unique group_mappings for secrets within a relation context.
+
+    Defined as a standalone function, as the choice on secret labels definition belongs to the
+    Application Logic. To be kept separate from classes below, which are simply to provide a
+    (smart) abstraction layer above Juju Secrets.
+    """
+    members = [charm.app.name, scope]
+    return f"{'.'.join(members)}"
+
+
+class CachedSecret:
+    """Abstraction layer above direct Juju access with caching.
+
+    The data structure is precisely re-using/simulating Juju Secrets behavior, while
+    also making sure not to fetch a secret multiple times within the same event scope.
+    """
+
+    def __init__(self, charm: CharmBase, label: str, secret_uri: Optional[str] = None):
+        self._secret_meta = None
+        self._secret_content = {}
+        self._secret_uri = secret_uri
+        self.label = label
+        self.charm = charm
+
+    def add_secret(self, content: Dict[str, str], scope: Scopes) -> Secret:
+        """Create a new secret."""
+        if self._secret_uri:
+            raise SecretAlreadyExistsError(
+                "Secret is already defined with uri %s", self._secret_uri
+            )
+
+        if scope == APP_SCOPE:
+            secret = self.charm.app.add_secret(content, label=self.label)
+        else:
+            secret = self.charm.unit.add_secret(content, label=self.label)
+        self._secret_uri = secret.id
+        self._secret_meta = secret
+        return self._secret_meta
+
+    @property
+    def meta(self) -> Optional[Secret]:
+        """Getting cached secret meta-information."""
+        if self._secret_meta:
+            return self._secret_meta
+
+        if not (self._secret_uri or self.label):
+            return
+
+        try:
+            self._secret_meta = self.charm.model.get_secret(label=self.label)
+        except SecretNotFoundError:
+            if self._secret_uri:
+                self._secret_meta = self.charm.model.get_secret(
+                    id=self._secret_uri, label=self.label
+                )
+        return self._secret_meta
+
+    def get_content(self) -> Dict[str, str]:
+        """Getting cached secret content."""
+        if not self._secret_content:
+            if self.meta:
+                self._secret_content = self.meta.get_content()
+        return self._secret_content
+
+    def set_content(self, content: Dict[str, str]) -> None:
+        """Setting cached secret content."""
+        if self.meta:
+            if content:
+                self.meta.set_content(content)
+                self._secret_content = content
+            else:
+                self.meta.remove_all_revisions()
+
+    def get_info(self) -> Optional[SecretInfo]:
+        """Wrapper function for get the corresponding call on the Secret object if any."""
+        if self.meta:
+            return self.meta.get_info()
+
+
+class SecretCache:
+    """A data structure storing CachedSecret objects."""
+
+    def __init__(self, charm):
+        self.charm = charm
+        self._secrets: Dict[str, CachedSecret] = {}
+
+    def get(self, label: str, uri: Optional[str] = None) -> Optional[CachedSecret]:
+        """Getting a secret from Juju Secret store or cache."""
+        if not self._secrets.get(label):
+            secret = CachedSecret(self.charm, label, uri)
+
+            # Checking if the secret exists, otherwise we don't register it in the cache
+            if secret.meta:
+                self._secrets[label] = secret
+        return self._secrets.get(label)
+
+    def add(self, label: str, content: Dict[str, str], scope: Scopes) -> CachedSecret:
+        """Adding a secret to Juju Secret."""
+        if self._secrets.get(label):
+            raise SecretAlreadyExistsError(f"Secret {label} already exists")
+
+        secret = CachedSecret(self.charm, label)
+        secret.add_secret(content, scope)
+        self._secrets[label] = secret
+        return self._secrets[label]