network-manager-wireguard-upstream/examples/dbus/dbus.py

#!/bin/env python

import subprocess

# three possibilities for DBUS in python:
# * dbus-python (it's legacy and when i tried it, it didn't really work -- missing SessionBus(), ...)
# * pydbus      (seems pretty neat)
# * txdbus      (haven't tried it; apparently it's a native python dbus implementation)
#
#
# official python doc on dbus
# https://wiki.python.org/moin/DbusExamples
#
# pydbus tutorial
# https://github.com/LEW21/pydbus/blob/master/doc/tutorial.rst
from pydbus import SessionBus, SystemBus

# required for the mainloop
from gi.repository import GLib

(rows, cols) = subprocess.check_output(["stty", "size"]).split()
show_introspect = True


def send_desktop_notification(title="Hello World", msg="pydbus works!"):
    """Send a notification to the desktop environment to display"""

    bus = SessionBus()
    notifications = bus.get('.Notifications')
    notifications.Notify('test', 0, 'dialog-information', title, msg, [], {}, 5000)

def list_systemd_units():
    """Fetch all systemd units and print them"""

    bus = SystemBus()

    # systemd is now a proxy for the .systemd1 remote object
    systemd = bus.get(".systemd1")

    for unit in systemd.ListUnits():
        print("Unit:\n")
        print(unit)
        print("-" * int(cols))

def stop_start_systemd_unit(name="ssh.service"):
    """Stop and restart a systemd unit"""

    bus = SystemBus()

    # systemd is now a proxy for the .systemd1 remote object
    systemd = bus.get(".systemd1")
    job1 = systemd.StopUnit(name, "fail")
    job2 = systemd.StartUnit(name, "fail")

def watch_for_new_systemd_jobs():
    """Wait for new systemd units and when they are created, print them out"""

    bus = SystemBus()

    # systemd is now a proxy for the .systemd1 remote object
    systemd = bus.get(".systemd1")
    systemd.JobNew.connect(print)
    GLib.MainLoop().run()

    # or
    #
    # systemd.onJobNew = print
    # GLib.MainLoop.run()

def hibernate():
    try:
        bus = SessionBus()
        power = bus.get('org.gnome.PowerManager', '/org/gnome/PowerManager')

        if power.CanHibernate():
            answer = input("Do you want to hibernate? [Y/n]:")
            if answer.lower() == "y":
                power.Hibernate()

        else:
            print("Cannot hibernate")

    except Exception as ex:
        print("Could not get PowerManager from DBUS")
        print(str(ex))


def get_wg_plugin(bus_name="org.freedesktop.NetworkManager.wireguard",
                  object_path="/org/freedesktop/NetworkManager/VPN/Plugin"):
    """Retrieve the WireGuard VPN plugin from the System Bus.

    Arguments:
    bus_name -- the bus name of the object to import
    object_path -- the object path of hte object (= where to find the interface)
    """

    # since our wireguard plugin implements the VPN plugin and does not export
    # an interface on its own, we need to use the VPN plugin interfce
    bus = SystemBus()
    wg = bus.get(bus_name, object_path)
    return wg


def wg_disconnect(wg_plugin):
    """Disconnect the WG VPN plugin"""

    wg_plugin.Disconnect()

def wg_connect(wg_plugin):
    """Send the Connect Command to the WireGuard Plugin"""

    # these are the settings that are expected by Connect(a{sa{sv}}) for a VPN plugin
    service_type = GLib.Variant("s", "service")
    user_name = GLib.Variant("s", "wireguard")
    persistent = GLib.Variant("b", False)
    data = GLib.Variant("a{ss}", {"maxi": "cool"})
    secrets = GLib.Variant("a{ss}", {"name": "maxi moser"})
    timeout = GLib.Variant("u", 1337)

    # The DBus type: a{sa{sv}}
    # is a Dictionary with...
    # Key: Type ("wireless", "wired", "vpn", ...) -- we want VPN
    # Value: Dictionary with Key: Setting Name, Value: Setting Value
    settings = {"vpn":
                    {"service-type": service_type,
                        "user-name": user_name,
                        "persistent": persistent,
                        "data": data,
                        "secrets": secrets,
                        "timeout": timeout}
                }
    wg_plugin.Connect(settings)


show_introspect = False

if __name__ == "__main__":
    # send_desktop_notification("Guten Tag", "pydbus funktioniert, mein Herr!")
    try:
        wg = get_wg_plugin()
        
        if show_introspect:
            print(wg.Introspect())
            help(wg)

        wg_connect(wg)

    except Exception as ex:
        print(str(ex))