Mirrors/WGDashboard
1
0
Fork 0
mirror of https://github.com/donaldzou/WGDashboard.git synced 2025-10-02 23:36:17 +00:00
Code Issues Actions 4 Packages Projects Releases Wiki Activity

feat: slight refactors to improve logic

Browse Source
This commit is contained in:
Daan Selen
2025-09-15 13:20:17 +02:00
parent b2f306789c
commit 73a969a6bf
4 changed files with 76 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
src/modules/AmneziaWireguardConfiguration.py
View File

@@ -174,12 +174,18 @@ class AmneziaWireguardConfiguration(WireguardConfiguration):
def getPeers(self):
self.Peers.clear()
current_app.logger.info(f"Refreshing {self.Name} peer list")
if self.configurationFileChanged():
with open(self.configPath, 'r') as configFile:
p = []
pCounter = -1
content = configFile.read().split('\n')
try:
if "[Peer]" not in content:
current_app.logger.info(f"{self.Name} config has no [Peer] section")
return
peerStarts = content.index("[Peer]")
content = content[peerStarts:]
for i in content:

134
src/modules/WireguardConfiguration.py
View File

@@ -399,80 +399,82 @@ class WireguardConfiguration:
def getPeers(self):
tmpList = []
current_app.logger.info(f"Refreshing {self.Name} peer list")
if self.configurationFileChanged():
with open(self.configPath, 'r') as configFile:
p = []
pCounter = -1
content = configFile.read().split('\n')
try:
if "[Peer]" in content:
peerStarts = content.index("[Peer]")
content = content[peerStarts:]
for i in content:
if not RegexMatch("#(.*)", i) and not RegexMatch(";(.*)", i):
if i == "[Peer]":
pCounter += 1
p.append({})
p[pCounter]["name"] = ""
else:
if len(i) > 0:
split = re.split(r'\s*=\s*', i, 1)
if len(split) == 2:
p[pCounter][split[0]] = split[1]
if RegexMatch("#Name# = (.*)", i):
split = re.split(r'\s*=\s*', i, 1)
if len(split) == 2:
p[pCounter]["name"] = split[1]
for i in p:
if "PublicKey" in i.keys():
with self.engine.connect() as conn:
tempPeer = conn.execute(
self.peersTable.select().where(
if "[Peer]" not in content:
current_app.logger.info(f"{self.Name} config has no [Peer] section")
return
peerStarts = content.index("[Peer]")
content = content[peerStarts:]
for i in content:
if not RegexMatch("#(.*)", i) and not RegexMatch(";(.*)", i):
if i == "[Peer]":
pCounter += 1
p.append({})
p[pCounter]["name"] = ""
else:
if len(i) > 0:
split = re.split(r'\s*=\s*', i, 1)
if len(split) == 2:
p[pCounter][split[0]] = split[1]
if RegexMatch("#Name# = (.*)", i):
split = re.split(r'\s*=\s*', i, 1)
if len(split) == 2:
p[pCounter]["name"] = split[1]
for i in p:
if "PublicKey" in i.keys():
with self.engine.connect() as conn:
tempPeer = conn.execute(
self.peersTable.select().where(
self.peersTable.columns.id == i['PublicKey']
)
).mappings().fetchone()
if tempPeer is None:
tempPeer = {
"id": i['PublicKey'],
"private_key": "",
"DNS": self.DashboardConfig.GetConfig("Peers", "peer_global_DNS")[1],
"endpoint_allowed_ip": self.DashboardConfig.GetConfig("Peers", "peer_endpoint_allowed_ip")[
1],
"name": i.get("name"),
"total_receive": 0,
"total_sent": 0,
"total_data": 0,
"endpoint": "N/A",
"status": "stopped",
"latest_handshake": "N/A",
"allowed_ip": i.get("AllowedIPs", "N/A"),
"cumu_receive": 0,
"cumu_sent": 0,
"cumu_data": 0,
"mtu": self.DashboardConfig.GetConfig("Peers", "peer_mtu")[1] if len(self.DashboardConfig.GetConfig("Peers", "peer_mtu")[1]) > 0 else None,
"keepalive": self.DashboardConfig.GetConfig("Peers", "peer_keep_alive")[1] if len(self.DashboardConfig.GetConfig("Peers", "peer_keep_alive")[1]) > 0 else None,
"remote_endpoint": self.DashboardConfig.GetConfig("Peers", "remote_endpoint")[1],
"preshared_key": i["PresharedKey"] if "PresharedKey" in i.keys() else ""
}
with self.engine.begin() as conn:
conn.execute(
self.peersTable.insert().values(tempPeer)
)
else:
with self.engine.begin() as conn:
conn.execute(
self.peersTable.update().values({
"allowed_ip": i.get("AllowedIPs", "N/A")
}).where(
self.peersTable.columns.id == i['PublicKey']
)
).mappings().fetchone()
if tempPeer is None:
tempPeer = {
"id": i['PublicKey'],
"private_key": "",
"DNS": self.DashboardConfig.GetConfig("Peers", "peer_global_DNS")[1],
"endpoint_allowed_ip": self.DashboardConfig.GetConfig("Peers", "peer_endpoint_allowed_ip")[
1],
"name": i.get("name"),
"total_receive": 0,
"total_sent": 0,
"total_data": 0,
"endpoint": "N/A",
"status": "stopped",
"latest_handshake": "N/A",
"allowed_ip": i.get("AllowedIPs", "N/A"),
"cumu_receive": 0,
"cumu_sent": 0,
"cumu_data": 0,
"mtu": self.DashboardConfig.GetConfig("Peers", "peer_mtu")[1] if len(self.DashboardConfig.GetConfig("Peers", "peer_mtu")[1]) > 0 else None,
"keepalive": self.DashboardConfig.GetConfig("Peers", "peer_keep_alive")[1] if len(self.DashboardConfig.GetConfig("Peers", "peer_keep_alive")[1]) > 0 else None,
"remote_endpoint": self.DashboardConfig.GetConfig("Peers", "remote_endpoint")[1],
"preshared_key": i["PresharedKey"] if "PresharedKey" in i.keys() else ""
}
with self.engine.begin() as conn:
conn.execute(
self.peersTable.insert().values(tempPeer)
)
else:
with self.engine.begin() as conn:
conn.execute(
self.peersTable.update().values({
"allowed_ip": i.get("AllowedIPs", "N/A")
}).where(
self.peersTable.columns.id == i['PublicKey']
)
)
tmpList.append(Peer(tempPeer, self))
else:
current_app.logger.warning(f"{self.Name} is an empty configuration")
)
tmpList.append(Peer(tempPeer, self))
except Exception as e:
current_app.logger.error(f"{self.Name} getPeers() Error", e)
else: