|
| 1 | +"""Secrets related helper classes/functions.""" |
| 2 | + |
| 3 | +# Copyright 2023 Canonical Ltd. |
| 4 | +# See LICENSE file for licensing details. |
| 5 | + |
| 6 | +from typing import Dict, Literal, Optional |
| 7 | + |
| 8 | +from ops import Secret, SecretInfo |
| 9 | +from ops.charm import CharmBase |
| 10 | +from ops.model import SecretNotFoundError |
| 11 | + |
| 12 | +# The unique Charmhub library identifier, never change it |
| 13 | +LIBID = "d77fb3d01aba41ed88e837d0beab6be5" |
| 14 | + |
| 15 | +# Increment this major API version when introducing breaking changes |
| 16 | +LIBAPI = 0 |
| 17 | + |
| 18 | +# Increment this PATCH version before using `charmcraft publish-lib` or reset |
| 19 | +# to 0 if you are raising the major API version |
| 20 | +LIBPATCH = 2 |
| 21 | + |
| 22 | + |
| 23 | +APP_SCOPE = "app" |
| 24 | +UNIT_SCOPE = "unit" |
| 25 | +Scopes = Literal["app", "unit"] |
| 26 | + |
| 27 | + |
| 28 | +class DataSecretsError(Exception): |
| 29 | + """A secret that we want to create already exists.""" |
| 30 | + |
| 31 | + |
| 32 | +class SecretAlreadyExistsError(DataSecretsError): |
| 33 | + """A secret that we want to create already exists.""" |
| 34 | + |
| 35 | + |
| 36 | +def generate_secret_label(charm: CharmBase, scope: Scopes) -> str: |
| 37 | + """Generate unique group_mappings for secrets within a relation context. |
| 38 | +
|
| 39 | + Defined as a standalone function, as the choice on secret labels definition belongs to the |
| 40 | + Application Logic. To be kept separate from classes below, which are simply to provide a |
| 41 | + (smart) abstraction layer above Juju Secrets. |
| 42 | + """ |
| 43 | + members = [charm.app.name, scope] |
| 44 | + return f"{'.'.join(members)}" |
| 45 | + |
| 46 | + |
| 47 | +# Secret cache |
| 48 | + |
| 49 | + |
| 50 | +class CachedSecret: |
| 51 | + """Abstraction layer above direct Juju access with caching. |
| 52 | +
|
| 53 | + The data structure is precisely re-using/simulating Juju Secrets behavior, while |
| 54 | + also making sure not to fetch a secret multiple times within the same event scope. |
| 55 | + """ |
| 56 | + |
| 57 | + def __init__(self, charm: CharmBase, label: str, secret_uri: Optional[str] = None): |
| 58 | + self._secret_meta = None |
| 59 | + self._secret_content = {} |
| 60 | + self._secret_uri = secret_uri |
| 61 | + self.label = label |
| 62 | + self.charm = charm |
| 63 | + |
| 64 | + def add_secret(self, content: Dict[str, str], scope: Scopes) -> Secret: |
| 65 | + """Create a new secret.""" |
| 66 | + if self._secret_uri: |
| 67 | + raise SecretAlreadyExistsError( |
| 68 | + "Secret is already defined with uri %s", self._secret_uri |
| 69 | + ) |
| 70 | + |
| 71 | + if scope == APP_SCOPE: |
| 72 | + secret = self.charm.app.add_secret(content, label=self.label) |
| 73 | + else: |
| 74 | + secret = self.charm.unit.add_secret(content, label=self.label) |
| 75 | + self._secret_uri = secret.id |
| 76 | + self._secret_meta = secret |
| 77 | + return self._secret_meta |
| 78 | + |
| 79 | + @property |
| 80 | + def meta(self) -> Optional[Secret]: |
| 81 | + """Getting cached secret meta-information.""" |
| 82 | + if self._secret_meta: |
| 83 | + return self._secret_meta |
| 84 | + |
| 85 | + if not (self._secret_uri or self.label): |
| 86 | + return |
| 87 | + |
| 88 | + try: |
| 89 | + self._secret_meta = self.charm.model.get_secret(label=self.label) |
| 90 | + except SecretNotFoundError: |
| 91 | + if self._secret_uri: |
| 92 | + self._secret_meta = self.charm.model.get_secret( |
| 93 | + id=self._secret_uri, label=self.label |
| 94 | + ) |
| 95 | + return self._secret_meta |
| 96 | + |
| 97 | + def get_content(self) -> Dict[str, str]: |
| 98 | + """Getting cached secret content.""" |
| 99 | + if not self._secret_content: |
| 100 | + if self.meta: |
| 101 | + self._secret_content = self.meta.get_content() |
| 102 | + return self._secret_content |
| 103 | + |
| 104 | + def set_content(self, content: Dict[str, str]) -> None: |
| 105 | + """Setting cached secret content.""" |
| 106 | + if self.meta: |
| 107 | + self.meta.set_content(content) |
| 108 | + self._secret_content = content |
| 109 | + |
| 110 | + def get_info(self) -> Optional[SecretInfo]: |
| 111 | + """Wrapper function for get the corresponding call on the Secret object if any.""" |
| 112 | + if self.meta: |
| 113 | + return self.meta.get_info() |
| 114 | + |
| 115 | + |
| 116 | +class SecretCache: |
| 117 | + """A data structure storing CachedSecret objects.""" |
| 118 | + |
| 119 | + def __init__(self, charm): |
| 120 | + self.charm = charm |
| 121 | + self._secrets: Dict[str, CachedSecret] = {} |
| 122 | + |
| 123 | + def get(self, label: str, uri: Optional[str] = None) -> Optional[CachedSecret]: |
| 124 | + """Getting a secret from Juju Secret store or cache.""" |
| 125 | + if not self._secrets.get(label): |
| 126 | + secret = CachedSecret(self.charm, label, uri) |
| 127 | + |
| 128 | + # Checking if the secret exists, otherwise we don't register it in the cache |
| 129 | + if secret.meta: |
| 130 | + self._secrets[label] = secret |
| 131 | + return self._secrets.get(label) |
| 132 | + |
| 133 | + def add(self, label: str, content: Dict[str, str], scope: Scopes) -> CachedSecret: |
| 134 | + """Adding a secret to Juju Secret.""" |
| 135 | + if self._secrets.get(label): |
| 136 | + raise SecretAlreadyExistsError(f"Secret {label} already exists") |
| 137 | + |
| 138 | + secret = CachedSecret(self.charm, label) |
| 139 | + secret.add_secret(content, scope) |
| 140 | + self._secrets[label] = secret |
| 141 | + return self._secrets[label] |
| 142 | + |
| 143 | + |
| 144 | +# END: Secret cache |
0 commit comments