Baionet-Wireguard-Config/create_new_wireguard_account

#!/usr/bin/env python3

import configparser
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):
        self.username  = username
        self.passwd    = passwd
        self.from_addr = from_addr
        self.to_addr   = to_addr
        self.server    = server

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_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 = %PUBKEY%
    PresharedKey = %PSK%
    AllowedIPs = 10.0.0.{1}/24, fd00::{1}/64
    ''').format(member_id)
    with open(config_file, "a") as wg_conf_file:
        wg_conf_file.write(wg_config)
    _run_cmd('sed -i "s/%PUBKEY%/$(cat %s)/" "%s"' % (pubkey_path, config_file))
    _run_cmd('sed -i "s/%PSK%/$(cat %s)/" "%s"' % (psk_path, config_file))

def generate_wg_quick_client_config(peer_priv_key, member_id,
                                    server_pub_key, psk, server_endpoint):
    template = 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, member_id, server_pub_key, psk, server_endpoint)

def send_mail(email, wgconfig_path):
    """
    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']    = formatdate(localtime=True)
    msg['From']    = email.from_addr
    msg['To']      = [email.to_addr]
    body           = textwrap.dedent('''
    blahblah, cf le wiki blahblahblah
    ''')
    msg.attach([MIMEText(body), MIMEText(config)])
    with open(wg_client_path, "rb") as f:
        part = MIMEApplication(
                f.read(),
                Name=os.path.basename(f))
        part['Content-Disposition'] = 'attachment; filename=%s' % basename(f)
        msg.attach(part)
    username = email.username
    server   = smtplib.SMTP(email.server)
    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()
    cp.read('/etc/wireguard/wg-create.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() == "o" :
                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)
                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é.")
        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.")