150 lines
4.3 KiB
Python
Executable File
150 lines
4.3 KiB
Python
Executable File
#!/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))
|