211 lines
8.6 KiB
Python
211 lines
8.6 KiB
Python
import os
|
|
import warnings
|
|
from collections.abc import Iterator
|
|
from functools import reduce
|
|
from pathlib import Path
|
|
from typing import TYPE_CHECKING, Any, Literal, Optional
|
|
|
|
from ...exceptions import SettingsError
|
|
from ...utils import path_type_label
|
|
from ..base import PydanticBaseSettingsSource
|
|
from ..utils import parse_env_vars
|
|
from .env import EnvSettingsSource
|
|
from .secrets import SecretsSettingsSource
|
|
|
|
if TYPE_CHECKING:
|
|
from ...main import BaseSettings
|
|
from ...sources import PathType
|
|
|
|
|
|
SECRETS_DIR_MAX_SIZE = 16 * 2**20 # 16 MiB seems to be a reasonable default
|
|
|
|
|
|
class NestedSecretsSettingsSource(EnvSettingsSource):
|
|
def __init__(
|
|
self,
|
|
file_secret_settings: PydanticBaseSettingsSource | SecretsSettingsSource,
|
|
secrets_dir: Optional['PathType'] = None,
|
|
secrets_dir_missing: Literal['ok', 'warn', 'error'] | None = None,
|
|
secrets_dir_max_size: int | None = None,
|
|
secrets_case_sensitive: bool | None = None,
|
|
secrets_prefix: str | None = None,
|
|
secrets_nested_delimiter: str | None = None,
|
|
secrets_nested_subdir: bool | None = None,
|
|
# args for compatibility with SecretsSettingsSource, don't use directly
|
|
case_sensitive: bool | None = None,
|
|
env_prefix: str | None = None,
|
|
) -> None:
|
|
# We allow the first argument to be settings_cls like original
|
|
# SecretsSettingsSource. However, it is recommended to pass
|
|
# SecretsSettingsSource instance instead (as it is shown in usage examples),
|
|
# otherwise `_secrets_dir` arg passed to Settings() constructor will be ignored.
|
|
settings_cls: type[BaseSettings] = getattr(
|
|
file_secret_settings,
|
|
'settings_cls',
|
|
file_secret_settings, # type: ignore[arg-type]
|
|
)
|
|
# config options
|
|
conf = settings_cls.model_config
|
|
self.secrets_dir: PathType | None = first_not_none(
|
|
getattr(file_secret_settings, 'secrets_dir', None),
|
|
secrets_dir,
|
|
conf.get('secrets_dir'),
|
|
)
|
|
self.secrets_dir_missing: Literal['ok', 'warn', 'error'] = first_not_none(
|
|
secrets_dir_missing,
|
|
conf.get('secrets_dir_missing'),
|
|
'warn',
|
|
)
|
|
if self.secrets_dir_missing not in ('ok', 'warn', 'error'):
|
|
raise SettingsError(f'invalid secrets_dir_missing value: {self.secrets_dir_missing}')
|
|
self.secrets_dir_max_size: int = first_not_none(
|
|
secrets_dir_max_size,
|
|
conf.get('secrets_dir_max_size'),
|
|
SECRETS_DIR_MAX_SIZE,
|
|
)
|
|
self.case_sensitive: bool = first_not_none(
|
|
secrets_case_sensitive,
|
|
conf.get('secrets_case_sensitive'),
|
|
case_sensitive,
|
|
conf.get('case_sensitive'),
|
|
False,
|
|
)
|
|
self.secrets_prefix: str = first_not_none(
|
|
secrets_prefix,
|
|
conf.get('secrets_prefix'),
|
|
env_prefix,
|
|
conf.get('env_prefix'),
|
|
'',
|
|
)
|
|
|
|
# nested options
|
|
self.secrets_nested_delimiter: str | None = first_not_none(
|
|
secrets_nested_delimiter,
|
|
conf.get('secrets_nested_delimiter'),
|
|
conf.get('env_nested_delimiter'),
|
|
)
|
|
self.secrets_nested_subdir: bool = first_not_none(
|
|
secrets_nested_subdir,
|
|
conf.get('secrets_nested_subdir'),
|
|
False,
|
|
)
|
|
if self.secrets_nested_subdir:
|
|
if secrets_nested_delimiter or conf.get('secrets_nested_delimiter'):
|
|
raise SettingsError('Options secrets_nested_delimiter and secrets_nested_subdir are mutually exclusive')
|
|
else:
|
|
self.secrets_nested_delimiter = os.sep
|
|
|
|
# ensure valid secrets_path
|
|
if self.secrets_dir is None:
|
|
paths = []
|
|
elif isinstance(self.secrets_dir, (Path, str)):
|
|
paths = [self.secrets_dir]
|
|
else:
|
|
paths = list(self.secrets_dir)
|
|
self.secrets_paths: list[Path] = [Path(p).expanduser().resolve() for p in paths]
|
|
for path in self.secrets_paths:
|
|
self.validate_secrets_path(path)
|
|
|
|
# construct parent
|
|
super().__init__(
|
|
settings_cls,
|
|
case_sensitive=self.case_sensitive,
|
|
env_prefix=self.secrets_prefix,
|
|
env_nested_delimiter=self.secrets_nested_delimiter,
|
|
env_ignore_empty=False, # match SecretsSettingsSource behaviour
|
|
env_parse_enums=True, # we can pass everything here, it will still behave as "True"
|
|
env_parse_none_str=None, # match SecretsSettingsSource behaviour
|
|
)
|
|
self.env_parse_none_str = None # update manually because of None
|
|
|
|
# update parent members
|
|
if not len(self.secrets_paths):
|
|
self.env_vars = {}
|
|
else:
|
|
secrets = reduce(
|
|
lambda d1, d2: dict((*d1.items(), *d2.items())),
|
|
(self.load_secrets(p) for p in self.secrets_paths),
|
|
)
|
|
self.env_vars = parse_env_vars(
|
|
secrets,
|
|
self.case_sensitive,
|
|
self.env_ignore_empty,
|
|
self.env_parse_none_str,
|
|
)
|
|
|
|
def validate_secrets_path(self, path: Path) -> None:
|
|
if not path.exists():
|
|
if self.secrets_dir_missing == 'ok':
|
|
pass
|
|
elif self.secrets_dir_missing == 'warn':
|
|
warnings.warn(f'directory "{path}" does not exist', stacklevel=2)
|
|
elif self.secrets_dir_missing == 'error':
|
|
raise SettingsError(f'directory "{path}" does not exist')
|
|
else:
|
|
raise ValueError # unreachable, checked before
|
|
else:
|
|
if not path.is_dir():
|
|
raise SettingsError(f'secrets_dir must reference a directory, not a {path_type_label(path)}')
|
|
secrets_dir_size = sum(f.stat().st_size for f in self._iter_secret_files(path))
|
|
if secrets_dir_size > self.secrets_dir_max_size:
|
|
raise SettingsError(f'secrets_dir size is above {self.secrets_dir_max_size} bytes')
|
|
|
|
@staticmethod
|
|
def _iter_secret_files(path: Path) -> Iterator[Path]:
|
|
"""Yield the secret files contained in ``path``.
|
|
|
|
``path`` is expected to already be resolved. The directory tree is walked
|
|
explicitly so that symbolic links are handled safely:
|
|
|
|
* a file is only yielded if its real location stays within ``path``; entries
|
|
that resolve outside of it (e.g. through a symbolic link) are skipped, so
|
|
they neither contribute to the ``secrets_dir_max_size`` accounting nor get
|
|
loaded;
|
|
* each real directory is visited at most once, so cyclic or repeated
|
|
symlinks cannot make the walk loop and inflate the size accounting or the
|
|
number of loaded secrets.
|
|
|
|
Because the size check and the loader share this iterator, they always see
|
|
the same set of files.
|
|
"""
|
|
seen_dirs: set[Path] = set()
|
|
|
|
def walk(directory: Path) -> Iterator[Path]:
|
|
# Guard against symlink loops / a directory reachable through multiple
|
|
# links being traversed more than once.
|
|
resolved_dir = directory.resolve()
|
|
if resolved_dir in seen_dirs:
|
|
return
|
|
seen_dirs.add(resolved_dir)
|
|
try:
|
|
entries = sorted(directory.iterdir())
|
|
except OSError:
|
|
return
|
|
for entry in entries:
|
|
resolved = entry.resolve()
|
|
if resolved.is_dir():
|
|
# Only descend into directories that stay within secrets_dir.
|
|
# A symlinked directory pointing outside of ``path`` is not
|
|
# followed at all, so we never walk (potentially large) external
|
|
# trees and never read files from outside secrets_dir.
|
|
if resolved == path or path in resolved.parents:
|
|
yield from walk(entry)
|
|
elif resolved.is_file() and path in resolved.parents:
|
|
# Defense in depth: a file whose real location escapes
|
|
# secrets_dir (e.g. a symlink pointing outside of ``path``) is
|
|
# skipped from both the size accounting and the load.
|
|
yield entry
|
|
|
|
yield from walk(path)
|
|
|
|
@classmethod
|
|
def load_secrets(cls, path: Path) -> dict[str, str]:
|
|
return {str(p.relative_to(path)): p.read_text().strip() for p in cls._iter_secret_files(path)}
|
|
|
|
def __repr__(self) -> str:
|
|
return f'NestedSecretsSettingsSource(secrets_dir={self.secrets_dir!r})'
|
|
|
|
|
|
def first_not_none(*objs: Any) -> Any:
|
|
return next(filter(lambda o: o is not None, objs), None)
|