Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1import os
2import warnings
3from collections.abc import Iterator, Mapping
4from pathlib import Path
5from typing import TYPE_CHECKING, Any, Literal, Optional
7from ...exceptions import SettingsError
8from ...utils import path_type_label
9from ..base import PydanticBaseSettingsSource
10from ..utils import InitState, parse_env_vars
11from .env import EnvSettingsSource
12from .secrets import SecretsSettingsSource
14if TYPE_CHECKING:
15 from ...main import BaseSettings
16 from ...sources import PathType
19SECRETS_DIR_MAX_SIZE = 16 * 2**20 # 16 MiB seems to be a reasonable default
22class NestedSecretsSettingsSource(EnvSettingsSource):
23 def __init__(
24 self,
25 file_secret_settings: PydanticBaseSettingsSource | SecretsSettingsSource,
26 secrets_dir: Optional['PathType'] = None,
27 secrets_dir_missing: Literal['ok', 'warn', 'error'] | None = None,
28 secrets_dir_max_size: int | None = None,
29 secrets_case_sensitive: bool | None = None,
30 secrets_prefix: str | None = None,
31 secrets_nested_delimiter: str | None = None,
32 secrets_nested_subdir: bool | None = None,
33 # args for compatibility with SecretsSettingsSource, don't use directly
34 case_sensitive: bool | None = None,
35 env_prefix: str | None = None,
36 _init_state: InitState | None = None,
37 ) -> None:
38 # We allow the first argument to be settings_cls like original
39 # SecretsSettingsSource. However, it is recommended to pass
40 # SecretsSettingsSource instance instead (as it is shown in usage examples),
41 # otherwise `_secrets_dir` arg passed to Settings() constructor will be ignored.
42 settings_cls: type[BaseSettings] = getattr(
43 file_secret_settings,
44 'settings_cls',
45 file_secret_settings, # type: ignore[arg-type]
46 )
47 # config options
48 conf = settings_cls.model_config
49 self.secrets_dir: PathType | None = first_not_none(
50 secrets_dir,
51 getattr(file_secret_settings, 'secrets_dir', None),
52 conf.get('secrets_dir'),
53 )
54 self.secrets_dir_missing: Literal['ok', 'warn', 'error'] = first_not_none(
55 secrets_dir_missing,
56 conf.get('secrets_dir_missing'),
57 'warn',
58 )
59 if self.secrets_dir_missing not in ('ok', 'warn', 'error'):
60 raise SettingsError(f'invalid secrets_dir_missing value: {self.secrets_dir_missing}')
61 self.secrets_dir_max_size: int = first_not_none(
62 secrets_dir_max_size,
63 conf.get('secrets_dir_max_size'),
64 SECRETS_DIR_MAX_SIZE,
65 )
66 self.case_sensitive: bool = first_not_none(
67 secrets_case_sensitive,
68 conf.get('secrets_case_sensitive'),
69 case_sensitive,
70 conf.get('case_sensitive'),
71 False,
72 )
73 self.secrets_prefix: str = first_not_none(
74 secrets_prefix,
75 conf.get('secrets_prefix'),
76 env_prefix,
77 conf.get('env_prefix'),
78 '',
79 )
81 # nested options
82 self.secrets_nested_delimiter: str | None = first_not_none(
83 secrets_nested_delimiter,
84 conf.get('secrets_nested_delimiter'),
85 conf.get('env_nested_delimiter'),
86 )
87 self.secrets_nested_subdir: bool = first_not_none(
88 secrets_nested_subdir,
89 conf.get('secrets_nested_subdir'),
90 False,
91 )
92 if self.secrets_nested_subdir:
93 if secrets_nested_delimiter or conf.get('secrets_nested_delimiter'):
94 raise SettingsError('Options secrets_nested_delimiter and secrets_nested_subdir are mutually exclusive')
95 self.secrets_nested_delimiter = os.sep
97 # ensure valid secrets_path
98 if self.secrets_dir is None:
99 paths = []
100 elif isinstance(self.secrets_dir, (Path, str)):
101 paths = [self.secrets_dir]
102 else:
103 paths = list(self.secrets_dir)
104 self.secrets_paths: list[Path] = [Path(p).expanduser().resolve() for p in paths]
105 for path in self.secrets_paths:
106 self.validate_secrets_path(path)
108 # construct parent, inheriting the init state of the wrapped source if not provided
109 super().__init__(
110 settings_cls,
111 case_sensitive=self.case_sensitive,
112 env_prefix=self.secrets_prefix,
113 env_nested_delimiter=self.secrets_nested_delimiter,
114 env_ignore_empty=False, # match SecretsSettingsSource behaviour
115 env_parse_enums=True, # we can pass everything here, it will still behave as "True"
116 env_parse_none_str=None, # match SecretsSettingsSource behaviour
117 _init_state=_init_state if _init_state is not None else getattr(file_secret_settings, '_init_state', None),
118 )
119 self.env_parse_none_str = None # update manually because of None
121 def _load_env_vars(self) -> Mapping[str, str | None]:
122 # Overridden so that the base implementation never runs: secrets come from files,
123 # not ``os.environ``, so the case-insensitivity of the Windows environment must not
124 # be applied here. NTFS preserves filename case, and the downgrade would silently
125 # ignore ``secrets_case_sensitive=True``. ``DotEnvSettingsSource`` overrides this
126 # method for the same reason.
127 self.env_parse_none_str = None # match SecretsSettingsSource behaviour
129 # Normalize each directory before merging so that the last directory wins. Merging
130 # raw filenames first would keep the earlier key's insertion position, so a later
131 # ``TOKEN`` could not displace an earlier ``token`` once names are case-folded.
132 secrets: dict[str, str | None] = {}
133 for path in self.secrets_paths:
134 secrets.update(
135 parse_env_vars(
136 self.load_secrets(path),
137 self.case_sensitive,
138 self.env_ignore_empty,
139 self.env_parse_none_str,
140 )
141 )
142 return secrets
144 def validate_secrets_path(self, path: Path) -> None:
145 if not path.exists():
146 if self.secrets_dir_missing == 'ok':
147 pass
148 elif self.secrets_dir_missing == 'warn':
149 warnings.warn(f'directory "{path}" does not exist', stacklevel=2)
150 elif self.secrets_dir_missing == 'error':
151 raise SettingsError(f'directory "{path}" does not exist')
152 else: # pragma: no cover
153 raise ValueError # unreachable, checked before
154 else:
155 if not path.is_dir():
156 raise SettingsError(f'secrets_dir must reference a directory, not a {path_type_label(path)}')
157 secrets_dir_size = sum(f.stat().st_size for f in self._iter_secret_files(path))
158 if secrets_dir_size > self.secrets_dir_max_size:
159 raise SettingsError(f'secrets_dir size is above {self.secrets_dir_max_size} bytes')
161 @staticmethod
162 def _iter_secret_files(path: Path) -> Iterator[Path]:
163 """Yield the secret files contained in ``path``.
165 ``path`` is expected to already be resolved. The directory tree is walked
166 explicitly so that symbolic links are handled safely:
168 * a file is only yielded if its real location stays within ``path``; entries
169 that resolve outside of it (e.g. through a symbolic link) are skipped, so
170 they neither contribute to the ``secrets_dir_max_size`` accounting nor get
171 loaded;
172 * each real directory is visited at most once, so cyclic or repeated
173 symlinks cannot make the walk loop and inflate the size accounting or the
174 number of loaded secrets.
176 Because the size check and the loader share this iterator, they always see
177 the same set of files.
178 """
179 seen_dirs: set[Path] = set()
181 def walk(directory: Path) -> Iterator[Path]:
182 # Guard against symlink loops / a directory reachable through multiple
183 # links being traversed more than once.
184 resolved_dir = directory.resolve()
185 if resolved_dir in seen_dirs:
186 return
187 seen_dirs.add(resolved_dir)
188 try:
189 entries = sorted(directory.iterdir())
190 except OSError:
191 return
192 for entry in entries:
193 resolved = entry.resolve()
194 if resolved.is_dir():
195 # Only descend into directories that stay within secrets_dir.
196 # A symlinked directory pointing outside of ``path`` is not
197 # followed at all, so we never walk (potentially large) external
198 # trees and never read files from outside secrets_dir.
199 if resolved == path or path in resolved.parents:
200 yield from walk(entry)
201 elif resolved.is_file() and path in resolved.parents:
202 # Defense in depth: a file whose real location escapes
203 # secrets_dir (e.g. a symlink pointing outside of ``path``) is
204 # skipped from both the size accounting and the load.
205 yield entry
207 yield from walk(path)
209 @classmethod
210 def load_secrets(cls, path: Path) -> dict[str, str]:
211 # Explicit encoding, matching SecretsSettingsSource: the locale default would
212 # corrupt UTF-8 secrets on a non-UTF-8 Windows code page.
213 return {str(p.relative_to(path)): p.read_text(encoding='utf-8').strip() for p in cls._iter_secret_files(path)}
215 def __repr__(self) -> str:
216 return f'NestedSecretsSettingsSource(secrets_dir={self.secrets_dir!r})'
219def first_not_none(*objs: Any) -> Any:
220 return next(filter(lambda o: o is not None, objs), None)