/usr/share/pyshared/JobService/backends/sysv.py is in jobservice 0.8.0-0ubuntu4.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 | # This file is part of jobservice.
# Copyright 2010 Jacob Peddicord <jpeddicord@ubuntu.com>
#
# jobservice is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# jobservice is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with jobservice. If not, see <http://www.gnu.org/licenses/>.
import os
from stat import ST_MODE, S_ISLNK, S_IXUSR
from subprocess import Popen, PIPE, check_call, CalledProcessError
from dbus import Array
from JobService import DBUS_IFACE, JobException
from JobService.backends import ServiceBase
class SysVException(JobException):
_dbus_error_name = DBUS_IFACE + '.SysVException'
class ServiceBackend(ServiceBase):
def __init__(self):
self.runlevels = self._get_runlevel_info()
self.current = self._get_current_runlevel()
def get_all_services(self):
svclist = []
for root, dirs, files in os.walk('/etc/init.d/'):
for svc in files:
path = os.path.join(root, svc)
mode = os.lstat(path).st_mode
# we only want regular, executable files
if not S_ISLNK(mode) and bool(mode & S_IXUSR):
# ignore files not linked in rc.d
if svc in self.runlevels:
svclist.append(svc)
break
return svclist
def get_service(self, name):
info = {
'name': name,
'description': '',
'version': '',
'author': '',
'running': False,
'automatic': False,
'pid': 0,
'starton': Array(signature='s'),
'stopon': Array(signature='s'),
'file': '',
}
# check the file for info
props = self._get_lsb_properties(name)
info['file'] = props['file']
if 'Short-Description' in props:
info['description'] = props['Short-Description']
# look through runlevel information
if name in self.runlevels:
for rlvl, start in self.runlevels[name].iteritems():
if start[0] == True:
info['starton'].append(rlvl)
else:
info['stopon'].append(rlvl)
if rlvl == self.current:
info['automatic'] = start[0]
p = Popen(['/etc/init.d/' + name, 'status'], stdout=PIPE, stderr=PIPE)
p.communicate() # eat stdout/stdin
info['running'] = (p.returncode == 0)
return info
def start_service(self, name):
try:
check_call(['/etc/init.d/' + name, 'start'])
except CalledProcessError, e:
raise SysVException('Start failed: code {0}'.format(e.returncode))
try:
check_call(['/etc/init.d/' + name, 'status'], stdout=PIPE, stderr=PIPE)
except CalledProcessError, e:
raise SysVException('Service stopped running unexpectedly.')
def stop_service(self, name):
try:
check_call(['/etc/init.d/' + name, 'stop'])
except CalledProcessError, e:
raise SysVException('Stop failed: code {0}'.format(e.returncode))
def set_service_automatic(self, name, auto):
self._remove_rc(name, self.current)
self._link_rc(name, self.current, auto)
self.runlevels[name][self.current] = (auto,
self.runlevels[name][self.current][1])
def get_service_settings(self, name, lang):
settings = []
if not name in self.runlevels:
return settings
for rlvl in sorted(self.runlevels[name].keys()):
if rlvl == '0' or rlvl == '6' or rlvl == 'S':
# skip 0 (shutdown), 6 (restart), or S (boot once)
continue
if rlvl == '1':
label = "Active in recovery mode" #XXX: i18n
elif rlvl == self.current:
label = "Active in current runlevel ({runlevel})".format(runlevel=rlvl) #XXX: i18n
else:
label = "Active on runlevel {runlevel}".format(runlevel=rlvl) #XXX: i18n
settings.append(('runlevel_{0}'.format(rlvl), 'bool', label,
'true' if self.runlevels[name][rlvl][0] else 'false',
(('true', ''), ('false', '')), {}
))
if settings:
settings.insert(0, ('lbl_runlevels', 'label',
"<b>Runlevels</b>", '', (), {})) #XXX: i18n
return settings
def set_service_settings(self, name, newsettings):
for sname, sval in newsettings.iteritems():
if sname.find('runlevel_') == 0:
rlvl = sname[-1:]
auto = (sval == 'true')
self._remove_rc(name, rlvl)
self._link_rc(name, rlvl, auto)
self.runlevels[name][rlvl] = (auto,
self.runlevels[name][rlvl][1])
def _get_runlevel_info(self):
"""Parse /etc/rc?.d and store symlink information.
Returns a dictionary with service names as keys, and a dict of
runlevel: (bool start, int priority) pairs with found information.
"""
svcs = {}
for runlevel in ('0', '1', '2', '3', '4', '5', '6', '7', 'S'):
for root, dirs, files in os.walk('/etc/rc{0}.d'.format(runlevel)):
for svc in files:
path = os.path.join(root, svc)
# exec only
if not bool(os.lstat(path).st_mode & S_IXUSR):
continue
start = (svc[:1] == 'S')
pri = int(svc[1:3])
name = svc[3:]
if not name in svcs:
svcs[name] = {}
svcs[name][runlevel] = (start, pri)
break
return svcs
def _remove_rc(self, name, rlvl):
"""Unlink a service from an rc#.d directory"""
pri = str(self.runlevels[name][rlvl][1])
mode = 'S' if self.runlevels[name][rlvl][0] else 'K'
os.unlink('/etc/rc{0}.d/{1}{2}{3}'.format(rlvl, mode, pri, name))
def _link_rc(self, name, rlvl, start):
"""Re-link an init script to the proper rc#.d location"""
pri = str(self.runlevels[name][rlvl][1])
mode = 'S' if start else 'K'
os.symlink('/etc/init.d/' + name,
'/etc/rc{0}.d/{1}{2}{3}'.format(rlvl, mode, pri, name))
def _get_current_runlevel(self):
out = Popen(['/sbin/runlevel'], stdout=PIPE).communicate()[0]
return out.split()[1]
def _get_lsb_properties(self, name):
"""
Scan a service's init.d entry for LSB information about it.
Returns a dictionary of the entries provided.
"""
props = {'file': '/etc/init.d/' + name}
try:
entry = open(props['file'])
except IOError: return props
parsing = False
for line in entry:
if not parsing:
if '### BEGIN INIT INFO' in line:
parsing = True
continue
if '### END INIT INFO' in line:
break
try:
key, value = line[2:].split(': ')
except: continue
props[key] = value.strip()
entry.close()
return props
|