From b42cab5cdef812fffe2fdb957f4c7b158182b540 Mon Sep 17 00:00:00 2001 From: jafnhaar Date: Wed, 12 Jul 2023 08:42:06 +0300 Subject: [PATCH 1/3] added ipaddress lib logic, changed default cidr from 24 to 22 --- wg-gen.py | 91 +++++++++++++++++++++++++++++++------------------------ 1 file changed, 51 insertions(+), 40 deletions(-) diff --git a/wg-gen.py b/wg-gen.py index 29f651d..1ab5a24 100644 --- a/wg-gen.py +++ b/wg-gen.py @@ -1,4 +1,5 @@ #!/usr/bin/env python3 +import ipaddress import json import os.path import random @@ -13,15 +14,18 @@ def get_public_ip(self) -> str: """Returns a string of a public ip address""" return json.loads(subprocess.check_output('curl -s https://ipinfo.io', shell=True).decode('utf-8'))['ip'] - def generate_private_ip(self): - """Returns random ip address in 10.0.0.0/8 range""" - return f'10.{random.randrange(0, 255)}.{random.randrange(0, 255)}.' + def generate_private_ip(self, subnet: int = 22): + """Returns random subnet in 10.0.0.0/8 range, defaults to 22 bit subnet""" + generated_ip_interface = ipaddress.IPv4Interface( + f'10.{random.randrange(0, 255)}.{random.randrange(0, 255)}.{random.randrange(0, 255)}/{subnet}' + ) + return (list(generated_ip_interface.network.hosts())[-1], generated_ip_interface.netmask) def get_current_time(self) -> str: """Returns current time w/o milliseconds""" return datetime.now().isoformat(' ', 'seconds') - def get_default_interface_name(self): + def get_default_interface_name(self): return subprocess.check_output("ip route show 0.0.0.0/0 | awk '{print $5}' | head -n 1", shell=True).decode('UTF-8').strip() def read_json(self) -> json: @@ -50,43 +54,51 @@ def generate_guest_configs(self, name: str, data: Dict) -> Dict: "Generates peer config and adds it to wghub.conf. Takes config json file and spits it back with changed seqno" peer_keys = self.generate_wg_keys() preshared_key = self.generate_preshared_key() - if int(data['seqno']) > 254: - return print("Maximum amount of IPs in /24 subnet exceeded.") - else: - with open('wghub.conf', 'a') as file: - file.write( - f'\n\n' - f'#{data["seqno"]} Generated at {self.get_current_time()} for {name}\n' - f'[Peer]\n' - f'PublicKey = {peer_keys[1]}\n' - f'PresharedKey = {preshared_key}\n' - f'AllowedIPs = {data["private_ip"]}{data["seqno"]}/32' - ) - with open('wgclient_' + data['seqno'] + '.conf', 'w') as file: - file.write( - f'#{data["seqno"]} generated at {self.get_current_time()} for {name}\n' - f'[Interface]\n' - f'PrivateKey = {peer_keys[0]}\n' - f'Address = {data["private_ip"]}{data["seqno"]}{data["cidr"]}\n' - f'MTU = 1280\n' - f'DNS = {data["DNS"]}\n\n' - f'[Peer]\n' - f'PublicKey = {data["hub_public_key"]}\n' - f'PresharedKey = {preshared_key}\n' - f'AllowedIPs = {data["client_allowed_ips"]}\n' - f'Endpoint = {data["public_ip"]}:{data["port"]}\n' - f'PersistentKeepalive = 25\n' - ) - data['seqno'] = str(int(data['seqno']) + 1) - return data + gw_address = ipaddress.IPv4Interface(f'{data["private_ip"]}/{data["private_ip_netmask"]}') + client_private_ip = list(gw_address.network.hosts())[int(data["seqno"])] + client_private_interface = ipaddress.IPv4Interface(f'{client_private_ip}/{data["private_ip_netmask"]}') + + if gw_address == client_private_ip: + raise Exception('You reached last IP address in your wireguard network. No more clients can be generated.') + + with open('wghub.conf', 'a') as file: + file.write( + f'\n\n' + f'# {data["seqno"]} Generated at {self.get_current_time()} for {name}\n' + f'[Peer]\n' + f'PublicKey = {peer_keys[1]}\n' + f'PresharedKey = {preshared_key}\n' + f'AllowedIPs = {data["private_ip"]}{data["seqno"]}/32' + ) + with open('wgclient_' + data['seqno'] + '.conf', 'w') as file: + file.write( + f'#{data["seqno"]} generated at {self.get_current_time()} for {name}\n' + f'[Interface]\n' + f'PrivateKey = {peer_keys[0]}\n' + f'Address = {client_private_interface.with_prefixlen}\n' + f'MTU = 1280\n' + f'DNS = {data["DNS"]}\n\n' + f'[Peer]\n' + f'PublicKey = {data["hub_public_key"]}\n' + f'PresharedKey = {preshared_key}\n' + f'AllowedIPs = {data["client_allowed_ips"]}\n' + f'Endpoint = {data["public_ip"]}:{data["port"]}\n' + f'PersistentKeepalive = 25\n' + ) + self.gen_qr_code(data) + + data['seqno'] = str(int(data['seqno']) + 1) + return data def generate_hub(self, data: Dict[str, str]) -> None: """Generates wghub.conf configuration file from dictionary""" + gw_address = ipaddress.IPv4Interface(f'{data["private_ip"]}/{data["private_ip_netmask"]}') + with open('wghub.conf', 'w') as file: file.write( f'# hub generated at {self.get_current_time()}\n' f'[Interface]\n' - f'Address = {data["private_ip"]}1{data["cidr"]}\n' + f'Address = {gw_address.with_prefixlen}\n' f'ListenPort = {data["port"]}\n' f'PrivateKey = {data["hub_private_key"]}\n' f'SaveConfig = False\n' @@ -100,7 +112,7 @@ def generate_hub(self, data: Dict[str, str]) -> None: def gen_qr_code(self, data: dict) -> None: """Generates qr code from a configuration file""" - subprocess.run(f'qrencode -t ansiutf8 < wgclient_{int(data["seqno"]) - 1}.conf', shell=True) + subprocess.run(f'qrencode -t ansiutf8 < wgclient_{int(data["seqno"])}.conf', shell=True) def main(): @@ -114,17 +126,17 @@ def main(): wireguard_data = wireguard.generate_guest_configs('client', wireguard_data) wireguard.save_json(wireguard_data) - wireguard.gen_qr_code(wireguard_data) else: hub_keys = wireguard.generate_wg_keys() + hub_ips = wireguard.generate_private_ip() wireguard_data = { - 'private_ip': wireguard.generate_private_ip(), + 'private_ip': str(hub_ips[0]), + 'private_ip_netmask': str(hub_ips[1]), 'public_ip': wireguard.get_public_ip(), 'hub_private_key': hub_keys[0], 'hub_public_key': hub_keys[1], - 'seqno': '2', + 'seqno': '1', 'port': str(random.randrange(10000, 60000)), - 'cidr': '/24', 'DNS': '1.1.1.1', 'oiface': wireguard.get_default_interface_name(), 'client_allowed_ips': '0.0.0.0/0' @@ -137,7 +149,6 @@ def main(): except IndexError: wireguard_data = wireguard.generate_guest_configs('client', wireguard_data) - wireguard.gen_qr_code(wireguard_data) wireguard.save_json(wireguard_data) From 29d7000ed22f3ec729114d5a0021be5cc3256e5a Mon Sep 17 00:00:00 2001 From: jafnhaar Date: Wed, 12 Jul 2023 11:48:43 +0300 Subject: [PATCH 2/3] changed vars name to be more descriptive --- wg-gen.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/wg-gen.py b/wg-gen.py index 1ab5a24..c7ed535 100644 --- a/wg-gen.py +++ b/wg-gen.py @@ -55,10 +55,10 @@ def generate_guest_configs(self, name: str, data: Dict) -> Dict: peer_keys = self.generate_wg_keys() preshared_key = self.generate_preshared_key() gw_address = ipaddress.IPv4Interface(f'{data["private_ip"]}/{data["private_ip_netmask"]}') - client_private_ip = list(gw_address.network.hosts())[int(data["seqno"])] - client_private_interface = ipaddress.IPv4Interface(f'{client_private_ip}/{data["private_ip_netmask"]}') + client_ip = list(gw_address.network.hosts())[int(data["seqno"])] + client_interface = ipaddress.IPv4Interface(f'{client_ip}/{data["private_ip_netmask"]}') - if gw_address == client_private_ip: + if gw_address == client_interface: raise Exception('You reached last IP address in your wireguard network. No more clients can be generated.') with open('wghub.conf', 'a') as file: @@ -75,7 +75,7 @@ def generate_guest_configs(self, name: str, data: Dict) -> Dict: f'#{data["seqno"]} generated at {self.get_current_time()} for {name}\n' f'[Interface]\n' f'PrivateKey = {peer_keys[0]}\n' - f'Address = {client_private_interface.with_prefixlen}\n' + f'Address = {client_interface.with_prefixlen}\n' f'MTU = 1280\n' f'DNS = {data["DNS"]}\n\n' f'[Peer]\n' From 2e1c4a85abca90b48afadffda99e3f08c3e9c0bf Mon Sep 17 00:00:00 2001 From: jafnhaar Date: Wed, 12 Jul 2023 12:14:17 +0300 Subject: [PATCH 3/3] fixed numbers of generated configs --- wg-gen.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/wg-gen.py b/wg-gen.py index c7ed535..c367418 100644 --- a/wg-gen.py +++ b/wg-gen.py @@ -70,7 +70,7 @@ def generate_guest_configs(self, name: str, data: Dict) -> Dict: f'PresharedKey = {preshared_key}\n' f'AllowedIPs = {data["private_ip"]}{data["seqno"]}/32' ) - with open('wgclient_' + data['seqno'] + '.conf', 'w') as file: + with open('wgclient_' + str(int(data['seqno']) + 1) + '.conf', 'w') as file: file.write( f'#{data["seqno"]} generated at {self.get_current_time()} for {name}\n' f'[Interface]\n' @@ -112,7 +112,7 @@ def generate_hub(self, data: Dict[str, str]) -> None: def gen_qr_code(self, data: dict) -> None: """Generates qr code from a configuration file""" - subprocess.run(f'qrencode -t ansiutf8 < wgclient_{int(data["seqno"])}.conf', shell=True) + subprocess.run(f'qrencode -t ansiutf8 < wgclient_{int(data["seqno"]) + 1}.conf', shell=True) def main(): @@ -135,7 +135,7 @@ def main(): 'public_ip': wireguard.get_public_ip(), 'hub_private_key': hub_keys[0], 'hub_public_key': hub_keys[1], - 'seqno': '1', + 'seqno': '0', 'port': str(random.randrange(10000, 60000)), 'DNS': '1.1.1.1', 'oiface': wireguard.get_default_interface_name(),