diff --git a/confidence/io.py b/confidence/io.py index f384786..1652925 100644 --- a/confidence/io.py +++ b/confidence/io.py @@ -11,6 +11,7 @@ from confidence.formats import YAML, Format from confidence.models import Configuration, Missing, NoDefault, NotConfigured +from confidence.secrets import SecretCallback, Secrets, to_secrets LOG = logging.getLogger(__name__) @@ -234,7 +235,12 @@ def loaders(*specifiers: Locality | Loadable) -> typing.Iterable[Loadable]: ) -def load(*fps: typing.TextIO, format: Format = YAML, missing: typing.Any = Missing.SILENT) -> Configuration: +def load( + *fps: typing.TextIO, + format: Format = YAML, + missing: typing.Any = Missing.SILENT, + secrets: Secrets | SecretCallback | None = None, +) -> Configuration: """ Read a `Configuration` instance from file-like objects. @@ -242,9 +248,10 @@ def load(*fps: typing.TextIO, format: Format = YAML, missing: typing.Any = Missi :param format: configuration (file) format to use :param missing: policy to be used when a configured key is missing, either as a `Missing` instance or a default value + :param secrets: an optional `Secrets` implementation or callback function :returns: a `Configuration` instance providing values from *fps* """ - return Configuration(*(format.load(fp) for fp in fps), missing=missing) + return Configuration(*(format.load(fp) for fp in fps), missing=missing, secrets=to_secrets(secrets)) def loadf( @@ -252,6 +259,7 @@ def loadf( format: Format = YAML, default: typing.Any = NoDefault, missing: typing.Any = Missing.SILENT, + secrets: Secrets | SecretCallback | None = None, ) -> Configuration: """ Read a `Configuration` instance from named files. @@ -262,6 +270,7 @@ def loadf( exist (default is to raise a `FileNotFoundError`) :param missing: policy to be used when a configured key is missing, either as a `Missing` instance or a default value + :param secrets: an optional `Secrets` implementation or callback function :returns: a `Configuration` instance providing values from *fnames* """ @@ -278,10 +287,19 @@ def readf(fpath: Path) -> typing.Mapping[str, typing.Any]: return default # expand the user directories here, format is not in charge of the file paths - return Configuration(*(readf(Path(fname).expanduser()) for fname in fnames), missing=missing) + return Configuration( + *(readf(Path(fname).expanduser()) for fname in fnames), + missing=missing, + secrets=to_secrets(secrets), + ) -def loads(*strings: str, format: Format = YAML, missing: typing.Any = Missing.SILENT) -> Configuration: +def loads( + *strings: str, + format: Format = YAML, + missing: typing.Any = Missing.SILENT, + secrets: Secrets | SecretCallback | None = None, +) -> Configuration: """ Read a `Configuration` instance from strings. @@ -289,9 +307,14 @@ def loads(*strings: str, format: Format = YAML, missing: typing.Any = Missing.SI :param format: configuration (file) format to use :param missing: policy to be used when a configured key is missing, either as a `Missing` instance or a default value + :param secrets: an optional `Secrets` implementation or callback function :returns: a `Configuration` instance providing values from *strings* """ - return Configuration(*(format.loads(string) for string in strings), missing=missing) + return Configuration( + *(format.loads(string) for string in strings), + missing=missing, + secrets=to_secrets(secrets), + ) def load_name( @@ -299,6 +322,7 @@ def load_name( load_order: typing.Iterable[Loadable] = DEFAULT_LOAD_ORDER, format: Format = YAML, missing: typing.Any = Missing.SILENT, + secrets: Secrets | SecretCallback | None = None, extension: None = None, # NB: parameter is deprecated, see below ) -> Configuration: """ @@ -317,6 +341,7 @@ def load_name( :param format: configuration (file) format to use :param missing: policy to be used when a configured key is missing, either as a `Missing` instance or a default value + :param secrets: an optional `Secrets` implementation or callback function :returns: a `Configuration` instances providing values loaded from *names* in *load_order* ordering """ @@ -351,7 +376,7 @@ def generate_sources() -> typing.Iterable[typing.Mapping[str, typing.Any]]: yield loadf(candidate, format=format, default=NotConfigured) - return Configuration(*generate_sources(), missing=missing) + return Configuration(*generate_sources(), missing=missing, secrets=to_secrets(secrets)) def _check_format_encoding(format: Format, encoding: str | None) -> Format: diff --git a/confidence/models.py b/confidence/models.py index acf8c87..8cef290 100644 --- a/confidence/models.py +++ b/confidence/models.py @@ -1,3 +1,4 @@ +import logging import re import typing from collections.abc import Mapping, Sequence @@ -5,9 +6,13 @@ from itertools import chain from confidence.exceptions import ConfiguredReferenceError, NotConfiguredError +from confidence.secrets import MappingSecrets, Secrets, SequenceSecrets, StrSecrets from confidence.utils import Conflict, merge_into, split_keys +LOG = logging.getLogger(__name__) + + class Missing(Enum): SILENT = 'silent' #: return `NotConfigured` for unconfigured keys, avoiding errors ERROR = 'error' #: raise an `AttributeError` for unconfigured keys @@ -50,7 +55,9 @@ def unwrap(source: typing.Any) -> typing.Any: return source -def merge(*sources: typing.Mapping[str, typing.Any], missing: typing.Any = None) -> 'Configuration': +def merge( + *sources: typing.Mapping[str, typing.Any], missing: typing.Any = None, secrets: typing.Any = None +) -> 'Configuration': """ Merges *sources* into a union, keeping right-side precedence. @@ -58,9 +65,11 @@ def merge(*sources: typing.Mapping[str, typing.Any], missing: typing.Any = None) most significance :param missing: policy for the resulting `Configuration` (defaults to `Missing.SILENT`) + :param secrets: an optional provider of secret values (defaults to `None`) :return: a `Configuration` instance that encompasses all of the keys and values in *sources* - :raises ValueError: when the missing policies of *source* cannot be aligned + :raises ValueError: when the missing policies or secrets implementation of + *sources* cannot be aligned """ if missing is None: # no explicit missing setting, collect settings from arguments, should be either nothing if sources are not @@ -70,7 +79,13 @@ def merge(*sources: typing.Mapping[str, typing.Any], missing: typing.Any = None) # use the one remaining missing setting, or default to Missing.SILENT missing = missing.pop() if missing else Missing.SILENT - return Configuration(*sources, missing=missing) + if secrets is None: + # no explicit secrets handler, use the same approach as with missing + if len(secrets := {source._secrets for source in sources if isinstance(source, Configuration)}) > 1: + raise ValueError(f'no union for incompatible instances: {secrets}') + secrets = secrets.pop() if secrets else None + + return Configuration(*sources, missing=missing, secrets=secrets) class Configuration(Mapping): @@ -82,7 +97,12 @@ class Configuration(Mapping): # match a reference as ${key.to.be.resolved} _reference_pattern = re.compile(r'\${(?P[^${}]+?)}') - def __init__(self, *sources: typing.Mapping[str, typing.Any], missing: typing.Any = Missing.SILENT): + def __init__( + self, + *sources: typing.Mapping[str, typing.Any], + missing: typing.Any = Missing.SILENT, + secrets: Secrets | None = None, + ): """ Create a new `Configuration`, based on one or multiple source mappings. @@ -90,9 +110,11 @@ def __init__(self, *sources: typing.Mapping[str, typing.Any], missing: typing.An ordered from least to most significant :param missing: policy to be used when a configured key is missing, either as a `Missing` instance or a default value + :param secrets: an optional `Secrets` implementation """ self._missing = missing self._root = self + self._secrets = secrets if isinstance(self._missing, Missing): self._missing = { @@ -113,7 +135,7 @@ def __init__(self, *sources: typing.Mapping[str, typing.Any], missing: typing.An def _wrap(self, value: typing.Mapping[str, typing.Any]) -> 'Configuration': # create an instance of our current type, copying 'configured' properties / policies - namespace = type(self)(missing=self._missing) + namespace = type(self)(missing=self._missing, secrets=self._secrets) namespace._source = value # type: ignore # mutability isn't needed after init # carry the root object from namespace to namespace, references are always resolved from root namespace._root = self._root @@ -180,7 +202,7 @@ def get( *default* was ``NoDefault`` :raises ConfiguredReferenceError: when a reference could not be resolved """ - value = self._source + value: Mapping[str, typing.Any] = self._source steps_taken = [] try: # walk through the values dictionary @@ -191,18 +213,36 @@ def get( if as_type: # explicit type conversion requested return as_type(value) - elif isinstance(value, Mapping): - # wrap value in a Configuration - return self._wrap(value) - elif isinstance(value, Sequence) and not isinstance(value, str | bytes): - # wrap value in a sequence that retains Configuration functionality - return ConfigurationSequence(value, self._root) - elif resolve_references and isinstance(value, str): - # only resolve references in str-type values (the only way they can be expressed) - return self._resolve(value) - else: - # a 'simple' value, nothing to do - return value + + match value, self._secrets: + case {}, MappingSecrets() if self._secrets.matches_mapping(value): + # value is a secret, let the local secret handler resolve this + LOG.debug(f'resolving value for key "{path}" as a mapping type secret') + return self._secrets.resolve(value) + case {}, _: + # wrap value in a Configuration + return self._wrap(value) + + # TODO: could we ever encounter other sequence types? + case ((list() | tuple()), SequenceSecrets()) if self._secrets.matches_sequence(value): + # value is a secret, let the local secret handler resolve this + LOG.debug(f'resolving value for key "{path}" as a sequence type secret') + return self._secrets.resolve(value) + case ((list() | tuple()), _): + # wrap value in a sequence that retains Configuration functionality + return ConfigurationSequence(value, self._root) + + case str(), StrSecrets() if self._secrets.matches_str(value): + # value is a secret, let the local secret handler resolve this + LOG.debug(f'resolving value for key "{path}" as a str type secret') + return self._secrets.resolve(value) + case str(), _ if resolve_references: + # only resolve references in str-type values (the only way they can be expressed) + return self._resolve(value) + + case _: + # no action needed, just return value + return value except ConfiguredReferenceError: # also a KeyError, but this one should bubble to caller raise @@ -276,6 +316,8 @@ def __repr__(self) -> str: keys = ', '.join(_repr_value(key) for key in self.keys()) return f'{self.__class__.__module__}.{self.__class__.__name__}(keys=[{keys}])' + # FIXME: pickling roundtrips will likely break when a Secrets is supplied... + def __getstate__(self) -> dict[str, typing.Any]: state = self.__dict__.copy() diff --git a/confidence/secrets.py b/confidence/secrets.py new file mode 100644 index 0000000..850ad12 --- /dev/null +++ b/confidence/secrets.py @@ -0,0 +1,92 @@ +import logging +import typing +from functools import partial + + +LOG = logging.getLogger(__name__) + + +@typing.runtime_checkable +class Secrets(typing.Protocol): + def resolve(self, value: typing.Any) -> typing.Any: ... + + +@typing.runtime_checkable +class MappingSecrets(Secrets, typing.Protocol): + # e.g. {"$secret": {"service": "github.com", "username": "akaIDIOT"}} + def matches_mapping(self, value: typing.Mapping[str, typing.Any]) -> bool: ... + + +@typing.runtime_checkable +class SequenceSecrets(Secrets, typing.Protocol): + # e.g. ["$secret", "github.com", "akaIDIOT"] # TODO: references inside of that sequence would likely not work? + def matches_sequence(self, value: typing.Sequence[typing.Any]) -> bool: ... + + +@typing.runtime_checkable +class StrSecrets(Secrets, typing.Protocol): + # e.g. "secret!akaIDIOT@github.com" # TODO: references inside of that str would likely not work? + def matches_str(self, value: str) -> bool: ... + + +@typing.runtime_checkable +class SecretCallback(typing.Protocol): + def __call__(self, *args: typing.Any) -> typing.Any: ... + + +# FIXME: load_name(..., secrets=keyring.get_password) does not yet specify a matching value (type) +# FIXME: the rest is very geared towards MappingSecrets + + +def to_secrets(secrets: Secrets | SecretCallback | None) -> Secrets | None: + if not secrets: + return None + elif isinstance(secrets, SecretCallback): + return SingleKeyCallback(secrets) + else: + return secrets + + +def is_single_key_secret(value: typing.Mapping[str, typing.Any], *, key: str) -> bool: + return len(value) == 1 and key in value + + +def resolve_n_key_secret_callback( + value: typing.Mapping[str, typing.Any], *, callback: SecretCallback, single_key: str, args: typing.Iterable[str] +) -> typing.Any: + try: + mapping = value[single_key] + LOG.debug(f'getting values for ({", ".join(args)}) to use as secret retrieval parameters') + parameters = tuple(mapping[arg] for arg in args) + # logging parameters' *values* might still leak things a user would rather not log + LOG.info(f'passing {len(parameters)} to secret callback {callback}') + return callback(*parameters) + except KeyError as e: + if missing_key := e.args[0] if e.args[0] == single_key else f'{single_key}.{e.args[0]}': + LOG.warning(f'resolving secret failed, missing key {missing_key}') + else: + LOG.warning('resolving secret failed') + # logging out of the way, there's not actually anything we can do to fix the error here + # if the caller was Configuration.get(), it will handle the KeyError according to it's policies + raise + + +class SingleKeyCallback: + def __init__( + self, + callback: SecretCallback, + single_key: str = '$secret', + args: typing.Iterable[str] = ('service', 'username'), + ): + # use is_single_key_secret and resolve_n_key_secret_callback to turn the callback we've been handed here into + # something that will implement the Secrets protocol + self.matches_mapping = partial( + is_single_key_secret, + key=single_key, + ) + self.resolve = partial( + resolve_n_key_secret_callback, + callback=callback, + single_key=single_key, + args=args, + )