travelrouter/app/network_utils.py

import re
import subprocess
import time

from config import ClientInterface

wpafile_wpa = """country=GB # Your 2-digit country code
ctrl_interface=DIR=/var/run/wpa_supplicant GROUP=netdev

network={{
    ssid="{}"
    psk="{}"
    key_mgmt=WPA-PSK
}}
"""

wpafile_nowpa = """country=GB # Your 2-digit country code
ctrl_interface=DIR=/var/run/wpa_supplicant GROUP=netdev

network={{
ssid="{}"
key_mgmt=NONE
}}
"""

wpafile_none = """country=GB # Your 2-digit country code
ctrl_interface=DIR=/var/run/wpa_supplicant GROUP=netdev
"""


def run_subprocess(cmd, check=True, delay=0):
    cmd_split = cmd.split(" ")
    output = subprocess.run(cmd_split, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=check).stdout.decode("utf-8")
    if delay > 0:
        time.sleep(delay)
    return output


def run_subprocess_interface(cmd, check=True):
    return run_subprocess(cmd.format(ClientInterface), check)


def parse_iwlist(iwlist_output, current_sid):
    """Parses iwlist scan output

    Args:
        iwlist_output (string): iwlist scan output
        current (string): Current Connected Sidd

    Returns:
        [dictionary]: Dictionary containing relavent data
    """
    data = []
    cell = []
    for line in iwlist_output.splitlines():
        if line.find("          Cell ") != -1 and cell != []:
            data.append(cell)
            cell = []
        elif line.find("Scan completed :") > 0:
            pass
        else:
            cell.append(line)

    try:
        del data[0][0]
    except:
        pass

    cells = []

    try:
        for item in data:
            cell = {}
            for line in item:
                line = line.strip()
                if line.find("ESSID:") != -1:
                    if line.partition("ESSID:")[2].strip('"') != "":
                        cell["SSID"] = line.partition("ESSID:")[2].strip('"')
                    else:
                        cell["SSID"] = "!***!"
                    if cell["SSID"] == current_sid:
                        cell["Connected"] = "Yes"
                    else:
                        cell["Connected"] = ""
                if line.partition("Quality=")[2].split("/")[0] != "" != -1:
                    cell["Signal"] = line.partition("Quality=")[2].split("/")[0] + "/70"
                if line.find("Encryption key:") != -1:
                    if line.find(":on") != -1:
                        cell["WPA"] = "WPA2"
                    else:
                        cell["WPA"] = "None"

            cells.append(cell)
    except:
        cells = []

    # Remove Blank SSID's
    cells = [cell for cell in cells if cell["SSID"] != "!***!" and cell["SSID"][0] != "\x00" and cell["SSID"][0:4] != "\\x00"]
    # Remove Blank SSID's
    cells = sorted(cells, key=lambda k: k["Signal"], reverse=True)

    seen = set()
    new_cells = []
    for d in cells:
        t = tuple(d["SSID"])
        if t not in seen:
            seen.add(t)
            new_cells.append(d)

    return new_cells


def scan_networks():
    """Scans for Networks"""
    output = run_subprocess("sudo iwgetid", check=False)

    current = output.partition("ESSID:")[2].strip().strip('"')

    output = run_subprocess("sudo iwlist {} scan".format(ClientInterface))

    scan = parse_iwlist(output, current)

    return scan


def connect_network(ssid, security, password):
    with open("/etc/network/interfaces", "r") as source:
        lines = source.readlines()
    with open("/etc/network/interfaces", "w") as source:
        for line in lines:
            source.write(
                re.sub(r"^iface {} inet manual".format(ClientInterface), "iface {} inet dhcp".format(ClientInterface), line)
            )

    run_subprocess_interface("sudo /usr/sbin/ifdown {}")

    run_subprocess("sudo /usr/bin/systemctl stop wpa_supplicant")

    with open("/etc/wpa_supplicant/wpa_supplicant.conf", "wt") as f:
        if security == "WPA2":
            f.write(wpafile_wpa.format(ssid, password))
        else:
            f.write(wpafile_nowpa.format(ssid))

    try:
        run_subprocess("sudo /usr/bin/systemctl start wpa_supplicant")
        run_subprocess_interface("sudo /usr/sbin/ifup {}")
    except:
        return False

    return True


def disconnect_network():
    run_subprocess_interface("sudo /usr/sbin/ifdown {}")

    with open("/etc/wpa_supplicant/wpa_supplicant.conf", "wt") as f:
        f.write(wpafile_none)

    run_subprocess("sudo /usr/bin/systemctl stop wpa_supplicant")

    with open("/etc/network/interfaces", "r") as sources:
        lines = sources.readlines()
    with open("/etc/network/interfaces", "w") as sources:
        for line in lines:
            sources.write(
                re.sub(r"^iface {} inet dhcp".format(ClientInterface), "iface {} inet manual".format(ClientInterface), line)
            )

    try:
        run_subprocess("sudo /usr/bin/systemctl start wpa_supplicant")
        run_subprocess_interface("sudo /usr/sbin/ifup {}")
    except:
        return False

    return True


def vpn_connected():
    if run_subprocess("nordvpn status").find("Disconnected") != -1:
        return False
    return True


def vpn_connect():
    if run_subprocess("nordvpn c").find("connected to") != -1:
        return True
    return False


def vpn_disconnect():
    if run_subprocess("nordvpn d").find("You are disconnected from NordVPN") != -1:
        return True
    return False


def killswitch_status():
    if run_subprocess("nordvpn settings").find("Kill Switch: disabled") != -1:
        return False
    return True


def killswich_enable():
    if run_subprocess("nordvpn set killswitch on", delay=2).find("Kill Switch is set to 'enabled' successfully") != -1:
        return True
    return False


def killswich_disaable():
    if run_subprocess("nordvpn set killswitch off", delay=2).find("Kill Switch is set to 'disabled' successfully") != -1:
        return True
    return False