Coverage for python/univention/config_registry/handler.py: 54%
633 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
6#
7# API stability :pylint: disable-msg=R0201,W0613,R0903
8# Too pedantic :pylint: disable-msg=W0704
9# Rewrite :pylint: disable-msg=R0912
11"""Univention Configuration Registry handlers."""
14import errno
15import os
16import pickle # noqa: S403
17import random
18import re
19import subprocess
20import sys
21from collections.abc import Iterable, Mapping
22from grp import getgrnam
23from pwd import getpwnam
24from typing import IO, Any
26from univention.config_registry.misc import asciify, directory_files
27from univention.debhelper import parseRfc822 # pylint: disable-msg=W0403
30_OPT = Mapping[str, Any]
31_UCR = Mapping[str, str]
32_CHANGES = Mapping[str, tuple[str | None, str | None]]
33_ARG = tuple[_UCR, _CHANGES]
34_INFO = Mapping[str, list[str]]
36__all__ = ['ConfigHandlers']
38VARIABLE_PATTERN = re.compile('@%@([^@]+)@%@')
39VARIABLE_TOKEN = re.compile('@%@')
40EXECUTE_TOKEN = re.compile(b'@!@')
41WARNING_PATTERN = re.compile('(UCRWARNING|BCWARNING|UCRWARNING_ASCII)=(.+)')
43INFO_DIR = '/etc/univention/templates/info'
44FILE_DIR = '/etc/univention/templates/files'
45SCRIPT_DIR = '/etc/univention/templates/scripts'
46MODULE_DIR = '/etc/univention/templates/modules'
48WARNING_TEXT = '''\
49Warning: This file is auto-generated and might be overwritten by
50 univention-config-registry.
51 Please edit the following file(s) instead:
52Warnung: Diese Datei wurde automatisch generiert und kann durch
53 univention-config-registry ueberschrieben werden.
54 Bitte bearbeiten Sie an Stelle dessen die folgende(n) Datei(en):
56'''
57assert asciify(WARNING_TEXT) == WARNING_TEXT, "Only ASCII allowed in WARNING_TEXT"
60def run_filter(template: str, directory: _UCR, srcfiles: Iterable[str] = set(), opts: _OPT = {}) -> bytes:
61 """
62 Process a template file: substitute variables.
64 :param template: Text string of template.
65 :param directory: UCR instance.
66 :param srcfiles: File names of source template.
67 :param opts: Command line options.
68 :returns: The modified template with all UCR variables and sections replaced.
69 """
70 template = _replace_variables(template, directory, srcfiles)
72 tmpl = template.encode("UTF-8")
74 if opts.get('disallow-execution', False):
75 return tmpl
77 tmpl = _replace_exec(tmpl)
79 return tmpl
82def _replace_variables(template: str, directory: _UCR, srcfiles: Iterable[str]) -> str:
83 while True:
84 i = VARIABLE_TOKEN.finditer(template)
85 try:
86 start = next(i)
87 end = next(i)
88 name = template[start.end():end.start()]
90 if name in directory:
91 value = directory[name]
92 if not isinstance(value, str): 92 ↛ 94line 92 didn't jump to line 94 because the condition on line 92 was never true
93 # Python 2 with unicode value
94 value = value.encode('UTF-8') # important! template must not be of type unicode ever (in py2), otherwise some characters are lost in the below subprocess stdinput
95 else:
96 match = WARNING_PATTERN.match(name)
97 if match:
98 mode, prefix = match.groups()
99 value = warning_string(prefix, srcfiles=srcfiles)
100 if mode == "UCRWARNING_ASCII":
101 value = asciify(value)
102 else:
103 value = ''
105 if isinstance(value, list | tuple): 105 ↛ 106line 105 didn't jump to line 106 because the condition on line 105 was never true
106 value = value[0]
107 template = template[:start.start()] + value + template[end.end():]
108 except StopIteration:
109 break
111 return template
114def _replace_exec(template: bytes) -> bytes:
115 while True:
116 i = EXECUTE_TOKEN.finditer(template)
117 try:
118 start = next(i)
119 end = next(i)
121 proc = subprocess.Popen(
122 (sys.executable,),
123 stdin=subprocess.PIPE, stdout=subprocess.PIPE,
124 close_fds=True)
125 value = proc.communicate(b'''\
126# -*- coding: utf-8 -*-
127import univention.config_registry
128configRegistry = univention.config_registry.ConfigRegistry()
129configRegistry.load()
130# for compatibility
131baseConfig = configRegistry
132%s
133''' % template[start.end():end.start()])[0]
134 template = template[:start.start()] + value + template[end.end():]
136 except StopIteration:
137 break
139 return template
142def run_script(script: str, arg: str, changes: _CHANGES) -> None:
143 """
144 Execute script with command line arguments using a shell and pass changes
145 on STDIN.
146 For each changed variable a line with the 'name of the variable', the 'old
147 value', and the 'new value' are passed separated by '@%@'.
149 :param script: File name of the script.
150 :param arg: Execution mode, e.g. `generate` or `postinst`.
151 :param changes: Dictionary of changed UCR variables, mapping UCR variable names to 2-tuple (old-value, new-value).
152 """
153 diff = [
154 '%s@%%@%s@%%@%s\n' % (key, old, new)
155 for (key, (old, new)) in changes.items()
156 if old and new
157 ]
159 cmd = script + " " + arg
160 proc = subprocess.Popen(cmd, shell=True, stdin=subprocess.PIPE, close_fds=True)
161 proc.communicate(''.join(diff).encode('UTF-8'))
164def run_module(modpath: str, fn: str, ucr: _UCR, changes: _CHANGES) -> None:
165 """
166 Load the Python module that MUST be located in :py:const:`MODULE_DIR` or any
167 subdirectory.
169 :param modpath: The path to the module relative to :py:const:`MODULE_DIR`.
170 :param fn: Execution mode, e.g. `handler` or `preinst` or `postinst`.
171 :param ucr: UCR instance.
172 :param changes: Dictionary of changed UCR variables, mapping UCR variable names to 2-tuple (old-value, new-value).
173 """
174 # temporarily prepend MODULE_DIR to load path
175 sys.path.insert(0, MODULE_DIR)
176 module_name = os.path.splitext(modpath)[0]
177 try:
178 module = __import__(module_name.replace(os.path.sep, '.'))
179 f = getattr(module, fn)
180 f(ucr, changes)
181 except (AttributeError, ImportError) as ex:
182 print(ex, file=sys.stderr)
183 del sys.path[0]
186def warning_string(prefix: str = '# ', srcfiles: Iterable[str] = set()) -> str:
187 """
188 Generate UCR warning text.
190 :param prefix: String to prepend before each line.
191 :param srcfiles: File names of source template.
192 :returns: A warning sting based on :py:const:`WARNING_TEXT`.
193 """
194 res = [
195 '%s%s' % (prefix, line)
196 for line in WARNING_TEXT.splitlines()
197 ] + [
198 '%s\t%s' % (prefix, srcfile)
199 for srcfile in sorted(srcfiles)
200 ] + [
201 prefix,
202 ]
204 return "\n".join(res)
207class ConfigHandler:
208 """Base class of all config handlers."""
210 variables: set[str] = set()
212 def __call__(self, args: _ARG) -> None:
213 raise NotImplementedError()
216class ConfigHandlerDiverting(ConfigHandler):
217 """
218 File diverting config handler.
220 :param to_file: Destination file name.
221 """
223 def __init__(self, to_file: str) -> None:
224 super().__init__()
225 self.to_file = os.path.join('/', to_file)
226 self.user: int | None = None
227 self.group: int | None = None
228 self.mode: int | None = None
229 self.preinst: str | None = None
230 self.postinst: str | None = None
232 def __hash__(self):
233 """Return unique hash."""
234 return hash(self.to_file)
236 def __eq__(self, other):
237 """Compare this to other handler."""
238 if isinstance(other, type(self)): 238 ↛ 240line 238 didn't jump to line 240 because the condition on line 238 was always true
239 return self.to_file == other.to_file
240 return NotImplemented
242 def __ne__(self, other):
243 """Compare this to other handler."""
244 if isinstance(other, type(self)): 244 ↛ 246line 244 didn't jump to line 246 because the condition on line 244 was always true
245 return self.to_file != other.to_file
246 return NotImplemented
248 def _set_perm(self, stat: os.stat_result | None, to_file: str | None = None) -> None:
249 """
250 Set file permissions.
252 :param stat: File status.
253 :param to_file: Destination file name.
254 """
255 if not to_file:
256 to_file = self.to_file
257 elif self.to_file != to_file: 257 ↛ 265line 257 didn't jump to line 265 because the condition on line 257 was always true
258 try:
259 old_stat = os.stat(self.to_file)
260 os.chmod(to_file, old_stat.st_mode)
261 os.chown(to_file, old_stat.st_uid, old_stat.st_gid)
262 except OSError:
263 pass
265 if self.user or self.group or self.mode:
266 if self.mode:
267 os.chmod(to_file, self.mode)
269 if self.user and self.group:
270 os.chown(to_file, self.user, self.group)
271 elif self.user:
272 os.chown(to_file, self.user, 0)
273 elif self.group:
274 os.chown(to_file, 0, self.group)
275 elif stat:
276 os.chmod(to_file, stat.st_mode)
278 def _call_silent(self, *cmd: str) -> int:
279 """
280 Call command with stdin, stdout, and stderr redirected from/to :file:`/dev/null`.
282 :param cmd: List of command with arguments.
283 :returns: Process exit code.
284 """
285 with open(os.path.devnull, 'w', encoding='utf-8') as null:
286 # tell possibly wrapped dpkg-divert to really do the work
287 env = dict(os.environ)
288 env['DPKG_MAINTSCRIPT_PACKAGE'] = 'univention-config'
289 return subprocess.call(cmd, stdin=null, stdout=null, stderr=null, env=env)
291 def need_divert(self) -> bool:
292 """Check if diversion is needed."""
293 return False
295 def install_divert(self) -> None:
296 """Prepare file for diversion."""
297 deb = '%s.debian' % self.to_file
298 self._call_silent('dpkg-divert', '--quiet', '--rename', '--local', '--divert', deb, '--add', self.to_file)
299 # Make sure a valid file still exists
300 if os.path.exists(deb) and not os.path.exists(self.to_file): 300 ↛ 302line 300 didn't jump to line 302 because the condition on line 300 was never true
301 # Don't use shutil.copy2() which looses file ownership (Bug #22596)
302 self._call_silent('cp', '-p', deb, self.to_file)
304 def uninstall_divert(self) -> bool:
305 """
306 Undo diversion of file.
308 :returns: `True` because the diversion was removed.
309 """
310 try:
311 os.unlink(self.to_file)
312 except OSError:
313 pass
314 deb = '%s.debian' % self.to_file
315 self._call_silent('dpkg-divert', '--quiet', '--rename', '--local', '--divert', deb, '--remove', self.to_file)
316 return True
318 def _temp_file_name(self) -> str:
319 dirname, basename = os.path.split(self.to_file)
320 filename = '.%s__ucr__commit__%s' % (basename, random.random())
321 return os.path.join(dirname, filename)
323 def _install_file(self, tmp_to_file: str) -> None:
324 """
325 Atomically move the generated temporary file to its destination.
327 :param tmp_to_file: Path of the temporary file to move into place.
328 :raises OSError: when the destination cannot be written, e.g. when a directory exists at the destination path.
329 """
330 try:
331 os.rename(tmp_to_file, self.to_file)
332 except OSError as ex:
333 if ex.errno == errno.EBUSY:
334 with open(self.to_file, 'w+', encoding='utf-8') as fd:
335 fd.write(open(tmp_to_file, encoding='utf-8').read())
336 os.unlink(tmp_to_file)
337 else:
338 print('E: failed to write %s: %s' % (self.to_file, ex.strerror), file=sys.stderr)
339 raise
342class ConfigHandlerMultifile(ConfigHandlerDiverting):
343 """
344 Handler for multifile.
346 :param dummy_from_file: Source file name used to copy file permissions from.
347 :param to_file: Destination file name.
348 """
350 def __init__(self, dummy_from_file: str, to_file: str) -> None:
351 super().__init__(to_file)
352 self.variables: set[str] = set()
353 self.from_files: set[str] = set()
354 self.dummy_from_file = dummy_from_file
355 self.def_count = 1
357 def __setstate__(self, state):
358 """Load state upon unpickling."""
359 self.__dict__.update(state)
360 # may raise AttributeError, which forces UCR to rebuild the cache
361 self.def_count # :pylint: disable-msg=W0104 # noqa: B018
363 def add_subfiles(self, subfiles: list[tuple[str, set[str]]]) -> None:
364 """
365 Add subfiles to multifile.
367 :param subfiles: List of 2-tuples (file-name, set-of-variable-names).
368 """
369 for from_file, variables in subfiles:
370 self.from_files.add(from_file)
371 self.variables |= variables
373 def remove_subfile(self, subfile: str) -> None:
374 """
375 Remove subfile.
377 Removed diversion of set of sub-files becomes empty.
378 """
379 self.from_files.discard(subfile)
380 if not self.need_divert(): 380 ↛ 381line 380 didn't jump to line 381 because the condition on line 380 was never true
381 self.uninstall_divert()
383 def __call__(self, args: _ARG) -> None:
384 """Generate multfile from subfile templates."""
385 ucr, changed = args
386 print('Multifile: %s' % self.to_file)
388 if hasattr(self, 'preinst') and self.preinst:
389 run_module(self.preinst, 'preinst', ucr, changed)
391 if self.def_count == 0 or not self.from_files:
392 return
394 to_dir = os.path.dirname(self.to_file)
395 if not os.path.isdir(to_dir):
396 os.makedirs(to_dir, 0o755)
398 if os.path.isfile(self.dummy_from_file):
399 stat: os.stat_result | None = os.stat(self.dummy_from_file)
400 elif os.path.isfile(self.to_file):
401 stat = os.stat(self.to_file)
402 else:
403 stat = None
405 tmp_to_file = self._temp_file_name()
406 try:
407 filter_opts: dict[str, Any] = {}
409 with open(tmp_to_file, 'wb') as to_fp:
410 self._set_perm(stat, tmp_to_file)
412 for from_file in sorted(self.from_files, key=os.path.basename):
413 try:
414 with open(from_file, encoding='utf-8') as from_fp:
415 to_fp.write(run_filter(from_fp.read(), ucr, srcfiles=self.from_files, opts=filter_opts))
416 except OSError:
417 continue
419 self._install_file(tmp_to_file)
420 except Exception:
421 if os.path.exists(tmp_to_file):
422 os.unlink(tmp_to_file)
423 raise
425 if hasattr(self, 'postinst') and self.postinst:
426 run_module(self.postinst, 'postinst', ucr, changed)
428 script_file = os.path.join(SCRIPT_DIR, self.to_file.strip("/"))
429 if os.path.isfile(script_file):
430 run_script(script_file, 'postinst', changed)
432 def need_divert(self) -> bool:
433 """
434 Diversion is needed when at least one multifile and one subfile
435 definition exists.
436 """
437 return self.def_count >= 1 and bool(self.from_files)
439 def install_divert(self) -> None:
440 """Prepare file for diversion."""
441 if self.need_divert():
442 super().install_divert()
444 def uninstall_divert(self) -> bool:
445 """
446 Undo diversion of file.
448 :returns: `True` when the diversion is removed, `False` when the diversion is still needed.
449 """
450 if self.need_divert():
451 return False
452 return super().uninstall_divert()
455class ConfigHandlerFile(ConfigHandlerDiverting):
456 """
457 Handler for (single)file.
459 :param from_file: Template source file name.
460 :param to_file: Destination file name.
461 """
463 def __init__(self, from_file: str, to_file: str) -> None:
464 super().__init__(to_file)
465 self.from_file = from_file
467 def __call__(self, args: _ARG) -> None:
468 """Generate file from template."""
469 ucr, changed = args
471 if hasattr(self, 'preinst') and self.preinst:
472 run_module(self.preinst, 'preinst', ucr, changed)
474 print('File: %s' % self.to_file)
476 to_dir = os.path.dirname(self.to_file)
477 if not os.path.isdir(to_dir):
478 os.makedirs(to_dir, 0o755)
480 try:
481 stat = os.stat(self.from_file)
482 except OSError:
483 print("The referenced template file does not exist", file=sys.stderr)
484 return
486 tmp_to_file = self._temp_file_name()
487 try:
488 filter_opts: dict[str, Any] = {}
490 with open(self.from_file, encoding='utf-8') as from_fp, open(tmp_to_file, 'wb') as to_fp:
491 self._set_perm(stat, tmp_to_file)
493 to_fp.write(run_filter(from_fp.read(), ucr, srcfiles=[self.from_file], opts=filter_opts))
495 self._install_file(tmp_to_file)
496 except Exception:
497 if os.path.exists(tmp_to_file):
498 os.unlink(tmp_to_file)
499 raise
501 if hasattr(self, 'postinst') and self.postinst:
502 run_module(self.postinst, 'postinst', ucr, changed)
504 script_file = self.from_file.replace(FILE_DIR, SCRIPT_DIR)
505 if os.path.isfile(script_file):
506 run_script(script_file, 'postinst', changed)
508 def need_divert(self) -> bool:
509 """For simple files the diversion is always needed."""
510 return True
513class ConfigHandlerScript(ConfigHandler):
514 """
515 Handler for UCR scripts.
517 :param script: Script file name.
518 """
520 def __init__(self, script: str) -> None:
521 super().__init__()
522 self.script = script
524 def __hash__(self):
525 """Return unique hash."""
526 return hash(self.script)
528 def __eq__(self, other):
529 """Compare this to other handler."""
530 if isinstance(other, type(self)): 530 ↛ 532line 530 didn't jump to line 532 because the condition on line 530 was always true
531 return self.script == other.script
532 return NotImplemented
534 def __ne__(self, other):
535 """Compare this to other handler."""
536 if isinstance(other, type(self)): 536 ↛ 538line 536 didn't jump to line 538 because the condition on line 536 was always true
537 return self.script != other.script
538 return NotImplemented
540 def __call__(self, args: _ARG) -> None:
541 """Call external programm after change."""
542 _ucr, changed = args
543 print('Script: %s' % self.script)
544 if os.path.isfile(self.script): 544 ↛ exitline 544 didn't return from function '__call__' because the condition on line 544 was always true
545 run_script(self.script, 'generate', changed)
548class ConfigHandlerModule(ConfigHandler):
549 """
550 Handler for UCR Python module.
552 :param module: Module file name.
553 """
555 def __init__(self, module: str) -> None:
556 super().__init__()
557 self.module = module
559 def __hash__(self):
560 """Return unique hash."""
561 return hash(self.module)
563 def __eq__(self, other):
564 """Compare this to other handler."""
565 if isinstance(other, type(self)): 565 ↛ 567line 565 didn't jump to line 567 because the condition on line 565 was always true
566 return self.module == other.module
567 return NotImplemented
569 def __ne__(self, other):
570 """Compare this to other handler."""
571 if isinstance(other, type(self)): 571 ↛ 573line 571 didn't jump to line 573 because the condition on line 571 was always true
572 return self.module != other.module
573 return NotImplemented
575 def __call__(self, args: _ARG) -> None:
576 """Call Python module after change."""
577 ucr, changed = args
578 print('Module: %s' % self.module)
579 run_module(self.module, 'handler', ucr, changed)
582def grep_variables(text: str) -> set[str]:
583 """
584 Search UCR template text for used variables.
586 :returns: Set of all variables inside `@%@` delimiters.
587 """
588 return set(VARIABLE_PATTERN.findall(text))
591class ConfigHandlers:
592 """Manage handlers for configuration variables."""
594 CACHE_FILE = '/var/cache/univention-config/cache'
595 # 0: without version
596 # 1: with version header
597 # 2: switch to handlers mapping to set, drop file, add multifile.def_count
598 # 3: split config_registry into sub modules
599 VERSION = 3
600 VERSION_MIN = 3
601 VERSION_MAX = 3
602 VERSION_TEXT = 'univention-config cache, version'
603 VERSION_NOTICE = '%s %s\n' % (VERSION_TEXT, VERSION)
604 VERSION_RE = re.compile('^%s (?P<version>[0-9]+)$' % VERSION_TEXT)
606 _handlers: dict[str, set[ConfigHandler]] = {} # variable -> set(handlers)
607 _multifiles: dict[str, ConfigHandlerMultifile] = {} # multifile -> handler
608 _subfiles: dict[str, list[tuple[str, set[str]]]] = {} # multifile -> [(subfile, variables)] // pending
610 def __init__(self) -> None:
611 pass
613 @staticmethod
614 def _get_cache_version(cache_file: IO) -> int:
615 """
616 Read cached `.info` data.
618 :param cache_file: Opened cache file.
619 :returns: Version.
620 """
621 line = cache_file.readline() # IOError is propagated
622 match = ConfigHandlers.VERSION_RE.match(line)
623 if match:
624 version = int(match.group('version'))
625 # "Old style" cache (version 0) doesn't contain version notice
626 else:
627 cache_file.seek(0)
628 version = 0
629 return version
631 @classmethod
632 def _parse_rfc822_file(cls, fname: str) -> list[dict[str, list[str]]]:
633 """
634 Parse :rfc:`822` file.
636 :param fname: Path to :rfc:`822` file.
637 :returns: A list of dictionaries.
638 """
639 with open(fname, encoding='utf-8') as fd:
640 return parseRfc822(fd.read())
642 def load(self) -> None:
643 """Load cached `.info` data or force update."""
644 try:
645 with open(ConfigHandlers.CACHE_FILE, 'rb') as cache_file: 645 ↛ exitline 645 didn't return from function 'load' because
646 version = self._get_cache_version(cache_file)
647 chv = ConfigHandlers
648 if not chv.VERSION_MIN <= version <= chv.VERSION_MAX:
649 raise TypeError("Invalid cache file version.")
650 pickler = pickle.Unpickler(cache_file)
651 self._handlers = pickler.load()
652 if version <= 1:
653 # version <= 1: _handlers[multifile] -> [handlers]
654 # version >= 2: _handlers[multifile] -> set([handlers])
655 self._handlers = {k: set(v) for k, v in self._handlers.items()}
656 # version <= 1: _files UNUSED
657 pickler.load()
658 self._subfiles = pickler.load()
659 self._multifiles = pickler.load()
660 except (Exception, pickle.UnpicklingError):
661 self.update()
663 def get_handler(self, entry: _INFO) -> ConfigHandler | None:
664 """
665 Parse entry and return Handler instance.
667 :param entry: `.info` file entry dictionary.
668 :returns: An instance of `None`.
669 """
670 try:
671 typ = entry['Type'][0]
672 handler = getattr(self, '_get_handler_%s' % typ)
673 except (LookupError, AttributeError):
674 return None
675 else:
676 return handler(entry)
678 def _parse_common_file_handler(self, handler: ConfigHandlerDiverting, entry: _INFO) -> None:
679 """
680 Parse common file and multifile entries.
682 :param handler: Handler instance.
683 :param entry: `.info` file entry dictionary.
684 """
685 try:
686 handler.preinst = entry['Preinst'][0]
687 except LookupError:
688 pass
690 try:
691 handler.postinst = entry['Postinst'][0]
692 except LookupError:
693 pass
695 handler.variables |= set(entry.get('Variables', set()))
697 try:
698 user = entry['User'][0]
699 except LookupError:
700 pass
701 else:
702 try:
703 handler.user = getpwnam(user).pw_uid
704 except LookupError:
705 print(('W: failed to convert the username %s to the uid' % (user,)), file=sys.stderr)
707 try:
708 group = entry['Group'][0]
709 except LookupError:
710 pass
711 else:
712 try:
713 handler.group = getgrnam(group).gr_gid
714 except LookupError:
715 print(('W: failed to convert the groupname %s to the gid' % (group,)), file=sys.stderr)
717 try:
718 mode = entry['Mode'][0]
719 except LookupError:
720 pass
721 else:
722 try:
723 handler.mode = int(mode, 8)
724 except ValueError:
725 print('W: failed to convert mode %s' % (mode,), file=sys.stderr)
727 def _get_handler_file(self, entry: _INFO) -> ConfigHandlerFile | None:
728 """
729 Parse file entry and return Handler instance.
731 :param entry: `.info` file entry dictionary.
732 :returns: An instance of `None`.
733 """
734 try:
735 name = entry['File'][0]
736 except LookupError:
737 return None
738 from_path = os.path.join(FILE_DIR, name)
739 handler = ConfigHandlerFile(from_path, name)
740 if os.path.exists(from_path): 740 ↛ 741line 740 didn't jump to line 741 because the condition on line 740 was never true
741 with open(from_path, encoding='utf-8') as fd:
742 handler.variables = grep_variables(fd.read())
744 self._parse_common_file_handler(handler, entry)
746 return handler
748 def _get_handler_script(self, entry: _INFO) -> ConfigHandlerScript | None:
749 """
750 Parse script entry and return Handler instance.
752 :param entry: `.info` file entry dictionary.
753 :returns: An instance of `None`.
754 """
755 try:
756 script = entry['Script'][0]
757 variables = entry['Variables']
758 except LookupError:
759 return None
760 handler = ConfigHandlerScript(os.path.join(SCRIPT_DIR, script))
761 handler.variables = set(variables)
762 return handler
764 def _get_handler_module(self, entry: _INFO) -> ConfigHandlerModule | None:
765 """
766 Parse module entry and return Handler instance.
768 :param entry: `.info` file entry dictionary.
769 :returns: An instance of `None`.
770 """
771 try:
772 module = entry['Module'][0]
773 variables = entry['Variables']
774 except LookupError:
775 return None
776 handler = ConfigHandlerModule(os.path.splitext(module)[0])
777 handler.variables = set(variables)
778 return handler
780 def _get_handler_multifile(self, entry: _INFO) -> ConfigHandlerMultifile | None:
781 """
782 Parse multifile entry and return Handler instance.
784 :param entry: `.info` file entry dictionary.
785 :returns: An instance of `None`.
786 """
787 try:
788 mfile = entry['Multifile'][0]
789 except LookupError:
790 return None
791 try:
792 handler = self._multifiles[mfile]
793 handler.def_count += 1
794 except KeyError:
795 from_path = os.path.join(FILE_DIR, mfile)
796 handler = ConfigHandlerMultifile(from_path, mfile)
798 self._parse_common_file_handler(handler, entry)
800 # Add pending subfiles from earlier entries
801 self._multifiles[mfile] = handler
802 try:
803 file_vars = self._subfiles.pop(mfile)
804 handler.add_subfiles(file_vars)
805 except KeyError:
806 pass
808 return handler
810 def _get_handler_subfile(self, entry: dict[str, list[str]]) -> ConfigHandlerMultifile | None:
811 """
812 Parse subfile entry and return Handler instance.
814 :param entry: `.info` file entry dictionary.
815 :returns: An instance of `None`.
816 """
817 try:
818 mfile = entry['Multifile'][0]
819 subfile = entry['Subfile'][0]
820 except LookupError:
821 return None
822 variables = set(entry.get('Variables', set()))
823 name = os.path.join(FILE_DIR, subfile)
824 try:
825 with open(name, encoding='utf-8') as temp_file: 825 ↛ 826, 825 ↛ 8302 missed branches: 1) line 825 didn't jump to line 826 because , 2) line 825 didn't jump to line 830 because
826 variables |= grep_variables(temp_file.read())
827 except OSError:
828 print("Failed to process Subfile %s" % (name,), file=sys.stderr)
829 return None
830 qentry = (name, variables)
831 # if multifile handler does not yet exists, queue subfiles for later
832 try:
833 handler = self._multifiles[mfile]
834 handler.add_subfiles([qentry])
835 except KeyError:
836 pending = self._subfiles.setdefault(mfile, [])
837 pending.append(qentry)
838 return None
839 return handler
841 def update(self) -> set[ConfigHandler]:
842 """
843 Parse all `.info` files to build list of handlers.
845 :returns: Set of all handlers.
846 """
847 self._handlers.clear()
848 self._multifiles.clear()
849 self._subfiles.clear()
851 handlers: set[ConfigHandler] = set()
852 for info in directory_files(INFO_DIR): 852 ↛ 853line 852 didn't jump to line 853 because the loop on line 852 never started
853 if not info.endswith('.info'):
854 continue
855 for section in self._parse_rfc822_file(info):
856 handler = self.get_handler(section)
857 if handler:
858 handlers.add(handler)
859 for handler in handlers: 859 ↛ 860line 859 didn't jump to line 860 because the loop on line 859 never started
860 for variable in handler.variables:
861 v2h = self._handlers.setdefault(variable, set())
862 v2h.add(handler)
864 self._save_cache()
865 return handlers
867 def update_divert(self, handlers: Iterable[ConfigHandler]) -> None:
868 """
869 Synchronize diversions with handlers.
871 :param handlers: List of handlers.
872 """
873 wanted = {h.to_file: h for h in handlers if isinstance(h, ConfigHandlerDiverting) and h.need_divert()}
874 to_remove: set[str] = set()
875 # Scan for diversions done by UCR
876 with open('/var/lib/dpkg/diversions', encoding='utf-8') as div_file:
877 # from \n to \n package \n
878 try:
879 while True:
880 path_from = next(div_file).rstrip()
881 path_to = next(div_file).rstrip()
882 diversion = next(div_file).rstrip()
883 if path_from + '.debian' != path_to:
884 continue
885 if diversion != ':': # local diversion
886 continue
887 assert path_from not in to_remove # no duplicates
888 try:
889 handler = wanted.pop(path_from)
890 except KeyError:
891 to_remove.add(path_from)
892 except StopIteration:
893 pass
895 # Remove existing diversion not wanted
896 for path in to_remove:
897 tmp_handler = ConfigHandlerDiverting(path)
898 tmp_handler.uninstall_divert()
899 # Install missing diversions still wanted
900 for path, handler in wanted.items():
901 handler.install_divert()
903 def _save_cache(self) -> None:
904 """Write cache file."""
905 try:
906 with open(ConfigHandlers.CACHE_FILE, 'wb') as cache_file:
907 cache_file.write(ConfigHandlers.VERSION_NOTICE.encode('utf-8'))
908 pickler = pickle.Pickler(cache_file)
909 pickler.dump(self._handlers)
910 pickler.dump(self._subfiles)
911 pickler.dump(self._multifiles)
912 except OSError as ex:
913 if ex.errno != errno.EACCES:
914 raise
916 def register(self, package: str, ucr: _UCR) -> set[ConfigHandler]:
917 """
918 Register new info file for package.
920 :param package: Name of the package to register.
921 :param ucr: UCR instance.
922 :returns: Set of (new) handlers.
923 """
924 handlers: set[ConfigHandler] = set()
925 fname = os.path.join(INFO_DIR, '%s.info' % package)
926 for section in self._parse_rfc822_file(fname):
927 handler = self.get_handler(section)
928 if handler:
929 handlers.add(handler)
931 for handler in handlers:
932 if isinstance(handler, ConfigHandlerDiverting):
933 handler.install_divert()
935 values: dict[str, tuple[None, str | None]] = {}
936 for variable in handler.variables:
937 v2h = self._handlers.setdefault(variable, set())
938 v2h.add(handler)
939 values[variable] = (None, ucr[variable])
940 try:
941 _re = re.compile(variable)
942 except re.error:
943 continue
945 values.update((key, (None, val)) for key, val in ucr.items() if _re.match(key))
947 handler((ucr, values))
949 self._save_cache()
950 return handlers
952 def unregister(self, package: str, ucr: _UCR) -> set[ConfigHandler]:
953 """
954 Un-register info file for package.
956 :param package: Name of the package to un-register.
957 :param ucr: UCR instance.
958 :returns: Set of (then obsolete) handlers.
959 """
960 obsolete_handlers: set[ConfigHandler] = set()
961 mf_handlers: set[ConfigHandler] = set() # Remaining Multifile handlers
962 fname = os.path.join(INFO_DIR, '%s.info' % package)
963 for section in self._parse_rfc822_file(fname):
964 try:
965 typ = section['Type'][0]
966 except LookupError:
967 continue
968 if typ == 'file':
969 handler = self.get_handler(section)
970 elif typ == 'subfile':
971 mfile = section['Multifile'][0]
972 sfile = section['Subfile'][0]
973 try:
974 handler = self._multifiles[mfile]
975 except KeyError:
976 continue # skip SubFile w/o MultiFile
977 name = os.path.join(FILE_DIR, sfile)
978 handler.remove_subfile(name)
979 mf_handlers.add(handler)
980 elif typ == 'multifile':
981 mfile = section['Multifile'][0]
982 handler = self._multifiles[mfile]
983 handler.def_count -= 1
984 mf_handlers.add(handler)
985 else:
986 continue
987 if not handler: # Bug #17913
988 print(("Skipping internal error: no handler for %r in %s" % (section, package)), file=sys.stderr)
989 continue
990 if isinstance(handler, ConfigHandlerDiverting) and handler.uninstall_divert():
991 obsolete_handlers.add(handler)
993 for handler in mf_handlers - obsolete_handlers:
994 self.call_handler(ucr, handler)
996 try:
997 # remove cache file to force rebuild of cache
998 os.unlink(ConfigHandlers.CACHE_FILE)
999 except OSError:
1000 pass
1001 return obsolete_handlers
1003 def __call__(self, variables: Iterable[str], arg: _ARG) -> None:
1004 """
1005 Call handlers registered for changes in variables.
1007 :param variables: Changed UCR variable names.
1008 :param arg: 2-tuple(UCR-instance, changed) where changed is a dictionary mapping ucs-variable-names to values.
1009 """
1010 if not variables:
1011 return
1012 pending_handlers: set[ConfigHandler] = set()
1014 for reg_var, handlers in self._handlers.items(): 1014 ↛ 1015line 1014 didn't jump to line 1015 because the loop on line 1014 never started
1015 try:
1016 _re = re.compile(reg_var)
1017 except re.error as ex:
1018 print('Failed to compile regular expression %s: %s' % (reg_var, ex), file=sys.stderr)
1019 continue
1021 for variable in variables:
1022 if _re.match(variable):
1023 pending_handlers |= handlers
1025 for handler in pending_handlers: 1025 ↛ 1026line 1025 didn't jump to line 1026 because the loop on line 1025 never started
1026 handler(arg)
1028 def commit(self, ucr: _UCR, filelist: Iterable[str] = []) -> None:
1029 """
1030 Call handlers to (re-)generate files.
1032 :param ucr: UCR instance.
1033 :param filelist: List of files to re-generate. By default *all* files will be re-generated and all modules and scripts will we re-invoked!
1034 """
1035 _filelist = []
1036 for fname in filelist:
1037 fname = os.path.expanduser(fname)
1038 fname = os.path.expandvars(fname)
1039 fname = os.path.abspath(fname)
1040 _filelist.append(fname)
1042 # find handlers
1043 pending_handlers = set()
1044 for fname in directory_files(INFO_DIR): 1044 ↛ 1045line 1044 didn't jump to line 1045 because the loop on line 1044 never started
1045 if not fname.endswith('.info'):
1046 continue
1047 for section in self._parse_rfc822_file(fname):
1048 if not section.get('Type'):
1049 continue
1050 handler = None
1051 if _filelist:
1052 files = section.get('File') or section.get('Multifile') or ()
1053 for filename in files:
1054 if not os.path.isabs(filename):
1055 filename = '/%s' % filename
1056 if filename in _filelist:
1057 handler = self.get_handler(section)
1058 break
1059 else:
1060 continue
1061 else:
1062 handler = self.get_handler(section)
1064 if handler:
1065 pending_handlers.add(handler)
1067 # print missing files
1068 for fname in set(_filelist) - {h.to_file for h in pending_handlers if isinstance(h, ConfigHandlerDiverting)}:
1069 print('Warning: The file %r is not registered as an UCR template.' % (fname,), file=sys.stderr)
1071 # call handlers
1072 for handler in pending_handlers: 1072 ↛ 1073line 1072 didn't jump to line 1073 because the loop on line 1072 never started
1073 self.call_handler(ucr, handler)
1075 def call_handler(self, ucr: _UCR, handler: ConfigHandler) -> None:
1076 """
1077 Call handler passing current configuration variables.
1079 :param ucr: UCR instance.
1080 :param handler: The handler to call.
1081 """
1082 values: dict[str, tuple[str | None, str | None]] = {}
1083 for variable in handler.variables:
1084 if variable in self._handlers.keys():
1085 if ".*" in variable:
1086 for i in range(4):
1087 var = variable.replace(".*", "%s" % i)
1088 val = ucr.get(var)
1089 values[var] = (val, val)
1090 else:
1091 val = ucr.get(variable)
1092 values[variable] = (val, val)
1093 handler((ucr, values))