Coverage for python/univention/config_registry/backend.py: 94%
291 statements
« prev ^ index » next coverage.py v7.6.0, created at 2026-09-11 14:11 +0000
1#
2# main configuration registry classes
3#
4# SPDX-FileCopyrightText: 2004-2026 Univention GmbH
5# SPDX-License-Identifier: AGPL-3.0-only
7"""Univention Configuration Registry backend for data storage."""
10import errno
11import fcntl
12import os
13import re
14import sys
15import time
16from collections.abc import ItemsView, Iterator, Mapping, MutableMapping
17from enum import IntEnum
18from stat import S_ISREG
19from types import TracebackType
20from typing import IO, Literal, NoReturn, Self, TypeVar, overload
22from univention.config_registry.handler import run_filter
25_VT = TypeVar('_VT')
27__all__ = ['SCOPE', 'ConfigRegistry', 'StrictModeException', 'exception_occured']
28MYPY = False
29INVALID_VALUE_CHARS = '\r\n'
32class StrictModeException(Exception):
33 """Attempt to store non-UTF-8 characters in strict UTF-8 mode."""
36def exception_occured(out: IO = sys.stderr) -> NoReturn:
37 """
38 Print exception message and exit.
40 :param out: Output stream for message.
41 """
42 print('E: your request could not be fulfilled', file=out)
43 print('try `univention-config-registry --help` for more information', file=out)
44 sys.exit(1)
47SCOPE = ['default', 'normal', 'ldap', 'schedule', 'forced', 'custom']
50class Load(IntEnum):
51 MANUAL = 0
52 ONCE = 1
53 ALWAYS = 2
56if MYPY: # pragma: no cover
57 _M = Mapping[str, str]
58 _MM = MutableMapping[str, str]
59else:
60 _M = Mapping
61 _MM = MutableMapping
64class BooleanConfigRegistry:
65 """Mixin class for boolean operations."""
67 TRUE = frozenset({'yes', 'true', '1', 'enable', 'enabled', 'on'})
68 FALSE = frozenset({'no', 'false', '0', 'disable', 'disabled', 'off'})
70 def is_true(self, key: str | None = None, default: bool = False, value: str | None = None) -> bool:
71 """
72 Return if the strings value of key is considered as true.
74 :param key: UCR variable name.
75 :param default: Default value to return, if UCR variable is not set.
76 :param value: text string to directly evaluate instead of looking up the key.
77 :returns: `True` when the value is one of `yes`, `true`, `1`, `enable`, `enabled`, `on`.
79 >>> ucr = ConfigRegistry('/dev/null')
80 >>> ucr['key'] = 'yes'
81 >>> ucr.is_true('key')
82 True
83 >>> ucr.is_true('other')
84 False
85 >>> ucr.is_true('other', True)
86 True
87 >>> ucr.is_true(value='1')
88 True
89 """
90 if value is None:
91 value = self.get(key) # type: ignore
92 if value is None:
93 return default
94 return value.lower() in self.TRUE
96 def is_false(self, key: str | None = None, default: bool = False, value: str | None = None) -> bool:
97 """
98 Return if the strings value of key is considered as false.
100 :param key: UCR variable name.
101 :param default: Default value to return, if UCR variable is not set.
102 :param value: text string to directly evaulate instead of looking up the key.
103 :returns: `True` when the value is one of `no`, `false`, `0`, `disable`, `disabled`, `off`.
105 >>> ucr = ConfigRegistry('/dev/null')
106 >>> ucr['key'] = 'no'
107 >>> ucr.is_false('key')
108 True
109 >>> ucr.is_false('other')
110 False
111 >>> ucr.is_false('other', True)
112 True
113 >>> ucr.is_false(value='0')
114 True
115 """
116 if value is None:
117 value = self.get(key) # type: ignore
118 if value is None:
119 return default
120 return value.lower() in self.FALSE
123class ViewConfigRegistry(_M, BooleanConfigRegistry):
124 """Immutable view of UCR."""
126 def __init__(self, ucr: Mapping[str, str]) -> None:
127 self.ucr = ucr
129 def __getitem__(self, key: str) -> str:
130 return self.ucr.__getitem__(key)
132 def __iter__(self) -> Iterator[str]:
133 return self.ucr.__iter__()
135 def __len__(self) -> int:
136 return self.ucr.__len__()
139class ReadOnlyConfigRegistry(_M, BooleanConfigRegistry):
140 """
141 Merged persistent read-only value store.
142 This is a merged view of several sub-registries.
144 :param filename: File name for custom layer text database file.
145 """
147 DEFAULTS, NORMAL, LDAP, SCHEDULE, FORCED, CUSTOM = range(6)
148 LAYER_PRIORITIES = (CUSTOM, FORCED, SCHEDULE, LDAP, NORMAL, DEFAULTS)
149 PREFIX = '/etc/univention'
150 BASES = {
151 NORMAL: 'base.conf',
152 LDAP: 'base-ldap.conf',
153 SCHEDULE: 'base-schedule.conf',
154 FORCED: 'base-forced.conf',
155 DEFAULTS: 'base-defaults.conf',
156 }
158 def __init__(self, filename: str = "") -> None:
159 super().__init__()
160 custom = os.getenv('UNIVENTION_BASECONF') or filename
161 self.autoload = Load.MANUAL
163 self._registry: dict[int, _ConfigRegistry] = {}
164 for reg in self.LAYER_PRIORITIES:
165 if reg == self.CUSTOM:
166 self._registry[reg] = _ConfigRegistry(custom or os.devnull)
167 else:
168 self._registry[reg] = _ConfigRegistry(os.devnull if custom else os.path.join(self.PREFIX, self.BASES[reg]))
170 def _walk(self) -> Iterator[tuple[int, Self]]:
171 """
172 Iterator over layers.
174 :returns: Iterator of 2-tuple (layers-mumber, layer)
175 """
176 if self.autoload:
177 self.load(Load.MANUAL if self.autoload == Load.ONCE else self.autoload)
179 for reg in self.LAYER_PRIORITIES:
180 registry = self._registry[reg]
181 yield (reg, registry)
183 def load(self, autoload: Load = Load.MANUAL) -> Self:
184 """
185 Load registry from file.
187 :param autoload: Automatically reload changed files.
188 """
189 for reg in self._registry.values():
190 reg.load()
192 self.autoload = Load.MANUAL # prevent recursion!
193 strict = False # self.is_true('ucr/encoding/strict')
194 self.autoload = autoload
196 for reg in self._registry.values():
197 reg.strict_encoding = strict
199 return self
201 def __enter__(self) -> ViewConfigRegistry:
202 """
203 Return immutable view despite `autoload`.
205 :returns: A frozen registry.
207 > ucr_live = ConfigRegistry().load(autoload=Load.ALWAYS)
208 > with ucr_live as ucr_frozen:
209 > for key, value in ucr_frozen.items():
210 > print(key, value)
211 """
212 return ViewConfigRegistry(self._merge())
214 def __exit__(self, exc_type: type[BaseException] | None, exc_value: BaseException | None, traceback: TracebackType | None) -> None:
215 """Release registry view."""
217 def __getitem__(self, key: str) -> str | None: # type: ignore
218 """
219 Return registry value.
221 :param key: UCR variable name.
222 :returns: the value or `None`.
224 Bug #28276: ucr[key] returns None instead of raising KeyError - it would break many UCR templates!
225 """
226 return self.get(key)
228 def __contains__(self, key: str) -> bool: # type: ignore
229 """
230 Check if registry key is set.
232 :param key: UCR variable name.
233 :returns: `True` is set, `False` otherwise.
234 """
235 return any(key in registry for _reg, registry in self._walk())
237 def __iter__(self) -> Iterator[str]:
238 """
239 Iterate over all registry keys.
241 :returns: Iterator over all UCR variable names.
242 """
243 yield from self._merge()
245 def __len__(self) -> int:
246 """
247 Return length.
249 :returns: Number of UCR variables set.
250 """
251 merge = self._merge()
252 return len(merge)
254 @overload # type: ignore
255 def get(self, key: str, default: _VT, getscope: Literal[True]) -> tuple[int, str] | _VT: # pragma: no cover
256 pass
258 @overload
259 def get(self, key: str, default: _VT | None = None) -> str | _VT: # pragma: no cover
260 pass
262 def get(self, key: str, default: _VT | None = None, getscope: bool = False) -> str | tuple[int, str] | _VT | None:
263 """
264 Return registry value (including optional scope).
266 :param key: UCR variable name.
267 :param default: Default value when the UCR variable is not set.
268 :param getscope: `True` makes the method return the scope level in addition to the value itself.
269 :returns: the value or a 2-tuple (level, value) or the default.
270 """
271 for reg, registry in self._walk():
272 try:
273 value: str = registry[key]
274 except KeyError:
275 continue
276 if reg == self.DEFAULTS:
277 value = self._eval_default(value)
278 return (reg, value) if getscope else value
279 return default
281 @overload
282 def get_int(self, key: str) -> int | None: # pragma: no cover
283 pass
285 @overload # type: ignore
286 def get_int(self, key: str, default: _VT) -> int | _VT: # pragma: no cover
287 pass
289 def get_int(self, key: str, default: _VT | None = None) -> int | _VT | None:
290 """
291 Return registry value as int.
293 :param key: UCR variable name.
294 :param default: Default value when the UCR variable is not set.
295 :returns: the registry value or the default.
296 """
297 try:
298 return int(self[key]) # type: ignore
299 except (KeyError, TypeError, ValueError):
300 return default
302 @overload
303 def _merge(self) -> dict[str, str]: # pragma: no cover
304 pass
306 @overload
307 def _merge(self, getscope: Literal[True]) -> dict[str, tuple[int, str]]: # pragma: no cover
308 pass
310 def _merge(self, getscope: bool = False) -> dict[str, str] | dict[str, tuple[int, str]]:
311 """
312 Merge sub registry.
314 :param getscope: `True` makes the method return the scope level in addition to the value itself.
315 :returns: A mapping from varibal ename to eiter the value (if `getscope` is False) or a 2-tuple (level, value).
316 """
317 merge: dict[str, str | tuple[int, str]] = {}
318 for reg, registry in self._walk():
319 for key, value in registry.items():
320 if key not in merge:
321 if reg == self.DEFAULTS:
322 value = self._eval_default(value)
323 merge[key] = (reg, value) if getscope else value
325 return merge # type: ignore
327 def _eval_default(self, default: str) -> str:
328 """
329 Recursively evaluate default value.
331 :param default: Default value.
332 :returns: Substituted value.
333 """
334 try:
335 value = run_filter(default, self, opts={'disallow-execution': True})
336 except RuntimeError: # maximum recursion depth exceeded
337 value = b''
339 return value.decode("UTF-8")
341 @overload
342 def items(self) -> ItemsView[str, str]: # pragma: no cover
343 pass
345 @overload
346 def items(self, getscope: Literal[True]) -> ItemsView[str, tuple[int, str]]: # pragma: no cover
347 pass
349 def items(self, getscope: bool = False) -> ItemsView[str, str] | ItemsView[str, tuple[int, str]]:
350 """
351 Return all registry entries a 2-tuple (key, value) or (key, (scope, value)) if getscope is True.
353 :param getscope: `True` makes the method return the scope level in addition to the value itself.
354 :returns: A mapping from varibal ename to eiter the value (if `getscope` is False) or a 2-tuple (level, value).
355 """
356 merge = self._merge(getscope=getscope)
357 return merge.items() # type: ignore
359 def __str__(self) -> str:
360 """Return registry content as string."""
361 merge = self._merge()
362 return '\n'.join(['%s: %s' % (key, val) for key, val in merge.items()])
365class ConfigRegistry(ReadOnlyConfigRegistry, _MM):
366 """
367 Merged persistent value store.
368 This is a merged view of several sub-registries.
370 :param filename: File name for custom layer text database file.
371 :param write_registry: The UCR level used for writing.
372 """
374 def __init__(self, filename: str = "", write_registry: int = ReadOnlyConfigRegistry.NORMAL) -> None:
375 super().__init__(filename)
376 custom = os.getenv('UNIVENTION_BASECONF') or filename
377 self.scope = self.CUSTOM if custom else write_registry
378 for reg in self.LAYER_PRIORITIES:
379 registry = self._registry[reg]
380 registry._create_base_conf()
382 @property
383 def _layer(self) -> Self:
384 """Return selected layer."""
385 return self._registry[self.scope]
387 def save(self) -> None:
388 """Save registry to file."""
389 self._layer.save()
391 def lock(self) -> None:
392 """Lock registry file."""
393 self._layer.lock()
395 def unlock(self) -> None:
396 """Un-lock registry file."""
397 self._layer.unlock()
399 def __enter__(self) -> Self: # type: ignore
400 """
401 Lock Config Registry for read-modify-write cycle.
403 :returns: The locked registry.
405 > with ConfigRegistry() as ucr:
406 > ucr['key'] = 'value'
407 """
408 self.lock()
409 self.load()
410 return self
412 def __exit__(self, exc_type: type[BaseException] | None, exc_value: BaseException | None, traceback: TracebackType | None) -> None:
413 """Unlock registry."""
414 if exc_type is None:
415 self.save()
416 self.unlock()
418 def clear(self) -> None:
419 """Clear all registry keys."""
420 self._layer.clear()
422 def __delitem__(self, key: str) -> None:
423 """
424 Delete registry key.
426 :param key: UCR variable name.
427 """
428 del self._layer[key]
430 def __setitem__(self, key: str, value: str) -> None:
431 """
432 Set registry value.
434 :param key: UCR variable name.
435 :param value: UCR variable value.
436 """
437 self._layer[key] = value
439 def update(self, changes: dict[str, str | None]) -> dict[str, tuple[str | None, str | None]]: # type: ignore
440 """
441 Set or unset the given config registry variables.
443 :param changes: dictionary of ucr-variable-name: value-or-None.
444 :returns: A mapping from UCR variable name to a 2-tuple (old-value, new-value)
445 """
446 registry = self._layer
447 changed = {}
448 for key, value in changes.items():
449 old_value = registry.get(key, None)
450 if value is None:
451 try:
452 del registry[key]
453 except KeyError:
454 continue
455 else:
456 registry[key] = value
457 new_value = registry.get(key, value)
458 changed[key] = (old_value, new_value)
459 return changed
461 def setdefault(self, key: str, default: str) -> str: # type: ignore
462 """
463 Set value for variable only when not yet set.
465 :param key: UCR variable name.
466 :param default: UCR variable value.
467 :returns: The old value, if the variable was not yet set, otherwise the new value.
468 """
469 # Bug #28276: setdefault() required KeyError
470 value = self.get(key, default=self)
471 if value is self: 471 ↛ 474line 471 didn't jump to line 474 because the condition on line 471 was always true
472 value = self[key] = default
474 return value # type: ignore
477class _ConfigRegistry(dict):
478 """
479 Persistent value store.
480 This is a single value store using a text file.
482 :param filename: File name for text database file.
483 """
485 RE_COMMENT = re.compile(r'^[^:]*#.*$')
487 def __init__(self, filename: str) -> None:
488 dict.__init__(self)
489 self.file = filename
490 self.backup_file = self.file + '.bak'
491 self.lock_filename = self.file + '.lock'
492 # will be set by <ConfigRegistry> for each <_ConfigRegistry> - <True>
493 # means the backend files are valid UTF-8 and should stay that way -->
494 # only accept valid UTF-8
495 self.strict_encoding = False
496 self.lock_file: IO | None = None
497 self.mtime = 0.0
499 def load(self) -> None:
500 """Load sub registry from file."""
501 for fn in (self.file, self.backup_file):
502 new = {}
503 try:
504 file_stat = os.stat(fn)
505 if file_stat.st_mtime <= self.mtime and fn == self.file:
506 return
508 with open(fn, encoding='utf-8') as reg_file:
509 if reg_file.readline() == '' or reg_file.readline() == '':
510 continue
512 reg_file.seek(0)
513 for line in reg_file:
514 line = self.RE_COMMENT.sub("", line)
515 if line == '':
516 continue
517 if line.find(': ') == -1:
518 continue
520 key, value = line.split(': ', 1)
521 new[key] = value.strip()
523 break
524 except OSError:
525 pass
526 else:
527 return
529 self.mtime = file_stat.st_mtime
530 self.update(new)
531 for key in set(self.keys()) - set(new.keys()): 531 ↛ 532line 531 didn't jump to line 532 because the loop on line 531 never started
532 self.pop(key, None)
534 if fn != self.file:
535 self._save_file(self.file)
537 def _create_base_conf(self) -> None:
538 """Create sub registry file."""
539 try:
540 reg_file = os.open(self.file, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
541 os.close(reg_file)
542 except OSError as ex:
543 if ex.errno == errno.EEXIST and not os.path.isdir(self.file):
544 return
545 msg = "E: could not create file '%s': %s" % (self.file, ex)
546 print(msg, file=sys.stderr)
547 exception_occured()
549 def _save_file(self, filename: str) -> None:
550 """
551 Save sub registry to file.
553 :param filename: File name for saving.
554 :raises EnvironmentError: on fatal errors.
555 """
556 temp_filename = '%s.temp' % filename
557 try:
558 try:
559 file_stat = os.stat(filename)
560 if not S_ISREG(file_stat.st_mode):
561 return
562 except OSError:
563 file_stat = os.stat_result((0o0644, -1, -1, -1, 0, 0, -1, -1, -1, -1))
565 # open temporary file for writing
566 self._save_to(temp_filename)
567 try:
568 os.chmod(temp_filename, file_stat.st_mode)
569 os.chown(temp_filename, file_stat.st_uid, file_stat.st_gid)
570 os.rename(temp_filename, filename)
571 except OSError as ex:
572 if ex.errno == errno.EBUSY: 572 ↛ 573line 572 didn't jump to line 573 because the condition on line 572 was never true
573 with open(filename, 'w+', encoding='utf-8') as fd:
574 fd.write(open(temp_filename, encoding='utf-8').read())
575 os.unlink(temp_filename)
576 else:
577 # In this case the temp file created above in this
578 # function was already moved by a concurrent UCR
579 # operation. Dump the current state to a backup file
580 temp_filename = '%s.concurrent_%s' % (filename, time.time())
581 self._save_to(temp_filename)
582 except OSError as ex:
583 # suppress certain errors
584 if ex.errno != errno.EACCES:
585 raise
587 def _save_to(self, filename: str) -> None:
588 """
589 Serialize sub registry to file.
591 :param filename: File name for saving.
592 """
593 with open(filename, 'w', encoding='utf-8') as fd:
594 # write data to file
595 fd.write('# univention_ base.conf\n\n')
596 fd.write(self.__unicode__())
597 # flush (meta)data
598 fd.flush()
599 os.fsync(fd.fileno())
601 def save(self) -> None:
602 """Save sub registry to file."""
603 for filename in (self.backup_file, self.file):
604 self._save_file(filename)
606 def lock(self) -> None:
607 """Lock sub registry file."""
608 self.lock_file = lock = open(self.lock_filename, "a+", encoding='utf-8')
609 fcntl.flock(lock.fileno(), fcntl.LOCK_EX)
611 def unlock(self) -> None:
612 """Un-lock sub registry file."""
613 if self.lock_file is not None: 613 ↛ exitline 613 didn't return from function 'unlock' because the condition on line 613 was always true
614 self.lock_file.close()
615 self.lock_file = None
617 def __setitem__(self, key, value):
618 """
619 Set value in sub registry.
621 :param key: UCR variable name.
622 :param value: UCR variable value.
623 """
624 if self.strict_encoding: 624 ↛ 625line 624 didn't jump to line 625 because the condition on line 624 was never true
625 try:
626 key.decode('UTF-8') # only accept valid UTF-8 encoded bytes
627 except UnicodeError:
628 raise StrictModeException('variable name is not UTF-8 encoded')
629 try:
630 value.decode('UTF-8') # only accept valid UTF-8 encoded bytes
631 except UnicodeError:
632 raise StrictModeException('value is not UTF-8 encoded')
633 return dict.__setitem__(self, key, value)
635 @staticmethod
636 def remove_invalid_chars(seq: str) -> str:
637 """
638 Remove non-UTF-8 characters from value.
640 :param seq: Text string.
641 :returns: Text string with invalid characters removed.
642 """
643 for letter in INVALID_VALUE_CHARS:
644 seq = seq.replace(letter, '')
645 return seq
647 def __str__(self) -> str:
648 """Return sub registry content as string."""
649 return '\n'.join(
650 '%s: %s' % (key, self.remove_invalid_chars(val))
651 for key, val in sorted(self.items())
652 )
654 def __unicode__(self): # noqa: PLW3201
655 data = self.__str__()
656 if isinstance(data, bytes): 656 ↛ 657line 656 didn't jump to line 657 because the condition on line 656 was never true
657 data = data.decode('UTF-8')
658 return data