kamp/WGDashboard
1
0
Fork 0
mirror of https://github.com/donaldzou/WGDashboard.git synced 2024-11-06 16:00:28 +01:00
Code

Fixed multiple IP address on a Configuration

Browse Source
This commit is contained in:
Donald Zou 2024-08-25 15:14:09 +08:00
parent 79c03db9a0
commit 453d474104
1 changed files with 96 additions and 96 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
src/dashboard.py
View File

@ -442,6 +442,8 @@ class WireguardConfiguration:
return self.message return self.message
def __init__(self, name: str = None, data: dict = None): def __init__(self, name: str = None, data: dict = None):
print(f"Initialized {name}")
self.__parser: configparser.ConfigParser = configparser.ConfigParser(strict=False) self.__parser: configparser.ConfigParser = configparser.ConfigParser(strict=False)
self.__parser.optionxform = str self.__parser.optionxform = str
self.__configFileModifiedTime = None self.__configFileModifiedTime = None
@ -589,10 +591,14 @@ class WireguardConfiguration:
for i in restricted: for i in restricted:
self.RestrictedPeers.append(Peer(i, self)) self.RestrictedPeers.append(Peer(i, self))
def __getPeers(self): def configurationFileChanged(self) :
mt = os.path.getmtime(os.path.join(WG_CONF_PATH, f'{self.Name}.conf')) mt = os.path.getmtime(os.path.join(WG_CONF_PATH, f'{self.Name}.conf'))
# if self.__configFileModifiedTime is None or self.__configFileModifiedTime != mt: changed = self.__configFileModifiedTime is None or self.__configFileModifiedTime != mt
self.__configFileModifiedTime = mt
return changed
def __getPeers(self):
if self.configurationFileChanged():
self.Peers = [] self.Peers = []
with open(os.path.join(WG_CONF_PATH, f'{self.Name}.conf'), 'r') as configFile: with open(os.path.join(WG_CONF_PATH, f'{self.Name}.conf'), 'r') as configFile:
p = [] p = []
@ -664,7 +670,7 @@ class WireguardConfiguration:
self.Peers.append(Peer(checkIfExist, self)) self.Peers.append(Peer(checkIfExist, self))
except Exception as e: except Exception as e:
print(f"[WGDashboard] {self.Name} Error: {str(e)}") print(f"[WGDashboard] {self.Name} Error: {str(e)}")
self.__configFileModifiedTime = mt
def addPeers(self, peers: list): def addPeers(self, peers: list):
for p in peers: for p in peers:
@ -1284,16 +1290,20 @@ def _regexMatch(regex, text):
return pattern.search(text) is not None return pattern.search(text) is not None
def _getConfigurationList() -> [WireguardConfiguration]: def _getConfigurationList():
configurations = {} # configurations = {}
for i in os.listdir(WG_CONF_PATH): for i in os.listdir(WG_CONF_PATH):
if _regexMatch("^(.{1,}).(conf)$", i): if _regexMatch("^(.{1,}).(conf)$", i):
i = i.replace('.conf', '') i = i.replace('.conf', '')
try: try:
configurations[i] = WireguardConfiguration(i) if i in WireguardConfigurations.keys():
if WireguardConfigurations[i].configurationFileChanged():
WireguardConfigurations[i] = WireguardConfiguration(i)
else:
WireguardConfigurations[i] = WireguardConfiguration(i)
except WireguardConfiguration.InvalidConfigurationFileException as e: except WireguardConfiguration.InvalidConfigurationFileException as e:
print(f"{i} have an invalid configuration file.") print(f"{i} have an invalid configuration file.")
return configurations
def _checkIPWithRange(ip): def _checkIPWithRange(ip):
@ -1354,8 +1364,7 @@ def _generatePrivateKey() -> [bool, str]:
except subprocess.CalledProcessError: except subprocess.CalledProcessError:
return False, None return False, None
def _getWireguardConfigurationAvailableIP(configName: str, all: bool = False) -> tuple[bool, list[str]] | tuple[bool, None]:
def _getWireguardConfigurationAvailableIP(configName: str) -> tuple[bool, list[str]] | tuple[bool, None]:
if configName not in WireguardConfigurations.keys(): if configName not in WireguardConfigurations.keys():
return False, None return False, None
configuration = WireguardConfigurations[configName] configuration = WireguardConfigurations[configName]
@ -1387,6 +1396,7 @@ def _getWireguardConfigurationAvailableIP(configName: str) -> tuple[bool, list[s
if h not in existedAddress: if h not in existedAddress:
availableAddress.append(ipaddress.ip_network(h).compressed) availableAddress.append(ipaddress.ip_network(h).compressed)
count += 1 count += 1
if not all:
if network.version == 6 and count > 255: if network.version == 6 and count > 255:
break break
return True, availableAddress return True, availableAddress
@ -1534,7 +1544,7 @@ def API_SignOut():
@app.route(f'{APP_PREFIX}/api/getWireguardConfigurations', methods=["GET"]) @app.route(f'{APP_PREFIX}/api/getWireguardConfigurations', methods=["GET"])
def API_getWireguardConfigurations(): def API_getWireguardConfigurations():
# WireguardConfigurations = _getConfigurationList() _getConfigurationList()
return ResponseObject(data=[wc for wc in WireguardConfigurations.values()]) return ResponseObject(data=[wc for wc in WireguardConfigurations.values()])
@ -1841,17 +1851,7 @@ def API_addPeers(configName):
if i not in availableIps[1]: if i not in availableIps[1]:
return ResponseObject(False, f"This IP is not available: {i}") return ResponseObject(False, f"This IP is not available: {i}")
config.addPeers([{"id": public_key, "allowed_ip": ''.join(allowed_ips)}]) config.addPeers([{"id": public_key, "allowed_ip": ','.join(allowed_ips)}])
# subprocess.check_output(
# f"wg set {config.Name} peer {public_key} allowed-ips {''.join(allowed_ips)}",
# shell=True, stderr=subprocess.STDOUT)
# if len(preshared_key) > 0:
# subprocess.check_output(
# f"wg set {config.Name} peer {public_key} preshared-key {preshared_key}",
# shell=True, stderr=subprocess.STDOUT)
# subprocess.check_output(
# f"wg-quick save {config.Name}", shell=True, stderr=subprocess.STDOUT)
# config.getPeersList()
found, peer = config.searchPeer(public_key) found, peer = config.searchPeer(public_key)
if found: if found:
return peer.updatePeer(name, private_key, preshared_key, dns_addresses, ",".join(allowed_ips), return peer.updatePeer(name, private_key, preshared_key, dns_addresses, ",".join(allowed_ips),
@ -2188,7 +2188,7 @@ _, WG_CONF_PATH = DashboardConfig.GetConfig("Server", "wg_conf_path")
WireguardConfigurations: dict[str, WireguardConfiguration] = {} WireguardConfigurations: dict[str, WireguardConfiguration] = {}
WireguardConfigurations = _getConfigurationList() _getConfigurationList()
def startThreads(): def startThreads():
bgThread = threading.Thread(target=backGroundThread) bgThread = threading.Thread(target=backGroundThread)