D-Bus-Service mit Python

vorhergehende Artikel in: Python
07.11.2020

Es ist einige Zeit vergangen seit dem ich mich zum letzten Mal zu Python geäußert habe. Während einer Recherche zu einem komplett anderen Thema habe ich einen Artikel gefunden, der beschrieb, wie man einen D-Bus-Service in Python realisiert. Ich habe die Erkenntnisse mit anderen kombiniert und stelle die Resultate hier vor

Zunächst wollte ich einen einfachen Service ausprobieren - der Link dazu wie zu allen anderen Resourcen, die mich meinem Ziel näher gebracht haben ist unten angehängt. Dieser Service war schnell zum funktionieren gebracht. Er gibt auf Anfrage an die Operation CurrentTime die aktuelle Systemzeit zurück.

Interessant war für mich, dass man die Möglichkeit hat, einen solchen Service nicht immer starten zu müssen - trifft man gewisse Vorkehrungen, kann man das D-Bus-System dazu bewegen, dass es den betreffenden Service startet wenn man die erste Anfrage an ihn richtet. Nach meinen Erkenntnissen funktioniert das nur mit Session-Services, nicht mit System-Services. Alles was man tun muss, ist eine Service-Datei zu schreiben - beispielsweise /usr/share/dbus-1/services/test.service

[D-BUS Service]
Name=de.elbosso.Time
Exec=/home/elbosso/time-dbus.py

Liegt die Python-Datei im angegebenen Pfad und ist sie ausführbar, startet sie das D-Bus-System sobald die erste Anfrage für diesen Service eingeht. Normalerweise enden D-Bus-Services nie (bzw. automatisch mit Ende der Session) - also habe ich noch eine weitere Operation hinzugefügt, die den Service beendet. Damit haben wir das Grundgerüst für einen Service, der über D-Bus gestartet wird und auch darüber beendet werden kann:

#!/usr/bin/python3

import dbus import dbus.service import time

class Time(dbus.service.Object): def __init__(self): #self.bus = dbus.SystemBus() self.bus = dbus.SessionBus() name = dbus.service.BusName('de.elbosso.Time', bus=self.bus) super().__init__(name, '/Time')

self.dbus_info = None self.polkit = None

@dbus.service.method('de.elbosso.Time', out_signature='s') def CurrentTime(self): """Use strftime to return a formatted timestamp that looks like 23-02-2018 06:57:04."""

formatter = '%d-%m-%Y %H:%M:%S' return time.strftime(formatter) @dbus.service.method('de.elbosso.Time', out_signature='s', sender_keyword='sender', connection_keyword='conn') def Quit(self, sender=None, conn=None): """End the service."""

loop.quit() return 'I quit!'

if __name__ == '__main__': import dbus.mainloop.glib from gi.repository import GLib

dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)

loop = GLib.MainLoop() object = Time() loop.run()

Es ist möglich, in solchen Services polkit zu nutzen um für sensible Bereiche eine gültige Authentifizierung zu verlangen. Dazu muss man den Service selbst als root starten. Weiterhin muss eine entsprechende Konfiguration für polkit vorliegen - zum Beispiel /usr/share/polkit-1/actions/de.elbosso.Time.policy

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE policyconfig PUBLIC
 "-//freedesktop//DTD PolicyKit Policy Configuration 1.0//EN"
 "http://www.freedesktop.org/standards/PolicyKit/1.0/policyconfig.dtd">
<policyconfig><vendor>Example</vendor>
  <vendor_url>https://example.com/example</vendor_url><action id="de.elbosso.Time.auth">
    <description gettext-domain="systemd">Authorization</description>
    <message gettext-domain="systemd">Authentication is needed to perform this action.</message>
    <defaults>
        <!--These describe the auth level needed to do this.
            Auth_admin, the current one, requires admin authentication every time.
            Auth_admin_keep behaves like sudo, saving the password for a few minutes.Allow_inactive allows it to be accessed from SSH etc. Allow_active allows it to be accessed from the desktop.
            Allow_any is a combo of both.
        -->
      <allow_any>auth_admin</allow_any>
      <allow_inactive>auth_admin</allow_inactive>
      <allow_active>auth_admin</allow_active>
    </defaults>
  </action>
</policyconfig>

Weiterhin benötigt ein D-Bus-Service, der von root gestartet werden soll eine Konfiguration - etwa /etc/dbus-1/system.d/de.elbosso.Time.conf

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE busconfig PUBLIC "-//freedesktop//DTD D-BUS Bus Configuration 1.0//EN"
"http://www.freedesktop.org/standards/dbus/1.0/busconfig.dtd">
<busconfig>
  <type>system</type>
  <!-- Only root can own the service -->
  <policy user="root">
    <allow own="de.elbosso.Time"/>
    <allow send_destination="de.elbosso.Time"/>
    <allow send_interface="de.elbosso.Time"/>
  </policy><!-- Allow anyone to invoke methods on the interfaces -->
  <policy context="default">
    <allow send_destination="de.elbosso.Time"/>
    <allow send_interface="de.elbosso.Time"/>
  </policy>
</busconfig>

Der Python-Code muss jetzt noch ein wenig angepasst werden - zunächst muss der System- und nicht der Session-Bus genutzt werden und dann muss natürlich der Service auch für mindestens eine Operation auch nach der Autorisierung fragen. Ich habe mich dafür entschieden, dass das Beenden des Service über eine D-Bus-Operation zwar weiterhin verfügbar ist, jedoch nur nach Autorisierung auch ausgeführt wird:

#!/usr/bin/python3

import dbus import dbus.service import time

class Time(dbus.service.Object): def __init__(self): #self.bus = dbus.SystemBus() self.bus = dbus.SessionBus() name = dbus.service.BusName('de.elbosso.Time', bus=self.bus) super().__init__(name, '/Time')

self.dbus_info = None self.polkit = None

@dbus.service.method('de.elbosso.Time', out_signature='s') def CurrentTime(self): """Use strftime to return a formatted timestamp that looks like 23-02-2018 06:57:04.""" formatter = '%d-%m-%Y %H:%M:%S' return time.strftime(formatter) @dbus.service.method('de.elbosso.Time', out_signature='s', sender_keyword='sender', connection_keyword='conn') def Quit(self, sender=None, conn=None): """End the service."""

#self._check_polkit_privilege(sender, conn, 'de.elbosso.Time.auth') loop.quit() return 'I quit!'

def _check_polkit_privilege(self, sender, conn, privilege): # Get Peer PID if self.dbus_info is None: # Get DBus Interface and get info thru that self.dbus_info = dbus.Interface(conn.get_object("org.freedesktop.DBus", "/org/freedesktop/DBus/Bus", False), "org.freedesktop.DBus") pid = self.dbus_info.GetConnectionUnixProcessID(sender)

# Query polkit if self.polkit is None: self.polkit = dbus.Interface(dbus.SystemBus().get_object( "org.freedesktop.PolicyKit1", "/org/freedesktop/PolicyKit1/Authority", False), "org.freedesktop.PolicyKit1.Authority")

# Check auth against polkit; if it times out, try again try: auth_response = self.polkit.CheckAuthorization( ("unix-process", {"pid": dbus.UInt32(pid, variant_level=1), "start-time": dbus.UInt64(0, variant_level=1)}), privilege, {"AllowUserInteraction": "true"}, dbus.UInt32(1), "", timeout=600) print(auth_response) (is_auth, _, details) = auth_response except dbus.DBusException as e: if e._dbus_error_name == "org.freedesktop.DBus.Error.ServiceUnknown": # polkitd timeout, retry self.polkit = None return self._check_polkit_privilege(sender, conn, privilege) else: # it's another error, propagate it raise

if not is_auth: # Aww, not authorized :( print(":(") return False

print("Successful authorization!") return True

if __name__ == '__main__': import dbus.mainloop.glib from gi.repository import GLib

dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)

loop = GLib.MainLoop() object = Time() loop.run()

Testen kann man den Service jeweils mittels folgender Kommandos beim Start über den Session-Bus:

dbus-send --session --print-reply --dest="de.elbosso.Time" /Time de.elbosso.Time.CurrentTime
dbus-send --session --print-reply --dest="de.elbosso.Time" /Time de.elbosso.Time.Quit

Und beim Start über den System-Bus:

dbus-send --system --print-reply --dest="de.elbosso.Time" /Time de.elbosso.Time.CurrentTime
dbus-send --system --print-reply --dest="de.elbosso.Time" /Time de.elbosso.Time.Quit

Alle Artikel rss Wochenübersicht Monatsübersicht Github Repositories Gitlab Repositories Mastodon Über mich home xmpp


Vor 5 Jahren hier im Blog

  • Alte Rechner zukunftsfähig (?)

    01.05.2019

    Ich mache immer wieder gerne Experimente mit Terminalsessions - meistens wegen der zentral möglichen Administration. Nunhabe ich das Thema von neuem angepackt: wegen einem alten(sehr alten) Laptop, der bei mir schon so lange im Schrank lag, dass sogar die BIOS-Batterie inzwischen leer war...

    Weiterlesen...

Neueste Artikel

  • Graphics2D Implementierung für Java mit verlegtem Koordinatenursprung

    Es gibt seit vielen Jahren immer mal wieder Leute, die im Internet fragen, ob man in Javas diversen Methoden zum Zeichnen von Graphiken das Koordinatensystem so ändern könnte, dass sich der Koordinatenursprung links unten befindet und die positive y-Achse nach oben weist. Meist sind die Antworten dann, dass eine Affine Transformation eingeschaltet werden solle, die das Bild spiegelt.

    Weiterlesen...
  • Unerwartete Probleme bei der Software Raid5 Erweiterung

    Ich bin an die Grenzen meiner Storagelösung gestoßen - es musste mehr Platz her...

    Weiterlesen...
  • Die sQLshell ist nun cloudnative!

    Die sQLshell hat eine weitere Integration erfahren - obwohl ich eigentlich selber nicht viel dazu tun musste: Es existiert ein Projekt/Produkt namens steampipe, dessen Slogan ist select * from cloud; - Im Prinzip eine Wrapperschicht um diverse (laut Eigenwerbung mehr als 140) (cloud) data sources.

    Weiterlesen...

Manche nennen es Blog, manche Web-Seite - ich schreibe hier hin und wieder über meine Erlebnisse, Rückschläge und Erleuchtungen bei meinen Hobbies.

Wer daran teilhaben und eventuell sogar davon profitieren möchte, muß damit leben, daß ich hin und wieder kleine Ausflüge in Bereiche mache, die nichts mit IT, Administration oder Softwareentwicklung zu tun haben.

Ich wünsche allen Lesern viel Spaß und hin und wieder einen kleinen AHA!-Effekt...

PS: Meine öffentlichen GitHub-Repositories findet man hier - meine öffentlichen GitLab-Repositories finden sich dagegen hier.