Coverage for python/univention/service_info.py: 96%
182 statements
« prev ^ index » next coverage.py v7.6.0, created at 2026-09-11 14:11 +0000
1#
2# Univention Configuration Registry
3# Service information: read information about registered Config Registry
4# variables
5#
6# SPDX-FileCopyrightText: 2007-2026 Univention GmbH
7# SPDX-License-Identifier: AGPL-3.0-only
9import os
10import shlex
11import subprocess
12from collections.abc import Iterable, Iterator, Sequence
13from logging import getLogger
14from typing import Any
16import univention.info_tools as uit
19class ServiceError(Exception):
20 """Error when starting, stopping or restarting a service."""
23class Service(uit.LocalizedDictionary):
24 """Description for a system service."""
26 REQUIRED = frozenset(('description', 'programs'))
27 OPTIONAL = frozenset(('start_type', 'systemd', 'icon', 'name', 'init_script'))
28 KNOWN = REQUIRED | OPTIONAL
30 def __init__(self, *args: Any, **kwargs: Any) -> None:
31 uit.LocalizedDictionary.__init__(self, *args, **kwargs)
32 self.running = False
34 def __repr__(self) -> str:
35 return '%s(%s)' % (
36 self.__class__.__name__,
37 dict.__repr__(self),
38 )
40 def check(self) -> list[str]:
41 """Check service entry for validity, returning list of incomplete entries."""
42 incomplete = [key for key in self.REQUIRED if not self.get(key, None)]
43 unknown = [key for key in self.keys() if key.lower() not in self.KNOWN]
44 return incomplete + unknown
46 def _update_status(self) -> None:
47 for prog in self['programs'].split(','):
48 prog = prog.strip()
49 if prog and not pidof(prog):
50 self.running = False
51 break
52 else:
53 self.running = True
55 def start(self) -> bool:
56 """Start the service."""
57 return self._change_state('start')
59 def stop(self) -> bool:
60 """Stop the service."""
61 return self._change_state('stop')
63 def restart(self) -> bool:
64 """Restart the service."""
65 return self._change_state('restart')
67 def status(self):
68 """Get status of the service."""
69 try:
70 return self.__change_state('status')[1]
71 except OSError:
72 return ''
74 def _change_state(self, action: str) -> bool:
75 rc, output = self.__change_state(action)
76 if rc:
77 raise ServiceError(self.status() or output)
78 return True
80 def __change_state(self, action: str) -> tuple[int, str]:
81 if self.get('init_script'):
82 # samba currently must not be started via systemd
83 return self._exec(('/etc/init.d/%s' % (self['init_script'],), action))
85 service_name = self._service_name()
86 return self._exec(('/usr/sbin/service', service_name, action))
88 def _service_name(self) -> str:
89 service_name = self.get('systemd') or self['name']
90 if service_name.endswith('.service'):
91 service_name = service_name.rsplit('.', 1)[0]
92 return service_name
94 def _exec(self, args: Sequence[str]) -> tuple[int, str]:
95 process = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, close_fds=True)
96 output = process.communicate()[0]
97 return process.returncode, output.decode('utf-8', 'replace').strip()
100def pidof(name: str, docker: int | str = '/var/run/docker.pid') -> list[int]:
101 """
102 Return list of process IDs matching name.
104 :param name: Procress name.
105 :param docker: File name containing process ID of docker process.
107 >>> import os,sys;os.getpid() in list(pidof(os.path.realpath(sys.executable))) + list(pidof(sys.executable)) + list(pidof(sys.argv[0]))
108 True
109 """
110 result: set[int] = set()
111 log = getLogger(__name__)
113 children: dict[int | str, list[int]] = {}
114 if isinstance(docker, str):
115 try:
116 with open(docker) as stream:
117 docker = int(stream.read(), 10)
118 log.info('Found docker.pid=%d', docker)
119 except (OSError, ValueError) as ex:
120 log.info('No docker found: %s', ex)
122 cmd = shlex.split(name)
123 for proc in os.listdir('/proc'):
124 try:
125 pid = int(proc, 10)
126 except ValueError:
127 continue
128 cmdline = os.path.join('/proc', proc, 'cmdline')
129 try:
130 with open(cmdline, 'rb') as fd:
131 commandline = fd.read().rstrip(b'\x00').decode('UTF-8', 'replace')
132 link = os.readlink(os.path.join('/proc', proc, 'exe'))
133 except OSError:
134 continue
135 # kernel thread
136 if not commandline: 136 ↛ 137line 136 didn't jump to line 137 because the condition on line 136 was never true
137 continue
139 if docker: 139 ↛ 150line 139 didn't jump to line 150 because the condition on line 139 was always true
140 stat = os.path.join('/proc', proc, 'stat')
141 status = None
142 try:
143 with open(stat, 'rb') as fd:
144 status = fd.readline()
145 ppid = int(status[status.rfind(b')') + 2:].split()[1], 10)
146 children.setdefault(ppid, []).append(pid)
147 except (OSError, ValueError) as ex:
148 log.error('Failed getting parent: %s: %r', ex, status)
150 def _running(cmd: list[str], link: str, commandline: str) -> Iterator[bool]:
151 yield cmd == [link]
152 args = commandline.split('\x00') if '\x00' in commandline else shlex.split(commandline)
153 yield len(cmd) == 1 and cmd[0] in args # FIXME: it detects "vim /usr/sbin/service" as running process!
154 yield len(cmd) > 1 and all(a == c for a, c in zip(args, cmd)) 154 ↛ exitline 154 didn't run the generator expression on line 154
155 if any(_running(cmd, link, commandline)):
156 log.info('found %d: %r', pid, commandline)
157 result.add(pid)
159 if docker:
160 remove = children.pop(docker, [])
161 while remove: 161 ↛ 162line 161 didn't jump to line 162 because the condition on line 161 was never true
162 pid = remove.pop()
163 log.debug('Removing docker child %s', pid)
164 result.discard(pid)
165 remove += children.pop(pid, [])
167 return list(result)
170class ServiceInfo:
171 BASE_DIR = '/etc/univention/service.info'
172 SERVICES = 'services'
173 CUSTOMIZED = '_customized'
174 FILE_SUFFIX = '.cfg'
176 def __init__(self, install_mode: bool = False) -> None:
177 self.services: dict = {}
178 if not install_mode:
179 self.__load_services()
180 self.update_services()
182 def update_services(self) -> None:
183 """Update the run state of all services."""
184 for serv in self.services.values():
185 serv._update_status()
187 def check_services(self) -> dict[str, list[str]]:
188 """
189 Check service descriptions for completeness.
191 :returns: dictionary of incomplete service descriptions.
192 """
193 incomplete: dict[str, list[str]] = {}
194 for name, srv in self.services.items():
195 miss = srv.check()
196 if miss:
197 incomplete[name] = miss
198 return incomplete
200 def write_customized(self) -> bool:
201 """Save service cusomization."""
202 filename = os.path.join(ServiceInfo.BASE_DIR, ServiceInfo.SERVICES, ServiceInfo.CUSTOMIZED)
203 try:
204 with open(filename, 'w') as fd:
205 cfg = uit.UnicodeConfig()
206 for name, srv in self.services.items():
207 cfg.add_section(name)
208 for key in srv.keys():
209 items = srv.normalize(key)
210 for item, value in items.items():
211 cfg.set(name, item, value)
213 cfg.write(fd)
215 return True
216 except OSError:
217 return False
219 def read_services(self, filename: str | None = None, package: str | None = None, override: bool = False) -> None:
220 """
221 Read start/stop levels of services.
223 :param filename: Explicit filename for loading.
224 :param package: Explicit package name.
225 :param override: `True` to overwrite already loaded descriptions.
226 :raises AttributeError: if neither `filename` nor `package` are given.
227 """
228 if not filename:
229 if not package:
230 raise AttributeError("neither 'filename' nor 'package' is specified")
231 filename = os.path.join(ServiceInfo.BASE_DIR, ServiceInfo.SERVICES, package + ServiceInfo.FILE_SUFFIX)
232 cfg = uit.UnicodeConfig()
233 cfg.read(filename)
234 for sec in cfg.sections():
235 # service already known?
236 if not override and sec in self.services:
237 continue
238 srv = Service()
239 srv['name'] = sec
240 for name, value in cfg.items(sec):
241 srv[name] = value
242 for path in srv.get('programs', '').split(','):
243 # "programs" defines the "/proc/self/cmdline" of the service,
244 # not the executable, therefore we test for a leading "/":
245 # check if it is a real file split to remove parameters
246 if path.startswith('/') and not os.path.exists(path.split(' ', 1)[0]):
247 break # ==> do not execute else
248 else:
249 self.services[sec] = srv
251 def __load_services(self) -> None:
252 """Load definition of all defined services."""
253 path = os.path.join(ServiceInfo.BASE_DIR, ServiceInfo.SERVICES)
254 for entry in os.listdir(path):
255 # customized service descrptions are read afterwards
256 if entry == ServiceInfo.CUSTOMIZED:
257 continue
258 cfgfile = os.path.join(path, entry)
259 if os.path.isfile(cfgfile) and cfgfile.endswith(ServiceInfo.FILE_SUFFIX):
260 self.read_services(cfgfile)
261 # read modified/added service descriptions
262 self.read_customized()
264 def read_customized(self) -> None:
265 """Read service cusomization."""
266 custom = os.path.join(ServiceInfo.BASE_DIR, ServiceInfo.SERVICES, ServiceInfo.CUSTOMIZED)
267 self.read_services(custom, override=True)
269 def get_services(self) -> Iterable[str]:
270 """
271 Return a list fo service names.
273 :returns: List of service names.
274 """
275 return self.services.keys()
277 def get_service(self, name: str) -> Service | None:
278 """
279 Return the service object associated with the given name.
281 :param name: Service name.
282 :returns: description object or `None`.
283 """
284 return self.services.get(name, None)
286 def add_service(self, name: str, service: Service) -> None:
287 """
288 Add a new service object or overrides an old entry.
290 :param name: Service name.
291 :param service: :py:class:`Service` instance.
292 """
293 if not service.check():
294 self.services[name] = service
297if __name__ == '__main__':
298 import doctest
299 doctest.testmod()