Add optional callback-style way to resolve secret values from config

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,24 +235,31 @@ 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.
 
     :param fps: file-like objects (supporting ``.read()``)
     :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(
     *fnames: str | PathLike,
     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,27 +287,42 @@ 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.
 
     :param strings: configuration contents
     :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(
     *names: str,
     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:

confidence/models.py

@@ -1,13 +1,18 @@
+import logging
 import re
 import typing
 from collections.abc import Mapping, Sequence
 from enum import Enum
 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,17 +55,21 @@ 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.
 
     :param sources: source mappings to base the union on, ordered from least to
         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,17 +97,24 @@ class Configuration(Mapping):
     # match a reference as ${key.to.be.resolved}
     _reference_pattern = re.compile(r'\${(?P<path>[^${}]+?)}')
 
-    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.
 
         :param sources: source mappings to base this `Configuration` on,
             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()
 

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,
+        )