117 lines
5.2 KiB
Python
117 lines
5.2 KiB
Python
"""Bounded read-only sd-bus queries over ONE kernel-authenticated PID1 peer."""
|
|
import ctypes as C
|
|
import errno
|
|
import socket
|
|
import struct
|
|
|
|
|
|
class Unavailable(OSError):
|
|
pass
|
|
|
|
|
|
class IdentityError(ValueError):
|
|
pass
|
|
|
|
|
|
class Bus:
|
|
def __init__(self):
|
|
self.lib = C.CDLL('libsystemd.so.0')
|
|
self.bus = C.c_void_p()
|
|
self.lib.sd_bus_new.argtypes = [C.POINTER(C.c_void_p)]
|
|
self.lib.sd_bus_set_fd.argtypes = [C.c_void_p, C.c_int, C.c_int]
|
|
self.lib.sd_bus_set_method_call_timeout.argtypes = [C.c_void_p, C.c_uint64]
|
|
self.lib.sd_bus_start.argtypes = [C.c_void_p]
|
|
self.lib.sd_bus_unref.argtypes = [C.c_void_p]
|
|
self.lib.sd_bus_message_unref.argtypes = [C.c_void_p]
|
|
self.lib.sd_bus_call_method.argtypes = [C.c_void_p, C.c_char_p, C.c_char_p,
|
|
C.c_char_p, C.c_char_p, C.c_void_p, C.POINTER(C.c_void_p), C.c_char_p]
|
|
self.lib.sd_bus_get_property.argtypes = [C.c_void_p, C.c_char_p, C.c_char_p,
|
|
C.c_char_p, C.c_char_p, C.c_void_p, C.POINTER(C.c_void_p), C.c_char_p]
|
|
self.lib.sd_bus_message_read.argtypes = [C.c_void_p, C.c_char_p]
|
|
self.lib.sd_bus_message_read_array.argtypes = [C.c_void_p, C.c_char,
|
|
C.POINTER(C.c_void_p), C.POINTER(C.c_size_t)]
|
|
channel = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
channel.settimeout(2)
|
|
try:
|
|
channel.connect('/run/systemd/private')
|
|
pid, uid, gid = struct.unpack('3i', channel.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12))
|
|
if (pid, uid) != (1, 0):
|
|
raise IdentityError('manager bus peer is not retained PID1')
|
|
channel.setblocking(True)
|
|
self.check(self.lib.sd_bus_new(C.byref(self.bus)))
|
|
fd = channel.fileno()
|
|
self.check(self.lib.sd_bus_set_fd(self.bus, fd, fd))
|
|
channel.detach() # sd-bus now owns this exact authenticated connection.
|
|
self.check(self.lib.sd_bus_set_method_call_timeout(self.bus, 2_000_000))
|
|
self.check(self.lib.sd_bus_start(self.bus))
|
|
except (OSError, Unavailable) as exc:
|
|
self.close()
|
|
raise Unavailable('manager bus unavailable') from exc
|
|
except BaseException:
|
|
self.close()
|
|
raise
|
|
finally:
|
|
channel.close()
|
|
|
|
@staticmethod
|
|
def check(result):
|
|
if result < 0:
|
|
code = -result
|
|
if code in (errno.ECONNRESET, errno.ECONNREFUSED, errno.ENOTCONN,
|
|
errno.EPIPE, errno.ETIMEDOUT, errno.ESHUTDOWN, errno.EHOSTDOWN):
|
|
raise Unavailable(code, 'manager transport unavailable')
|
|
raise IdentityError('manager query refused: errno ' + str(code))
|
|
return result
|
|
|
|
def close(self):
|
|
if self.bus:
|
|
self.lib.sd_bus_unref(self.bus)
|
|
self.bus = C.c_void_p()
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *unused):
|
|
self.close()
|
|
|
|
def unit(self, name):
|
|
reply = C.c_void_p()
|
|
try:
|
|
self.check(self.lib.sd_bus_call_method(self.bus, b'org.freedesktop.systemd1',
|
|
b'/org/freedesktop/systemd1', b'org.freedesktop.systemd1.Manager',
|
|
# An inactive app unit can be garbage-collected. LoadUnit loads
|
|
# its fixed configuration only; it does not enqueue/start a job.
|
|
b'LoadUnit', None, C.byref(reply), b's', C.c_char_p(name.encode())))
|
|
value = C.c_char_p()
|
|
self.check(self.lib.sd_bus_message_read(reply, b'o', C.byref(value)))
|
|
path = value.value
|
|
finally:
|
|
self.lib.sd_bus_message_unref(reply)
|
|
result = {}
|
|
for key in ('Id', 'LoadState', 'ActiveState', 'SubState', 'ControlGroup', 'InvocationID', 'Job', 'MainPID'):
|
|
interface = b'org.freedesktop.systemd1.Service' if key in ('MainPID', 'ControlGroup') else b'org.freedesktop.systemd1.Unit'
|
|
signature = {'InvocationID': b'ay', 'Job': b'(uo)', 'MainPID': b'u'}.get(key, b's')
|
|
reply = C.c_void_p()
|
|
try:
|
|
self.check(self.lib.sd_bus_get_property(self.bus, b'org.freedesktop.systemd1', path,
|
|
interface, key.encode(), None, C.byref(reply), signature))
|
|
if key == 'InvocationID':
|
|
data, size = C.c_void_p(), C.c_size_t()
|
|
self.check(self.lib.sd_bus_message_read_array(reply, b'y', C.byref(data), C.byref(size)))
|
|
result[key] = C.string_at(data, size.value).hex()
|
|
elif key == 'Job':
|
|
number, job = C.c_uint32(), C.c_char_p()
|
|
self.check(self.lib.sd_bus_message_read(reply, b'(uo)', C.byref(number), C.byref(job)))
|
|
result[key] = [number.value, job.value.decode()]
|
|
elif key == 'MainPID':
|
|
number = C.c_uint32()
|
|
self.check(self.lib.sd_bus_message_read(reply, b'u', C.byref(number)))
|
|
result[key] = number.value
|
|
else:
|
|
value = C.c_char_p()
|
|
self.check(self.lib.sd_bus_message_read(reply, b's', C.byref(value)))
|
|
result[key] = value.value.decode()
|
|
finally:
|
|
self.lib.sd_bus_message_unref(reply)
|
|
return result
|