Sanitize user-provided inputs

api/modules/client.py

@@ -2,27 +2,37 @@ import subprocess
 import json
 
 def get_active_clients_amount(interface):
-    output =  subprocess.run(f'''cat '/var/lib/misc/dnsmasq.leases' | grep -iwE "$(arp -i '{interface}' | grep -oE "(([0-9]|[a-f]|[A-F]){{{2}}}:){{{5}}}([0-9]|[a-f]|[A-F]){{{2}}}")"''', shell=True, capture_output=True, text=True)
-    return(len(output.stdout.splitlines()))
+    arp_output = subprocess.run(['arp', '-i', interface], capture_output=True, text=True)
+    mac_addresses = arp_output.stdout.splitlines()
+
+    if mac_addresses:
+        grep_pattern = '|'.join(mac_addresses)
+        output = subprocess.run(['grep', '-iwE', grep_pattern, '/var/lib/misc/dnsmasq.leases'], capture_output=True, text=True)
+        return len(output.stdout.splitlines())
+    else:
+        return 0
 
 def get_active_clients(interface):
-    #does not run like intended, but it works....
-    output =  subprocess.run(f'''cat '/var/lib/misc/dnsmasq.leases' | grep -iwE "$(arp -i '{interface}' | grep -oE "(([0-9]|[a-f]|[A-F]){{{2}}}:){{{5}}}([0-9]|[a-f]|[A-F]){{{2}}}")"''', shell=True, capture_output=True, text=True)
-    clients_list = []
+    arp_output = subprocess.run(['arp', '-i', interface], capture_output=True, text=True)
+    arp_mac_addresses = set(line.split()[2] for line in arp_output.stdout.splitlines()[1:])
 
-    for line in output.stdout.splitlines():
+    dnsmasq_output = subprocess.run(['cat', '/var/lib/misc/dnsmasq.leases'], capture_output=True, text=True)
+    active_clients = []
+
+    for line in dnsmasq_output.stdout.splitlines():
         fields = line.split()
+        mac_address = fields[1]
 
-        client_data = {
-            "timestamp": int(fields[0]),
-            "mac_address": fields[1],
-            "ip_address": fields[2],
-            "hostname": fields[3],
-            "client_id": fields[4],
-        }
+        if mac_address in arp_mac_addresses:
+            client_data = {
+                "timestamp": int(fields[0]),
+                "mac_address": fields[1],
+                "ip_address": fields[2],
+                "hostname": fields[3],
+                "client_id": fields[4],
+            }
+            active_clients.append(client_data)
 
-        clients_list.append(client_data)
+    json_output = json.dumps(active_clients, indent=2)
+    return json_output
 
-    json_output = json.dumps(clients_list, indent=2)
-
-    return json_output

api/modules/openvpn.py

@@ -34,8 +34,8 @@ def client_login_active():
     return(active_config[1])
 
 def client_config_list(client_config):
-    output = subprocess.run(f"cat /etc/openvpn/client/{client_config}", shell=True, capture_output=True, text=True).stdout.strip()
+    output = subprocess.run(["cat", f"/etc/openvpn/client/{client_config}"], capture_output=True, text=True).stdout.strip()
     return output.split('\n')
 
 #TODO: where is the logfile??
-#TODO: is service connected?
+#TODO: is service connected?

api/modules/wireguard.py

@@ -19,8 +19,13 @@ def client_config_active():
     return(active_config[1])
 
 def client_config_list(client_config):
-    output = subprocess.run(f"cat /etc/wireguard/{client_config}", shell=True, capture_output=True, text=True).stdout.strip()
-    return output.split('\n')
+    config_path = f"/etc/wireguard/{client_config}"
+    try:
+        with open(config_path, 'r') as f:
+            output = f.read().strip()
+            return output.split('\n')
+    except FileNotFoundError:
+        raise FileNotFoundError("Client configuration file not found")
 
 #TODO: where is the logfile??
-#TODO: is service connected?
+#TODO: is service connected?