Attachment #5166 for bug #31030

View | Details | Raw Unified | Return to bug 31030
Collapse All | Expand All




# /usr/share/common-licenses/AGPL-3; if not, see
# <http://www.gnu.org/licenses/>.

import locale
import os
import re
import string
import shlex

import univention.info_tools as uit

class Service(uit.LocalizedDictionary):
	def __init__(self):
		uit.LocalizedDictionary.__init__(self)
		self.start_runlevel = []
		self.stop_runlevel = []
		self.start_code = 0
		self.stop_code = 0
		self.running = False

	def check(self):
		"""Check service entry for validity, returning list of incomplete entries."""
		incomplete = []
		for key in ('description', 'programs'):
			if not self.get(key, None):
				incomplete.append(key)
		return incomplete


def pidof(name):
	"""
	Return list of process IDs matching name.
	>>> import os,sys;str(os.getpid()) in pidof(sys.executable)
	True
	"""
	cmd = shlex.split(name)
	for proc in os.listdir('/proc'):
		cmdline = os.path.join('/proc', proc, 'cmdline')
		try:
			with open(cmdline, 'r') as fd:
				commandline = fd.read()
		except EnvironmentError:
			continue
		fd = open( cmdline )
		cmd = fd.readline()
		# kernel thread
		if not commandline:
			continue
		if '\x00' in commandline:
			args = commandline.split('\x00')
		else:
			args = commandline.split(' ')
		cmd = shlex.split( name )
		if cmd[0] in args:
			if len(args) >= len(cmd) > 1:
				if all(a == c for a, c in zip(args, cmd)):
					yield proc
					if cmd[ i ] != args[ i ]:
						break
				else:
					result.append( file )
			else:
				yield proc

	return result

class ServiceInfo( object ):
	BASE_DIR = '/etc/univention/service.info'

	CUSTOMIZED = '_customized'
	FILE_SUFFIX = '.cfg'

	RUNLEVELS = "0123456S"
	INIT_SCRIPT_REGEX = re.compile('(?P<action>[SK])(?P<code>[0-9]{2})(?P<name>.+)')

	def __init__( self, install_mode = False ):
		self.services = {}

			self.update_services()

	def sysv_infos( self ):
		"""Read start/stop levels of services."""
		for level in ServiceInfo.RUNLEVELS:
			for link in os.listdir('/etc/rc%s.d/' % (level,)):
			for link in os.listdir( '/etc/rc%s.d/' % level ):
				if not os.path.islink( link ):
					continue
				match = ServiceInfo.INIT_SCRIPT_REGEX.match(link)
				if not match:
					continue
				action, code, name = match.groups()
				try:
					service = self.service[name]
				except LookupError:
					continue
				if action == 'S':
					service.start_runlevels.append(level)
					service.start_code = int(code)
				elif action == 'K':
					service.stop_runlevels.append(level)
					service.stop_code = int(code)

	def __update_status(self, name, service):
		for prog in service['programs'].split(','):
			if prog and not pidof(prog.strip()):
				service.running = False
				break
		else:
			service.running = True

	def update_services( self ):
		"""Update the run state of all services."""
		for name, serv in self.services.items():
			self.__update_status(name, serv)

	def check_services( self ):
		"""Return dictionary of incomplete service descriptions."""

		return incomplete

	def write_customized( self ):
		"""Save service cusomization."""
		filename = os.path.join( ServiceInfo.BASE_DIR, ServiceInfo.SERVICES,
								 ServiceInfo.CUSTOMIZED )
		try:
			fd = open(filename, 'w')
		except IOError:
			return False

		cfg = uit.UnicodeConfig()

			filename = os.path.join( ServiceInfo.BASE_DIR, ServiceInfo.SERVICES,
									 package + ServiceInfo.FILE_SUFFIX )
		cfg = uit.UnicodeConfig()
		cfg.read(filename)
		for sec in cfg.sections():
			# service already known?
			if not override and sec in self.services:
				continue
			srv = Service()
			for name, value in cfg.items(sec):
				srv[name] = value
			for path in srv.get('programs', '').split(','):
				# "programs" defines the "/proc/self/cmdline" of the service,
				# not the executable, therefore we test for a leading "/":

				if path.startswith('/') and not os.path.exists(path.split(' ', 1)[0]):
					break # ==> do not execute else
			else:
				self.services[sec] = srv

	def __load_services( self ):
		"""Load definition of all defined services."""
		path = os.path.join( ServiceInfo.BASE_DIR, ServiceInfo.SERVICES )
		for entry in os.listdir( path ):
			# customized service descrptions are read afterwards

		self.read_customized()

	def read_customized( self ):
		"""Read service cusomization."""
		custom = os.path.join( ServiceInfo.BASE_DIR, ServiceInfo.SERVICES,
							   ServiceInfo.CUSTOMIZED )
		self.read_services( custom, override = True )

	def get_service( self, name ):
		'''returns a service object associated with the given name or
		None if it does not exist'''
		return self.services.get( name, None )

	def add_service( self, name, service ):
		'''this methods adds a new service object or overrides an old
		entry'''
		if not service.check():
			self.services[ name ] = service


if __name__ == '__main__':
	import doctest
	doctest.testmod()

Return to bug 31030