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 Codeberg Repositories Mastodon Über mich home xmpp


Vor 5 Jahren hier im Blog

  • Multi-User-WebDAV, Docker, GitHub

    17.11.2019

    Nachdem ich mich in letzter Zeit verstärkt mit Docker und dem zugehörigen Ökosystem beschäftige, habe ich begonnen, verschiedenste Dienste in Containern zu testen um zu sehen, ob in manchen Fällen LXC oder KVM nicht doch die bessere Wahl wäre...

    Weiterlesen...

Neueste Artikel

  • Migration der Webseite und aller OpenSource Projekte

    In eigener Sache...

    Weiterlesen...
  • AutoHideToolbar für Java Swing

    Ich habe eine neue Java Swing Komponente erstellt: Es handelt sich um einen Wrapper für von JToolBar abgeleitete Klassen, die die Werkzeugleiste minimieren und sie nur dann einblenden, wenn der Mauszeiger über ihnen schwebt.

    Weiterlesen...
  • Integration von EBMap4D in die sQLshell

    Ich habe bereits in einem früheren Artikel über meine ersten Erfolge berichtet, der sQLshell auf Basis des bestehenden Codes aus dem Projekt EBMap4D eine bessere Integration für Geo-Daten zu spendieren und entsprechende Abfragen, bzw. deren Ergebnisse auf einer Kartenansicht zu visualisieren.

    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, muss 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 Codeberg-Repositories findet man hier.