vpngate.py

# https://gist.github.com/Lazza/bbc15561b65c16db8ca8

import os
import requests
import base64
import time
import subprocess as sp

class VpnGate:
    def __init__(self, country, config='/tmp/vpngate.ovpn'):
        self.config = config
        self.country = country

    @staticmethod
    def get_openvpn_config(country):
        assert len(country) >= 2, "Country name is too short!"
        i = 2 if len(country) == 2 else 5

        try:
            vpn_data = requests.get("http://www.vpngate.net/api/iphone/").text.replace("\r", "")
            servers = [line.split(",") for line in vpn_data.split("\n")]
            labels = servers[1]
            labels[0] = labels[0][1:]
            servers = [s for s in servers[2:] if len(s) > 1]
        except BaseException:
            print("Cannot get VPN servers data")
            return

        desired = [s for s in servers if country.lower() in s[i].lower()]
        supported = [s for s in desired if len(s[-1]) > 0]
        winner = sorted(supported, key=lambda s: float(s[2].replace(",", ".")), reverse=True)[0]
        data = base64.b64decode(winner[-1]).decode()
        return data
    
    def connection_info(self):
        output = sp.check_output("openvpn3 sessions-list", shell=True).decode().strip()
        if output != "No sessions available":
            output = output.split("Status: ")
            
            tunnels = []
            for tunnel in output:
                tunnel = tunnel.split("\n")
                if len(tunnel) < 7: continue
                parsed = {l.split(":")[0].strip().lower().replace(" ", "_"):l.split(":")[1].strip() for l in tunnel[1:-1] if l.strip()}
                parsed.pop("owner")
                parsed['config_name'] = parsed['config_name'].split()[0]
                tunnels.append(parsed)
            return tunnels
        return []
    
    def stop_tunnel(self, config):
        if config.endswith(".ovpn"):
            sp.call(f"openvpn3 session-manage --config {config} --disconnect", shell=True)
        else:
            sp.call(f"openvpn3 session-manage --path {config} --disconnect", shell=True)
    
    def connect(self, force=False):
        old_tunnel = self.connection_info()
        if (force == False) and old_tunnel: return old_tunnel
        
        # stop all other tunnels first
        for tunnel in old_tunnel:
            self.stop_tunnel(tunnel['path'])
        
        config = self.get_openvpn_config(self.country)
        if config:
            with open(self.config, "w") as f:
                f.write(config)
            
            sp.call(f"openvpn3 session-start --config {self.config}", shell=True)
            return self.connection_info()
    
    def disconnect(self):
        for tunnel in self.connection_info():
            self.stop_tunnel(tunnel['path'])
        return self.connection_info()

if __name__ == "__main__":
    import argparse
    ap = argparse.ArgumentParser()
    ap.add_argument("-c", "--country", required=False, help="country name")
    ap.add_argument("-d", "--disconnect", action="store_true", help="to disconnec")
    args = ap.parse_args()

    vpn = VpnGate(args.country)
    if args.disconnect:
        if not vpn.connection_info():
            print("No VPN connection found")
        else:
            vpn.disconnect()
    else:
        if vpn.connection_info():
            print("VPN already connected")
        else:
            vpn.connect()