Baionet-Wireguard-Config/create_new_wireguard_account

#!/usr/bin/env python3

import configparser
import datetime
import os
import smtplib
import subprocess
import sys
import tempfile
import textwrap
from email.mime.multipart   import MIMEMultipart
from email.mime.application import MIMEApplication
from email.mime.text        import MIMEText

class Email:
    """
    Not really necessary, but I keep on forgetting most of an email arguments
    and I'm tired of debugging this class of error... Too bad we don't have any
    record type in this language.

    PS: I hate python.
    """
    def __init__(self, username, passwd, from_addr, to_addr, server, port):
        self.username  = username
        self.passwd    = passwd
        self.from_addr = from_addr
        self.to_addr   = to_addr
        self.server    = server
        self.port      = port

class Config:
    """
    Same as Email
    """
    def __init__(self, config):
        self.smtp_user = config['smtp']['username']
        self.smtp_pass = config['smtp']['password']
        self.smtp_server = config['smtp']['server']
        self.smtp_port = int(config['smtp']['port'])
        self.smtp_from = config['smtp']['from']
        self.wg_service = config['wireguard-service']['name']
        self.pubkey = config['wireguard-service']['pubkey']
        self.endpoint = config['wireguard-service']['endpoint']

def _run_cmd(cmd):
    print("> $ %s" % (cmd))
    subprocess.run(cmd, shell=True, check=True)

def check_env ():
    """
    Checks if wireguard is correctly installed, the script is run as
    root and the member id is correct.
    """
    wgInstalled = os.system("wg")
    if os.geteuid() != 0:
        print("On a besoin des droits root pour créer un accès VPN.")
        print("Relancez la commande préfixée d'un sudo.")
        raise Exception("Got root?")
    if wgInstalled != 0:
        print("Wireguard ne semble pas être installé sur votre système.")
        print("Il faudrait installer wireguard-tools et wireguard-dkms avant"
              " d'utiliser ce script")
        raise Exception("Install wg")
    member_email = input("EMail du nouveau membre: ")
    member_id = int(input("Numéro d'adhérant du nouveau membre: "))
    if member_id < 2 or member_id > 253:
        print("On est parti du principe que les IDs des membres commencent à 1.")
        print("Ah, aussi, pour le moment, on est aussi partis du principe que ça "
              "s'arrête à 253.")
        print("Si on a plus de 253 membres, premièrement, FÉLICITATIONS venant "
              "du Félix du passé. Par contre, il faut repenser l'adressage des "
              "IPs du VPN maintenant :( Ranson du succès j'immagine.")
        raise Exception("Wrong member_id")
    return (member_email, member_id)

def gen_wg_keys (temp_dir):
    """
    Generates both the private and the public wireguard key of the new member.
    """
    pubkey_path = os.path.join(temp_dir, "pub.key")
    privkey_path = os.path.join(temp_dir, "priv.key")
    psk_path = os.path.join(temp_dir, "psk.key")
    _run_cmd("wg genkey | tee %s | wg pubkey > %s" % (privkey_path, pubkey_path))
    _run_cmd("wg genpsk > %s" % (psk_path))
    return (privkey_path, pubkey_path, psk_path)

def is_duplicate_entry_wg_conf (member_id, config_file):
    """
    Look for a potential wireguard duplicate entry.

    Note: wg config format is not really a init file because of the multiple
    init entries. Hence, we cannot use a proper parser to to the check. We'll
    only try to pattern match the ip addr.
    """
    with open(config_file, "r") as wg_conf_file:
        return ("10.0.0.{}".format(member_id) in wg_conf_file.read())

def update_wg_config (member_id, config_file, pubkey_path, psk_path):
    """
    Generate the wireguard peer entry for this new member.
    """
    wg_new_peer = textwrap.dedent('''
    [Peer]
    PublicKey = pubksubs
    PresharedKey = psksubs
    AllowedIPs = 10.0.0.{0}/24, fd00::{0}/64
    ''').format(member_id)
    with open(config_file, "a") as wg_conf_file:
        wg_conf_file.write(wg_new_peer)
    _run_cmd('sed -i "s|pubksubs|"$(cat %s)"|g" %s' % (pubkey_path, config_file))
    _run_cmd('sed -i "s|psksubs|"$(cat %s)"|g" %s' % (psk_path, config_file))

def generate_wg_quick_client_config(peer_priv_key, member_id,
                                    server_pub_key, psk, server_endpoint):
    return textwrap.dedent('''\
        [Interface]
        PrivateKey = {0}
        Address = 10.0.0.{1}/24, fd00::{1}/64
        SaveConfig = false
        DNS = 80.67.169.12, 80.67.169.40, 2001:910:800::12, 2001:910:800::40

        [Peer]
        PublicKey = {2}
        PresharedKey = {3}
        AllowedIPs = 0.0.0.0/0, ::/0
        Endpoint = {4}
        ''').format(peer_priv_key.strip(), member_id, server_pub_key, psk.strip(),\
                    server_endpoint)

def send_email(email, wg_client_config):
    """
    Send the private key by email.

    email:
        - username
        - passwd
        - from_addr
        - to_addr
        - server
    """
    from_addr      = 'bureau@baionet.fr'
    password       = email.passwd
    msg            = MIMEMultipart()
    msg['Subject'] = "Votre acces VPN Baionet"
    msg['Date']    = str(datetime.datetime.now())
    msg['From']    = email.from_addr
    msg['To']      = email.to_addr
    body           = textwrap.dedent('''
    blahblah, cf le wiki blahblahblah
    ''')
    part = MIMEApplication(
            wg_client_config.encode("utf-8"),
            Name="wg0.conf")
    part['Content-Disposition'] = 'attachment; filename=wg0.conf'
    msg.attach(part)
    username = email.username
    server   = smtplib.SMTP(email.server, email.port)
    server.ehlo()
    server.starttls()
    server.login(email.username, email.passwd)
    server.sendmail(email.from_addr, email.to_addr, msg.as_string())
    server.quit()
    print("[+] E-Mail envoyé à %s." % email.to_addr)

# Main Function
# =============
# *************
# =============
# 1- Parse email/member id.
# 2- Génère les clés wireguard.
# 3- Crée/déploie la configuration wireguard.
# 5- Envoie la clé à l'utilisateur (email/manuellement)
if __name__ == '__main__':
    cp = configparser.ConfigParser()
    config_file = "/etc/wireguard/wg0.conf"
    cp.read('/etc/wireguard/wg-create-account.ini')
    config = Config(cp)
    (member_email, member_id) = check_env()
    with tempfile.TemporaryDirectory() as temp_dir:
        try:
            print("[+] Génération des clés wireguard")
            (privkey_path, pubkey_path, psk_path) = gen_wg_keys(temp_dir)
            print("[+] Envoi de la clé privée au nouveau membre")
            print("Deux solutions ici:")
            print("1- On envoie la configuration du nouveau membre en ligne. (automatique)")
            print("2- On utilise une autre méthode pour passer la configuration au nouveau membre. (manuel)")
            print("")
            print("Suivant votre modèle de menace, envoyer la clé privée par e-mail peut ou peut ne pas être une bonne idée.")
            use_email = input("Envoyer la configuration (contenant la clé privée) par email? (O/n)")
            if use_email.strip().lower() != "n" :
                with open(privkey_path, "r") as pkh:
                   peer_privkey = pkh.read()
                with open(psk_path, "r") as pskh:
                   peer_psk = pskh.read()
                email = Email(config.smtp_user, config.smtp_pass, config.smtp_from,\
                              member_email, config.smtp_server, config.smtp_port)
                send_email(email, generate_wg_quick_client_config(peer_privkey, member_id, config.pubkey,\
                           peer_psk, config.endpoint))
            else:
                print("Mode utilisateur avancé")
                print("=======================")
                print("À vous de vous débrouiller pour passer les clés/config à l'utilisateur")
                print("Clé privée: %s" % (privkey_path))
                print("Clé pré-partagée (psk): %s" % (psk_path))
                print("Clé publique (psk): %s" % (pubkey_path))
                input("Appuyez sur entrée pour continuer (les clés privées seront détruites): ")
        except Exception as e:
            print("ERREUR: erreur lors de la génération/transfert de la clé:")
            print(e)
            print("Si vous ne comprenez pas le problème, transférez le message d'erreur complet à la liste "
                  "de diffusion technique.")
            print("Pas de panique: on n'a pas touché à la configuration du serveur de wireguard, "
                  "rien n'a cassé.")
            sys.exit(1)
        try:
            print("[+] Modification de la configuration wireguard")
            if not is_duplicate_entry_wg_conf(member_id, config_file):
                update_wg_config(member_id, config_file, pubkey_path, psk_path)
            else:
                print("Le membre {} semble déja avoir un compte VPN.".format(member_id))
                print("Veuillez contacter la liste de diffusion technique "
                      " si son compte necessite une ré-activation.")
                sys.exit(1)
            print("[+] Chargement de la nouvelle interface réseau")
            _run_cmd("systemctl restart %s" % (config.wg_service))
            print("[+] Nettoyage des clés")
            _run_cmd("shred -u %s %s %s" % (privkey_path, pubkey_path, psk_path))
            print("[+] COMPTE CRÉE AVEC SUCCÈS")
        except Exception as e:
            print("ERREUR CRITIQUE: attention, la configuration wireguard est cassée, il y a probablement urgence.")
            print(e)
            print("Si vous ne savez pas quoi faire pour régler le problème, contactez en urgence la "
                  "liste de diffusion technique en joignant le message d'erreur ci-dessus.")
            print("Je le répète: le serveur VPN est probablement cassé, c'est une urgence.")
            sys.exit(1)