From 8def7548bd85ec26120230d58686fe1782fd8971 Mon Sep 17 00:00:00 2001 From: Matthias Lanter Date: Thu, 4 Sep 2025 16:21:25 +0200 Subject: [PATCH 1/4] Add basic WireGuard functions to connect and disconnect tunnel based on WireGuard configurations --- NetworkMgr/trayicon.py | 39 +++++++++++++++++++++++ NetworkMgr/wg_api.py | 72 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 111 insertions(+) create mode 100644 NetworkMgr/wg_api.py diff --git a/NetworkMgr/trayicon.py b/NetworkMgr/trayicon.py index 76fcc51..9455c5f 100755 --- a/NetworkMgr/trayicon.py +++ b/NetworkMgr/trayicon.py @@ -25,6 +25,12 @@ ) from NetworkMgr.configuration import network_card_configuration +from NetworkMgr.wg_api import ( + wireguardDictionary, + disableWG, + enableWG, + wg_status +) gettext.bindtextdomain('networkmgr', '/usr/local/share/locale') gettext.textdomain('networkmgr') @@ -64,6 +70,30 @@ def icon_clicked(self, status_icon, button, time): def nm_menu(self): self.menu = Gtk.Menu() + wg_title = Gtk.MenuItem() + wg_title.set_label(_("WireGuard VPN")) + wg_title.set_sensitive(False) + self.menu.append(wg_title) + self.menu.append(Gtk.SeparatorMenuItem()) + wgDevices = self.wginfo['configs'] + for wgDev in wgDevices: + connection_state = wgDevices[wgDev]['state'] + connection_info = wgDevices[wgDev]['info'] + if connection_state == "Connected": + wg_item = Gtk.MenuItem(_("%s Connected") % connection_info) + wg_item.set_sensitive(False) + self.menu.append(wg_item) + disconnectwg_item = Gtk.ImageMenuItem(_(f"Disable {wgDev}")) + disconnectwg_item.connect("activate", self.disconnectWG, wgDev) + self.menu.append(disconnectwg_item) + else: + notonlinewg = Gtk.MenuItem(_("%s Disconnected") % connection_info) + notonlinewg.set_sensitive(False) + self.menu.append(notonlinewg) + wiredwg_item = Gtk.MenuItem(_("Enable")) + wiredwg_item.connect("activate", self.connectWG, wgDev) + self.menu.append(wiredwg_item) + self.menu.append(Gtk.SeparatorMenuItem()) e_title = Gtk.MenuItem() e_title.set_label(_("Ethernet Network")) e_title.set_sensitive(False) @@ -220,6 +250,14 @@ def disconnectcard(self, widget, netcard): stopnetworkcard(netcard) self.updateinfo() + def connectWG(self, widget, wginfo): + enableWG(wginfo) + self.updateinfo() + + def disconnectWG(self, widget, wginfo): + disableWG(wginfo) + self.updateinfo() + def closeNetwork(self, widget): stopallnetwork() self.updateinfo() @@ -258,6 +296,7 @@ def updateinfo(self): defaultcard = self.cardinfo['default'] default_type = self.network_type(defaultcard) GLib.idle_add(self.updatetray, defaultcard, default_type) + self.wginfo = wireguardDictionary() self.if_running = False def updatetray(self, defaultdev, default_type): diff --git a/NetworkMgr/wg_api.py b/NetworkMgr/wg_api.py new file mode 100644 index 0000000..c921cac --- /dev/null +++ b/NetworkMgr/wg_api.py @@ -0,0 +1,72 @@ +#!/usr/local/bin/python3.11 + +from platform import system +from subprocess import Popen, PIPE, run, check_output, os + +prefix = '/usr/local' if system() == 'FreeBSD' else sys.prefix +wgconfigpath = prefix + '/etc/wireguard/' + +def wg_service_state(): + run('service wireguard status', shell=True) + +def wireguardDictionary(): + + maindictionary = { + 'service': wg_service_state(), + 'default': '', + } + configs = {} + + if os.path.exists(wgconfigpath): + wgconfigs = sorted(os.listdir(wgconfigpath)) + for wgconfig in wgconfigs: + content = open(wgconfigpath + wgconfig) + for line in content: + if "# Name = " in line: + wgName = line.split('=')[1].strip() + wgDevice = wgconfig.replace('.conf', '') + wgState = wg_status(wgDevice) + seconddictionary = { 'state': wgState, 'info': wgName } + configs[wgDevice] = seconddictionary + + maindictionary['configs'] = configs + + return maindictionary + + +# def stopWG(): +# run('service wireguard stop', shell=True) + + +# def startWG(): +# run('service wireguard start', shell=True) + + +def disableWG(wgconfig): + run(f'wg-quick down {wgconfig}', shell=True) + + +def enableWG(wgconfig): + run(f'wg-quick up {wgconfig}', shell=True) + + +def wg_status(wgconfig): + # out = Popen( + # f'wg show {wgconfig}', + # shell=True, stdout=PIPE, stderr=PIPE, + # universal_newlines=True + # ) + result=run(['wg', 'show', wgconfig], stdout=PIPE, stderr=PIPE) + out = result.stdout.decode('utf-8') + error = result.stderr.decode('utf-8') +# print (out) +# print (error) + if len(out) == 0 and 'Unable to access interface: Device not configured' in error: + status = "Disconnected" + else: + status = "Connected" + + return status + + + From 15a017ab2177611e1d6e447b0ba7b69281330f28 Mon Sep 17 00:00:00 2001 From: Eric Turgeon <4249848+ericbsd@users.noreply.github.com> Date: Sat, 6 Sep 2025 11:14:29 -0300 Subject: [PATCH 2/4] Use f-string instead of string concatenation Co-authored-by: sourcery-ai[bot] <58596630+sourcery-ai[bot]@users.noreply.github.com> --- NetworkMgr/wg_api.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/NetworkMgr/wg_api.py b/NetworkMgr/wg_api.py index c921cac..e5ee5e9 100644 --- a/NetworkMgr/wg_api.py +++ b/NetworkMgr/wg_api.py @@ -4,7 +4,7 @@ from subprocess import Popen, PIPE, run, check_output, os prefix = '/usr/local' if system() == 'FreeBSD' else sys.prefix -wgconfigpath = prefix + '/etc/wireguard/' +wgconfigpath = f'{prefix}/etc/wireguard/' def wg_service_state(): run('service wireguard status', shell=True) From 286500e7460f0d42ba4749f68ef7dce2d5fe1817 Mon Sep 17 00:00:00 2001 From: Matthias Lanter Date: Fri, 19 Sep 2025 17:44:16 +0200 Subject: [PATCH 3/4] Revised version, checked with Pylint Wireguard menu entries only if configuration exists and Wireguard service is stopped --- NetworkMgr/trayicon.py | 44 +++++++++++++------------ NetworkMgr/wg_api.py | 75 ++++++++++++++++++++---------------------- 2 files changed, 59 insertions(+), 60 deletions(-) diff --git a/NetworkMgr/trayicon.py b/NetworkMgr/trayicon.py index 9455c5f..b0690f2 100755 --- a/NetworkMgr/trayicon.py +++ b/NetworkMgr/trayicon.py @@ -26,9 +26,9 @@ from NetworkMgr.configuration import network_card_configuration from NetworkMgr.wg_api import ( - wireguardDictionary, - disableWG, - enableWG, + wg_dictionary, + disable_wg, + enable_wg, wg_status ) @@ -70,30 +70,32 @@ def icon_clicked(self, status_icon, button, time): def nm_menu(self): self.menu = Gtk.Menu() - wg_title = Gtk.MenuItem() - wg_title.set_label(_("WireGuard VPN")) - wg_title.set_sensitive(False) - self.menu.append(wg_title) - self.menu.append(Gtk.SeparatorMenuItem()) - wgDevices = self.wginfo['configs'] - for wgDev in wgDevices: - connection_state = wgDevices[wgDev]['state'] - connection_info = wgDevices[wgDev]['info'] - if connection_state == "Connected": + if len(self.wginfo['configs'])> 0 and self.wginfo['service'] == '"NO"': + wg_title = Gtk.MenuItem() + wg_title.set_label(_("WireGuard VPN")) + wg_title.set_sensitive(False) + self.menu.append(wg_title) + self.menu.append(Gtk.SeparatorMenuItem()) + wg_devices = self.wginfo['configs'] + for wg_dev in wg_devices: + connection_state = wg_devices[wg_dev]['state'] + connection_info = wg_devices[wg_dev]['info'] + if connection_state == "Connected": wg_item = Gtk.MenuItem(_("%s Connected") % connection_info) wg_item.set_sensitive(False) self.menu.append(wg_item) - disconnectwg_item = Gtk.ImageMenuItem(_(f"Disable {wgDev}")) - disconnectwg_item.connect("activate", self.disconnectWG, wgDev) + disconnectwg_item = Gtk.ImageMenuItem(_(f"Disable {wg_dev}")) + disconnectwg_item.connect("activate", self.disconnectWG, wg_dev) self.menu.append(disconnectwg_item) - else: + else: notonlinewg = Gtk.MenuItem(_("%s Disconnected") % connection_info) notonlinewg.set_sensitive(False) self.menu.append(notonlinewg) wiredwg_item = Gtk.MenuItem(_("Enable")) - wiredwg_item.connect("activate", self.connectWG, wgDev) + wiredwg_item.connect("activate", self.connectWG, wg_dev) self.menu.append(wiredwg_item) - self.menu.append(Gtk.SeparatorMenuItem()) + self.menu.append(Gtk.SeparatorMenuItem()) + e_title = Gtk.MenuItem() e_title.set_label(_("Ethernet Network")) e_title.set_sensitive(False) @@ -251,11 +253,11 @@ def disconnectcard(self, widget, netcard): self.updateinfo() def connectWG(self, widget, wginfo): - enableWG(wginfo) + enable_wg(wginfo) self.updateinfo() def disconnectWG(self, widget, wginfo): - disableWG(wginfo) + disable_wg(wginfo) self.updateinfo() def closeNetwork(self, widget): @@ -296,7 +298,7 @@ def updateinfo(self): defaultcard = self.cardinfo['default'] default_type = self.network_type(defaultcard) GLib.idle_add(self.updatetray, defaultcard, default_type) - self.wginfo = wireguardDictionary() + self.wginfo = wg_dictionary() self.if_running = False def updatetray(self, defaultdev, default_type): diff --git a/NetworkMgr/wg_api.py b/NetworkMgr/wg_api.py index e5ee5e9..47091c8 100644 --- a/NetworkMgr/wg_api.py +++ b/NetworkMgr/wg_api.py @@ -1,72 +1,69 @@ #!/usr/local/bin/python3.11 from platform import system -from subprocess import Popen, PIPE, run, check_output, os +from subprocess import PIPE, run, os -prefix = '/usr/local' if system() == 'FreeBSD' else sys.prefix -wgconfigpath = f'{prefix}/etc/wireguard/' +PREFIX = '/usr/local' if system() == 'FreeBSD' else sys.prefix +WG_CONFIG_PATH = f'{PREFIX}/etc/wireguard/' def wg_service_state(): - run('service wireguard status', shell=True) + """Function returns the WireGuard service status.""" + result = run(['service', 'wireguard', 'rcvar'], stdout=PIPE, stderr=PIPE, check=False) + out = result.stdout.decode('utf-8') + + state = 'Unknown' -def wireguardDictionary(): + for line in out.splitlines(): + if "wireguard_enable=" in line: + state = line.split('=')[1].strip() + break + return state + +def wg_dictionary(): + """Function returns the WireGuard configurations.""" maindictionary = { 'service': wg_service_state(), 'default': '', } configs = {} - if os.path.exists(wgconfigpath): - wgconfigs = sorted(os.listdir(wgconfigpath)) + if os.path.exists(WG_CONFIG_PATH): + wgconfigs = sorted(os.listdir(WG_CONFIG_PATH)) for wgconfig in wgconfigs: - content = open(wgconfigpath + wgconfig) + content = open(WG_CONFIG_PATH + wgconfig, encoding="utf-8") + wg_device = wgconfig.replace('.conf', '') + wg_state = wg_status(wg_device) + wg_name = wg_device for line in content: if "# Name = " in line: - wgName = line.split('=')[1].strip() - wgDevice = wgconfig.replace('.conf', '') - wgState = wg_status(wgDevice) - seconddictionary = { 'state': wgState, 'info': wgName } - configs[wgDevice] = seconddictionary + wg_name = line.split('=')[1].strip() + break + + seconddictionary = { 'state': wg_state, 'info': wg_name } + configs[wg_device] = seconddictionary maindictionary['configs'] = configs return maindictionary +def disable_wg(wgconfig): + """Function disable the specified WireGuard configuration (device).""" + run(f'wg-quick down {wgconfig}', shell=True, check=False) -# def stopWG(): -# run('service wireguard stop', shell=True) - - -# def startWG(): -# run('service wireguard start', shell=True) - - -def disableWG(wgconfig): - run(f'wg-quick down {wgconfig}', shell=True) - - -def enableWG(wgconfig): - run(f'wg-quick up {wgconfig}', shell=True) - +def enable_wg(wgconfig): + """Function enable the specified WireGuard configuration (device).""" + run(f'wg-quick up {wgconfig}', shell=True, check=False) def wg_status(wgconfig): - # out = Popen( - # f'wg show {wgconfig}', - # shell=True, stdout=PIPE, stderr=PIPE, - # universal_newlines=True - # ) - result=run(['wg', 'show', wgconfig], stdout=PIPE, stderr=PIPE) + """Function returning the WireGuard configuration (device) is connected or not.""" + result = run(['wg', 'show', wgconfig], stdout=PIPE, stderr=PIPE, check=False) out = result.stdout.decode('utf-8') error = result.stderr.decode('utf-8') -# print (out) -# print (error) + if len(out) == 0 and 'Unable to access interface: Device not configured' in error: status = "Disconnected" else: status = "Connected" return status - - - From ec48633d0793c03e2c949c3dee7f461094a539b8 Mon Sep 17 00:00:00 2001 From: Eric Turgeon <4249848+ericbsd@users.noreply.github.com> Date: Mon, 22 Sep 2025 16:44:13 -0300 Subject: [PATCH 4/4] suggestion (bug_risk): File handle 'content' is not closed after reading. Co-authored-by: sourcery-ai[bot] <58596630+sourcery-ai[bot]@users.noreply.github.com> --- NetworkMgr/wg_api.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/NetworkMgr/wg_api.py b/NetworkMgr/wg_api.py index 47091c8..ea052cf 100644 --- a/NetworkMgr/wg_api.py +++ b/NetworkMgr/wg_api.py @@ -31,14 +31,14 @@ def wg_dictionary(): if os.path.exists(WG_CONFIG_PATH): wgconfigs = sorted(os.listdir(WG_CONFIG_PATH)) for wgconfig in wgconfigs: - content = open(WG_CONFIG_PATH + wgconfig, encoding="utf-8") wg_device = wgconfig.replace('.conf', '') wg_state = wg_status(wg_device) wg_name = wg_device - for line in content: - if "# Name = " in line: - wg_name = line.split('=')[1].strip() - break + with open(WG_CONFIG_PATH + wgconfig, encoding="utf-8") as content: + for line in content: + if "# Name = " in line: + wg_name = line.split('=')[1].strip() + break seconddictionary = { 'state': wg_state, 'info': wg_name } configs[wg_device] = seconddictionary