fix: peer key validation regex (#1158)

* fix: peer key validation regex

* refactor: cache cleaned AllowedIPs from validation pass to avoid duplication

---------

Co-authored-by: Dan Hollis <dh@redteam.sh>

src/modules/AmneziaConfiguration.py

@@ -6,7 +6,7 @@ from flask import current_app
 from .PeerJobs import PeerJobs
 from .AmneziaPeer import AmneziaPeer
 from .PeerShareLinks import PeerShareLinks
-from .Utilities import RegexMatch, CheckAddress
+from .Utilities import RegexMatch, CheckAddress, CheckPeerKey
 from .WireguardConfiguration import WireguardConfiguration
 from .DashboardWebHooks import DashboardWebHooks
 
@@ -241,6 +241,15 @@ class AmneziaConfiguration(WireguardConfiguration):
             "peers": []
         }
         try:
+            cleanedAllowedIPs = {}
+            for p in peers:
+                newAllowedIPs = p['allowed_ip'].replace(" ", "")
+                if not CheckAddress(newAllowedIPs):
+                    return False, [], "Allowed IPs entry format is incorrect"
+                if not CheckPeerKey(p["id"]):
+                    return False, [], "Peer key format is incorrect"
+                cleanedAllowedIPs[p["id"]] = newAllowedIPs
+
             with self.engine.begin() as conn:
                 for i in peers:
                     newPeer = {
@@ -276,14 +285,7 @@ class AmneziaConfiguration(WireguardConfiguration):
                     with open(uid, "w+") as f:
                         f.write(p['preshared_key'])
 
-                newAllowedIPs = p['allowed_ip'].replace(" ", "")
+                command = [self.Protocol, "set", self.Name, "peer", p['id'], "allowed-ips", cleanedAllowedIPs[p["id"]], "preshared-key", uid if presharedKeyExist else "/dev/null"]
-                if not CheckAddress(newAllowedIPs):
-                    return False, [], "Allowed IPs entry format is incorrect"
-                
-                if not re.match(r"^[A-Za-z0-9+/]{42}[A-Ea-e0-9]=$", p["id"]):
-                    return False, [], "Peer key format is incorrect"
-
-                command = [self.Protocol, "set", self.Name, "peer", p['id'], "allowed-ips", newAllowedIPs, "preshared-key", uid if presharedKeyExist else "/dev/null"]
                 subprocess.check_output(command, stderr=subprocess.STDOUT)
 
                 if presharedKeyExist:

src/modules/WireguardConfiguration.py

@@ -512,6 +512,15 @@ class WireguardConfiguration:
             "peers": []
         }
         try:
+            cleanedAllowedIPs = {}
+            for p in peers:
+                newAllowedIPs = p['allowed_ip'].replace(" ", "")
+                if not CheckAddress(newAllowedIPs):
+                    return False, [], "Allowed IPs entry format is incorrect"
+                if not CheckPeerKey(p["id"]):
+                    return False, [], "Peer key format is incorrect"
+                cleanedAllowedIPs[p["id"]] = newAllowedIPs
+
             with self.engine.begin() as conn:
                 for i in peers:
                     newPeer = {
@@ -547,14 +556,7 @@ class WireguardConfiguration:
                     with open(uid, "w+") as f:
                         f.write(p['preshared_key'])
 
-                newAllowedIPs = p['allowed_ip'].replace(" ", "")
+                command = [self.Protocol, "set", self.Name, "peer", p['id'], "allowed-ips", cleanedAllowedIPs[p["id"]], "preshared-key", uid if presharedKeyExist else "/dev/null"]
-                if not CheckAddress(newAllowedIPs):
-                    return False, [], "Allowed IPs entry format is incorrect"
-
-                if not CheckPeerKey(p["id"]):
-                    return False, [], "Peer key format is incorrect"
-
-                command = [self.Protocol, "set", self.Name, "peer", p['id'], "allowed-ips", newAllowedIPs, "preshared-key", uid if presharedKeyExist else "/dev/null"]
                 subprocess.check_output(command, stderr=subprocess.STDOUT)
 
                 if presharedKeyExist: