Open mizoil opened 1 week ago
donaldzou
commented
1 week ago Hi! Thank you for providing input. In latest version v4.1.0, you are able to just copy the configuration file directly from WGDashboard :) I'm not too familiar with VLESS, Vmess, and Shadowsocks... will need to do some homework on this. Also I'm currently working to support AmneziaWG, please stay tuned!
mizoil
commented
1 week ago Hi! Thank you for providing input. In latest version
v4.1.0, you are able to just copy the configuration file directly from WGDashboard :) I'm not too familiar with VLESS, Vmess, and Shadowsocks... will need to do some homework on this. Also I'm currently working to support AmneziaWG, please stay tuned!
Thank you for the suggestion. Great, I will be waiting for the VLESS setup. I hope it works out for you, and also that AmneziaWG will be released after the update.
NOXCIS
commented
1 week ago @donaldzou
Also use the system links
if [[ "$AMNEZIA_WG" == "true" ]]; then
mkdir -p /usr/local/bin
ln -sf /usr/bin/awg /usr/local/bin/wg
ln -sf /usr/bin/awg-quick /usr/local/bin/wg-quick
fi
def allowAccessPeers(self, listOfPublicKeys):
if not self.getStatus():
self.toggleConfiguration()
interface_address = self.get_awg_iface_address()
for public_key in listOfPublicKeys:
peer_data = sqlSelect("SELECT * FROM '%s_restrict_access' WHERE id = ?" % self.Name, (public_key,)).fetchone()
if peer_data is not None:
sqlUpdate("INSERT INTO '%s' SELECT * FROM %s_restrict_access WHERE id = ?" % (self.Name, self.Name), (peer_data['id'],))
sqlUpdate("DELETE FROM '%s_restrict_access' WHERE id = ?" % self.Name, (peer_data['id'],))
presharedKeyExist = len(peer_data['preshared_key']) > 0
uid = uuid.uuid4()
if presharedKeyExist:
with open(f"{uid}", "w+") as f:
f.write(peer_data['preshared_key'])
subprocess.check_output(f"wg set {self.Name} peer {peer_data['id']} allowed-ips {peer_data['allowed_ip'].replace(' ', '')}{f' preshared-key {uid}' if presharedKeyExist else ''}",
shell=True, stderr=subprocess.STDOUT)
if presharedKeyExist:
os.remove(str(uid))
else:
return ResponseObject(False, "Failed to allow access of peer " + public_key)
if not self.__wgSave():
return ResponseObject(False, "Failed to save configuration through WireGuard")
write_error = self.patch_awg_iface_address(interface_address)
if write_error:
return write_error
self.__getPeers()
return ResponseObject(True, "Allow access successfully!")
def restrictPeers(self, listOfPublicKeys):
numOfRestrictedPeers = 0
numOfFailedToRestrictPeers = 0
if not self.getStatus():
self.toggleConfiguration()
interface_address = self.get_awg_iface_address()
for public_key in listOfPublicKeys:
found, pf = self.searchPeer(public_key)
if found:
try:
subprocess.check_output(f"wg set {self.Name} peer {pf.id} remove", shell=True, stderr=subprocess.STDOUT)
sqlUpdate("INSERT INTO '%s_restrict_access' SELECT * FROM %s WHERE id = ?" % (self.Name, self.Name), (pf.id,))
sqlUpdate("UPDATE '%s_restrict_access' SET status = 'stopped' WHERE id = ?" % (self.Name,), (pf.id,))
sqlUpdate("DELETE FROM '%s' WHERE id = ?" % self.Name, (pf.id,))
numOfRestrictedPeers += 1
except Exception:
numOfFailedToRestrictPeers += 1
if not self.__wgSave():
return ResponseObject(False, "Failed to save configuration through WireGuard")
write_error = self.patch_awg_iface_address(interface_address)
if write_error:
return write_error
self.__getPeers()
if numOfRestrictedPeers == len(listOfPublicKeys):
return ResponseObject(True, f"Restricted {numOfRestrictedPeers} peer(s)")
return ResponseObject(False, f"Restricted {numOfRestrictedPeers} peer(s) successfully. Failed to restrict {numOfFailedToRestrictPeers} peer(s)")
def deletePeers(self, listOfPublicKeys):
numOfDeletedPeers = 0
numOfFailedToDeletePeers = 0
if not self.getStatus():
self.toggleConfiguration()
interface_address = self.get_awg_iface_address()
for public_key in listOfPublicKeys:
found, pf = self.searchPeer(public_key)
if found:
try:
subprocess.check_output(f"wg set {self.Name} peer {pf.id} remove", shell=True, stderr=subprocess.STDOUT)
sqlUpdate("DELETE FROM '%s' WHERE id = ?" % self.Name, (pf.id,))
numOfDeletedPeers += 1
except Exception:
numOfFailedToDeletePeers += 1
if not self.__wgSave():
return ResponseObject(False, "Failed to save configuration through WireGuard")
write_error = self.patch_awg_iface_address(interface_address)
if write_error:
return write_error
self.__getPeers()
if numOfDeletedPeers == len(listOfPublicKeys):
return ResponseObject(True, f"Deleted {numOfDeletedPeers} peer(s)")
return ResponseObject(False, f"Deleted {numOfDeletedPeers} peer(s) successfully. Failed to delete {numOfFailedToDeletePeers} peer(s)")def updatePeer(self, name: str, private_key: str, preshared_key: str, dns_addresses: str, allowed_ip: str, endpoint_allowed_ip: str, mtu: int, keepalive: int) -> ResponseObject: if not self.configuration.getStatus(): self.configuration.toggleConfiguration()
existingAllowedIps = [item for row in list(
map(lambda x: [q.strip() for q in x.split(',')],
map(lambda y: y.allowed_ip,
list(filter(lambda k: k.id != self.id, self.configuration.getPeersList()))))) for item in row]
if allowed_ip in existingAllowedIps:
return ResponseObject(False, "Allowed IP already taken by another peer")
if not _checkIPWithRange(endpoint_allowed_ip):
return ResponseObject(False, f"Endpoint Allowed IPs format is incorrect")
if len(dns_addresses) > 0 and not _checkDNS(dns_addresses):
return ResponseObject(False, f"DNS format is incorrect")
if mtu < 0 or mtu > 1460:
return ResponseObject(False, "MTU format is not correct")
if keepalive < 0:
return ResponseObject(False, "Persistent Keepalive format is not correct")
if len(private_key) > 0:
pubKey = _generatePublicKey(private_key)
if not pubKey[0] or pubKey[1] != self.id:
return ResponseObject(False, "Private key does not match with the public key")
try:
#Retrieve the current interface address
interface_address = subprocess.check_output(
f"ip addr show {self.configuration.Name} | grep 'inet ' | awk '{{print $2}}'",
shell=True
).decode().strip()
# Set up peer update
rd = random.Random()
uid = uuid.UUID(int=rd.getrandbits(128), version=4)
pskExist = len(preshared_key) > 0
if pskExist:
with open(f"{uid}", "w+") as f:
f.write(preshared_key)
newAllowedIPs = allowed_ip.replace(" ", "")
# Update peer configuration with wg set
updateAllowedIp = subprocess.check_output(
f"wg set {self.configuration.Name} peer {self.id} allowed-ips {newAllowedIPs}{f' preshared-key {uid}' if pskExist else ''}",
shell=True, stderr=subprocess.STDOUT
)
if pskExist:
os.remove(str(uid))
if len(updateAllowedIp.decode().strip("\n")) != 0:
return ResponseObject(False, "Update peer failed when updating Allowed IPs")
# Save configuration with wg-quick save
saveConfig = subprocess.check_output(f"wg-quick save {self.configuration.Name}",
shell=True, stderr=subprocess.STDOUT)
if f"wg showconf {self.configuration.Name}" not in saveConfig.decode().strip('\n'):
return ResponseObject(False, "Update peer failed when saving the configuration")
# Rewrite the interface address to config file
try:
# Read the existing configuration file
with open(f"/etc/wireguard/{self.configuration.Name}.conf", "r") as conf_file:
lines = conf_file.readlines()
# Find the position of the [Interface] section
interface_index = None
for i, line in enumerate(lines):
if line.strip() == "[Interface]":
interface_index = i
break
# Check if Address is already present
address_present = any(line.strip().startswith("Address") for line in lines)
# If [Interface] section was found and Address is not present, insert the address below it
if interface_index is not None:
if not address_present:
lines.insert(interface_index + 1, f"Address = {interface_address}\n")
# Write the updated configuration back to the file
with open(f"/etc/wireguard/{self.configuration.Name}.conf", "w") as conf_file:
conf_file.writelines(lines)
except IOError:
return ResponseObject(False, "Failed to write the interface address to the config file")
# Update database
sqlUpdate(
'''UPDATE '%s' SET name = ?, private_key = ?, DNS = ?, endpoint_allowed_ip = ?, mtu = ?,
keepalive = ?, preshared_key = ? WHERE id = ?''' % self.configuration.Name,
(name, private_key, dns_addresses, endpoint_allowed_ip, mtu,
keepalive, preshared_key, self.id,)
)
return ResponseObject()
except subprocess.CalledProcessError as exc:
return ResponseObject(False, exc.output.decode("UTF-8").strip())def toggleConfiguration(self) -> [bool, str]: self.getStatus() interface_address = self.get_awg_iface_address()
if self.Status:
try:
check = subprocess.check_output(f"wg-quick down {self.Name}",
shell=True, stderr=subprocess.STDOUT)
except subprocess.CalledProcessError as exc:
return False, str(exc.output.strip().decode("utf-8"))
# Write the interface address after bringing it down
write_error = self.patch_awg_iface_address(interface_address)
if write_error:
return write_error
else:
try:
check = subprocess.check_output(f"wg-quick up {self.Name}",
shell=True, stderr=subprocess.STDOUT)
except subprocess.CalledProcessError as exc:
return False, str(exc.output.strip().decode("utf-8"))
self.getStatus()
return True, Nonedef get_awg_iface_address(self):
try:
interface_address = subprocess.check_output(
f"ip addr show {self.Name} | grep 'inet ' | awk '{{print $2}}'",
shell=True).decode().strip()
return interface_address
except subprocess.CalledProcessError:
return None # or handle the error as needed
def patch_awg_iface_address(self, interface_address):
try:
with open(f"/etc/wireguard/{self.Name}.conf", "r") as conf_file:
lines = conf_file.readlines()
interface_index = next((i for i, line in enumerate(lines) if line.strip() == "[Interface]"), None)
address_present = any(line.strip().startswith("Address") for line in lines)
if interface_index is not None and not address_present:
lines.insert(interface_index + 1, f"Address = {interface_address}\n")
with open(f"/etc/wireguard/{self.Name}.conf", "w") as conf_file:
conf_file.writelines(lines)
except IOError:
return ResponseObject(False, "Failed to write the interface address to the config file")
Hi!
I want to propose my idea, which supports WireGuard and includes the protocols AmneziaWG, VLESS, Vmess, and Shadowsocks.
Why is my idea good? I chose these protocols because they provide convenience in setup and usage. When I create a configuration, it's easier for me to copy it from the clipboard or use a QR code, rather than manually sending a file. This is especially useful if there's no access to a laptop (for example, when you're not at home and can't activate the QR code — in that case, you'll have to send the file manually, which can be tricky due to errors in file transfer or problems with the file name, especially if it's too long or if there are issues in messaging apps like Telegram). In these situations, VLESS is a more convenient solution, as it allows you to simply copy the text from the clipboard and paste it on a mobile device.
Using WireGuard combined with VLESS and other protocols ensures stable internet speed and ease of use. It's a good option for providing security and high speed.
I also like the idea of using AmneziaWG — this protocol has several advantages and is a useful addition.
What do you think? Do you think I made the right choice with the protocols and ease of use?
I'd love to hear your opinion! =)