Small script that helps with the wireguard configuration

wg-configurator.py

#!/usr/bin/env python

import os
import sys
import re
import click
import subprocess


class WGConfigurator:

    def __init__(self):
        pass

    def generate_private_key(self):
        genkey = subprocess.Popen('wg genkey', stdout=subprocess.PIPE, shell=True)
        client_private_key, error = genkey.communicate()
        if error:
            print(f"An error with wg genkey occured {error}")
            sys.exit(1)
        return client_private_key.decode("utf-8").strip()

    def generate_public_key(self, private_key):
        echo = subprocess.Popen(f'echo {private_key}', stdout=subprocess.PIPE, shell=True)
        pubkey = subprocess.Popen('wg pubkey', stdin=echo.stdout, stdout=subprocess.PIPE, shell=True)
        client_public_key, error = pubkey.communicate()
        if error:
            print(f"An error with wg pubkey occured {error}")
            sys.exit(1)
        return client_public_key.decode("utf-8").strip()

    def get_server_public_key(self, conf="/etc/wireguard/wg0.conf"):
        with open(conf) as f:
            data = f.read()
            private_key = re.search(r"PrivateKey = (.*)", data)
            if not private_key:
                print(f"Error, couldn't read server private key from config file {conf} ")
                sys.exit(1)
            private_key = private_key.group(1)
        return self.generate_public_key(private_key)

    def get_next_ip(self, conf="/etc/wireguard/wg0.conf"):
        with open(conf) as f:
            data = f.read()
            client_ips = re.findall(r"AllowedIPs = (\d+.\d+.\d+.)(\d+)(\/32)",data)
            if not client_ips:
                print(f"Error, couldn't get next ip from config file {conf} ")
                sys.exit(1)
            ip = sorted(client_ips,key=lambda x: int(x[1]), reverse=True)[0]
            return f"{ip[0]}{int(ip[1])+1}{ip[2]}"

    def generate_server_config(self, address, port, interface):
        server_config = "# Server Config\n"
        server_config += "[Interface]\n"
        server_config += f"Address = {address}\n"
        private_key = self.generate_private_key()
        server_config += f"PrivateKey = {private_key}\n"
        server_config += f"ListenPort = {port}\n"
        address = ".".join(address.split(".")[:-1])
        server_config += f"PostUp = iptables -t nat -A POSTROUTING -o {interface} -s {address}.0/24 -j MASQUERADE\n"
        server_config += f"PostDown = iptables -t nat -D POSTROUTING -o {interface} -s {address}.0/24 -j MASQUERADE\n"
        return server_config

    def generate_client_config(self, endpoint, port, allowed_ips, ip=None, name=None):
        client_config = "# Client Config\n"
        client_config += "[Interface]\n"
        if not ip:
            ip = self.get_next_ip()
        client_config += f"Address = {ip}\n"
        client_private_key = self.generate_private_key()
        client_config += f"PrivateKey = {client_private_key}\n"
        client_config += f"\n"
        client_config += f"# VPN Server\n"
        client_config += f"[Peer]\n"
        server_public_key = self.get_server_public_key()
        client_config += f"PublicKey = {server_public_key}\n"
        client_config += f"Endpoint = {endpoint}:{port}\n"
        client_config += f"AllowedIPs = {allowed_ips}\n"
        client_config += f"PersistentKeepalive = 21\n"

        server_config = "[Peer]\n"
        if name:
            server_config += f"# {name}\n"
        client_public_key = self.generate_public_key(client_private_key)
        server_config += f"PublicKey = {client_public_key}\n"
        server_config += f"AllowedIPs = {ip}\n"

        return client_config, server_config

@click.group()
def cli():
    pass

@click.command()
@click.option('-a', '--address', required=True, help='ip address for the server tunnel, for example 192.168.114.1')
@click.option('-i', '--interface', required=True, help='interface for iptables rules, for example ens18')
@click.option('-p', '--port', default=51820, help='port on which the server listens for connections, for example 51820')
def server(address, interface, port):
    wgc = WGConfigurator()
    config = wgc.generate_server_config(address,port,interface)
    print(f"# {'='*40}")
    print("# Put this output into a file, wg0.conf for example and put it into /et/wireguard/wg0.conf")
    print(f"# {'='*40}\n")
    print(config)

@click.command()
@click.option('-e', '--endpoint', required=True, help='Endpoint, the IP or domain where the VPN server listens, for example vpn.mydomain.com')
@click.option('-p', '--port', default=51820, help='port on which the server listens for connections, for example 51820')
@click.option('-a', '--allowed_ips', required=True, help='The allowed IPs for the client, for example 192.168.1.0/24')
@click.option('-i', '--ip', help='The ip of this client, for example 192.168.114.2/32. If not passed the next ip will be calculated from the server config.')
@click.option('-n', '--name', help='A name for this client, will be put as comment into the config.')
def client(endpoint, port, allowed_ips, ip, name):
    wgc = WGConfigurator()
    client_config, server_config = wgc.generate_client_config(endpoint, port, allowed_ips, ip, name)
    print(f"# {'='*40}")
    print("# Put this output into a file, wg0.conf for example for use on the client")
    print(f"# {'='*40}\n")
    print(client_config)
    print(f"# {'='*40}")
    print("# Append this output to your server config, wg0.conf for example")
    print(f"# {'='*40}\n")
    print(server_config)

cli.add_command(server)
cli.add_command(client)

if __name__ == "__main__":
    cli()