config_ninja.contrib.secretsmanager

  1"""Integrate with the AWS SecretsManager service.
  2
  3## Example
  4
  5The following `config-ninja`_ settings file configures the `SecretsManagerBackend` to install
  6`~/.docker/config.json` from the latest version of the secret:
  7
  8```yaml
  9.. include:: ../../../examples/secretsmanager-backend.yaml
 10```
 11
 12.. _config-ninja: https://config-ninja.readthedocs.io/home.html
 13"""
 14
 15from __future__ import annotations
 16
 17import asyncio
 18import logging
 19import typing
 20
 21import boto3
 22
 23from config_ninja.backend import Backend
 24from config_ninja.contrib.appconfig import MINIMUM_POLL_INTERVAL_SECONDS
 25
 26if typing.TYPE_CHECKING:  # pragma: no cover
 27    from mypy_boto3_secretsmanager import SecretsManagerClient
 28
 29
 30__all__ = ['SecretsManagerBackend']
 31
 32logger = logging.getLogger(__name__)
 33
 34
 35class SecretsManagerBackend(Backend):
 36    """Retrieve config data from the AWS SecretsManager service.
 37
 38    ## Usage
 39
 40    >>> backend = SecretsManagerBackend(secretsmanager_client, 'secret-id')
 41    >>> print(backend.get())
 42    {"username": "admin", "password": 1234}
 43    """
 44
 45    client: SecretsManagerClient
 46    """The `boto3` client used to communicate with the AWS Secrets Manager service."""
 47
 48    secret_id: str
 49    """The ID of the secret to retrieve"""
 50
 51    version_id: str | None = None
 52
 53    def __init__(self, client: SecretsManagerClient, secret_id: str) -> None:
 54        """Initialize the backend."""
 55        self.client = client
 56        self.secret_id = secret_id
 57        logger.debug('Initialize: %s', repr(self))
 58
 59    def __str__(self) -> str:
 60        """Return the secret ID.
 61
 62        >>> print(str(SecretsManagerBackend(secretsmanager_client, 'secret-id')))
 63        secret-id
 64        """
 65        return self.secret_id if not self.version_id else f'{self.secret_id} (version: {self.version_id})'
 66
 67    @classmethod
 68    def new(cls, secret_id: str, session: boto3.Session | None = None) -> SecretsManagerBackend:  # pylint: disable=arguments-differ
 69        """Instantiate a new `boto3` client and `SecretsManagerBackend` object.
 70
 71        >>> backend = SecretsManagerBackend.new('secret-id')
 72        >>> print(backend)
 73        secret-id
 74        """
 75        logger.info('Create new instance: %s(secret_id="%s")', cls.__name__, secret_id)
 76        session = session or boto3.Session()
 77        client: SecretsManagerClient = session.client('secretsmanager')  # pyright: ignore[reportUnknownMemberType]
 78        return cls(client, secret_id)
 79
 80    def get(self) -> str:
 81        """Retrieve the secret data."""
 82        response = self.client.get_secret_value(SecretId=self.secret_id)
 83        self.version_id = response.get('VersionId')
 84        return response['SecretString']
 85
 86    def _retrieve_current_version(self) -> str:
 87        """Retrieve the version ID of the current value of the secret.
 88
 89        A value error is raised if no current version was found:
 90
 91        >>> backend = SecretsManagerBackend(secretsmanager_client_no_current, 'secret-id')
 92        >>> with pytest.raises(ValueError):
 93        ...     backend._retrieve_current_version()
 94        """
 95        response = self.client.list_secret_version_ids(SecretId=self.secret_id)
 96        for version in response['Versions']:
 97            if 'AWSCURRENT' in version.get('VersionStages', []) and (version_id := version.get('VersionId')):
 98                return version_id
 99
100        raise ValueError(f"No current version found for secret '{self}'")
101
102    async def poll(self, interval: int = MINIMUM_POLL_INTERVAL_SECONDS) -> typing.AsyncIterator[str]:
103        """Poll for changes to the secret."""
104        while True:
105            logger.debug('Poll for configuration changes')
106            try:
107                version_id = self._retrieve_current_version()
108            except ValueError as exc:
109                logger.warning('%s', exc)
110            else:
111                if version_id and version_id != self.version_id:
112                    yield self.get()
113
114            await asyncio.sleep(interval)
115
116    def _async_doctests(self) -> None:
117        """Define `async` `doctest` tests in this method to improve documentation.
118
119        Verify that an empty response to the `boto3` client is handled and the polling continues:
120        >>> backend = SecretsManagerBackend(secretsmanager_client, 'secret-id')
121        >>> content = asyncio.run(anext(backend.poll(interval=0.01)))
122        >>> print(content)
123        {"username": "admin", "password": 1234}
124
125        >>> backend = SecretsManagerBackend(secretsmanager_client_no_current_initially, 'secret-id')
126        >>> _ = backend.get(); backend.version_id   # fetch the initial value from the mock fixture
127        'v6'
128
129        >>> content = asyncio.run(anext(backend.poll(interval=0.01)))
130        >>> print(content)
131        {"username": "admin", "password": 1234}
132        """