switch to using sudo

Author: Grotax

README.md

@@ -28,7 +28,7 @@ python3 -m venv .venv
 Run the existing scheduled check with mail credentials and a log file:
 
 ```bash
-.venv/bin/python wireguard.py check \
+sudo .venv/bin/python wireguard.py check \
   --user noc@example.org \
   --password 'mail-password' \
   --log /var/log/ffshmon.log
@@ -41,7 +41,7 @@ The command exits after one health cycle. If the FastD service is down, the conn
 Start the long-running monitor with:
 
 ```bash
-.venv/bin/python wireguard.py serve \
+sudo .venv/bin/python wireguard.py serve \
   --user noc@example.org \
   --password 'mail-password' \
   --log /var/log/ffshmon.log
@@ -71,7 +71,7 @@ A value of `1` means the latest check succeeded. A value of `0` means the FastD
 The listener and polling interval can be changed with `--host`, `--port`, and `--interval`:
 
 ```bash
-.venv/bin/python wireguard.py serve \
+sudo .venv/bin/python wireguard.py serve \
   --user noc@example.org \
   --password 'mail-password' \
   --log /var/log/ffshmon.log \

config_manager.py

@@ -1,12 +1,16 @@
+"""Regenerate and restart the WireGuard configuration."""
+
 import subprocess
 import logging
 
 
 def new_config(interface):
+    """Regenerate the WireGuard configuration and restart its service."""
     result = subprocess.run(
-        ["python3", "/opt/wg-conf-gen/wg-conf-gen.py", "recreate"],
+        ["sudo", "python3", "/opt/wg-conf-gen/wg-conf-gen.py", "recreate"],
         capture_output=True,
         text=True,
+        check=False,
     )
     if result.returncode != 0:
         logging.error("Error while recreating wireguard config:")
@@ -18,8 +22,9 @@ def new_config(interface):
     logging.info("Output of wg-conf-gen:")
     logging.info(result.stdout)
     subprocess.run(
-        ["systemctl", "restart", f"wg-quick@{interface}.service"],
+        ["sudo", "systemctl", "restart", f"wg-quick@{interface}.service"],
         stdout=subprocess.DEVNULL,
         stderr=subprocess.DEVNULL,
+        check=False,
     )
     logging.info("Restarted wireguard service")

hard_stop.py

@@ -1,11 +1,13 @@
+"""Stop the FastD and WireGuard services during failed recovery."""
+
 import subprocess
 import logging
 
 
 def stop_fastd(service_name):
     """Stop fastd"""
     subprocess.run(
-        ["systemctl", "stop", f"fastd@{service_name}.service"],
+        ["sudo", "systemctl", "stop", f"fastd@{service_name}.service"],
         stdout=subprocess.DEVNULL,
         stderr=subprocess.DEVNULL,
         check=True,
@@ -15,8 +17,9 @@ def stop_fastd(service_name):
 
 # stop wg tunnel
 def stop_wg(service_name):
+    """Stop the WireGuard service for the configured interface."""
     subprocess.run(
-        ["systemctl", "stop", f"wg-quick@{service_name}.service"],
+        ["sudo", "systemctl", "stop", f"wg-quick@{service_name}.service"],
         stdout=subprocess.DEVNULL,
         stderr=subprocess.DEVNULL,
         check=True,

test_wireguard.py

@@ -20,6 +20,7 @@ def test_interface_probe_uses_requested_interface(self):
             self.assertTrue(wireguard.test_interface("wg-test"))
 
         command = run.call_args.args[0]
+        self.assertEqual(command[0], "sudo")
         self.assertIn("wg-test", command)
 
     def test_metrics_endpoint_exposes_cached_status(self):

wireguard.py

@@ -21,7 +21,14 @@
 def is_service_running(service_name):
     """Return whether the configured FastD service is running."""
     result = subprocess.run(
-        ["systemctl", "show", "-p", "SubState", f"fastd@{service_name}.service"],
+        [
+            "sudo",
+            "systemctl",
+            "show",
+            "-p",
+            "SubState",
+            f"fastd@{service_name}.service",
+        ],
         capture_output=True,
         text=True,
         check=False,
@@ -32,6 +39,7 @@ def is_service_running(service_name):
 def test_interface(interface_name):
     """Returns True if interface is ok, returns False if interface is not ok."""
     curl_cmd = [
+        "sudo",
         "curl",
         "--connect-timeout",
         "10",