Feature: NVRAM connect and kill support for Asuswrt-merlin

README.md

@@ -23,7 +23,7 @@ A python3 script (systemd service as well) to manage OpenVPN connections. Create
     "Tor Over VPN" --tor, "Double VPN" --double, "Anti DDos" --anti-ddos support.
 -   Desktop notification are shown when VPN connects and disconnects. (needs to run without sudo)
 -   Auto retry if \[soft,auth-failure\] received, auto failover to next best server if connection dies.
--   NVRAM write support for Asuswrt-merlin
+-   NVRAM support for Asuswrt-merlin
 -   Pass through OpenVPN options, e.g. openpyn uk -o '--status /var/log/status.log --log /var/log/log.log'
 -   Logs are stored in '/var/log/openpyn/' for information and troubleshooting.
 -   Temporarily disable IPv6 to prevent leakage (when using -f).
@@ -246,10 +246,16 @@ sudo openpyn -x # optionally --allow 22 if using as SSH server
 openpyn --update
 ```
 
--   To save the least busy NordVPN server in US to NVRAM for "OpenVPN Client 5".
+-   To quickly connect to United States "OpenVPN Client 5" (Asuswrt-Merlin).
 
 ```bash
-openpyn us --nvram 5 # (ASUSWRT-Merlin)
+openpyn us --nvram 5
+```
+
+-   To kill "OpenVPN Client 5" (Asuswrt-Merlin).
+
+```bash
+openpyn -k --nvram 5
 ```
 
 ## Usage Options
@@ -299,7 +305,7 @@ optional arguments:
                         server
   -n NVRAM, --nvram NVRAM
                         Specify client to save configuration to NVRAM
-                        (ASUSWRT-Merlin)
+                        (Asuswrt-Merlin)
 
 OpenVPN Options:
   Configurable Options Being Passed Downed To OpenVPN

openpyn/asus.py

@@ -71,7 +71,7 @@ def run(server, client, options=None, rgw=None, comp=None, adns=None, tcp=False,
     service = "client"
 
     for key, value in extracted_info.items():
-        write(c, key, value, unit, service, test)
+        write(key, value, unit, service, test)
 
     extracted_info = dict(extracted_info)
     if T_CLIENT in extracted_info:
@@ -86,28 +86,109 @@ def run(server, client, options=None, rgw=None, comp=None, adns=None, tcp=False,
     service = "client"
 
     for key, value in extracted_info.items():
-        write(c, key, value, unit, service, test)
+        write(key, value, unit, service, test)
 
     # 'vpn_upload_unit'
     key = T_CLIENT
     value = client
     unit = ""
     service = "upload"
 
-    write(c, key, value, unit, service, test)
+    write(key, value, unit, service, test)
 
 
-def write(c, key, value, unit, service, test=False):
+def write(key, value, unit, service, test=False):
     argument1 = "vpn" + "_" + service + unit + "_" + key
     argument2 = argument1 + "=" + value
     try:
-        c.pprint("/bin/nvram" + " " + "get" + " " + argument1)
+        pprint("/bin/nvram" + " " + "get" + " " + argument1)
         if not test:
             current = subprocess.run(["/bin/nvram", "get", argument1], check=True, stdout=subprocess.PIPE).stdout
             if current.decode("utf-8").strip() == value:
                 return
-        c.pprint("/bin/nvram" + " " + "set" + " " + argument2)
+        pprint("/bin/nvram" + " " + "set" + " " + argument2)
         if not test:
             subprocess.run(["sudo", "-u", sudo_user, "/bin/nvram", "set", argument2], check=True)
     except subprocess.CalledProcessError as e:
         raise RuntimeError(e.output)
+
+
+def connect(unit, test=False):
+    argument1 = "vpn" + "_" + "client" + unit + "_" + "state"
+    argument2 = "start" + "_" + "vpnclient" + unit
+    try:
+        pprint("/bin/nvram" + " " + "get" + " " + argument1)
+        if not test:
+            current = subprocess.run(["/bin/nvram", "get", argument1], check=True, stdout=subprocess.PIPE).stdout
+            if current.decode("utf-8").strip() in {"1", "2"}:  # Connected
+                return
+        pprint("/sbin/service" + " " + argument2)
+        if not test:
+            subprocess.run(["sudo", "-u", sudo_user, "/sbin/service", argument2], check=True)
+    except subprocess.CalledProcessError as e:
+        raise RuntimeError(e.output)
+
+
+def disconnect(unit, test=False):
+    argument1 = "vpn" + "_" + "client" + unit + "_" + "state"
+    argument2 = "stop" + "_" + "vpnclient" + unit
+    try:
+        pprint("/bin/nvram" + " " + "get" + " " + argument1)
+        if not test:
+            current = subprocess.run(["/bin/nvram", "get", argument1], check=True, stdout=subprocess.PIPE).stdout
+            if not current.decode("utf-8").strip() in {"1", "2"}:  # Disconnected
+                return
+        pprint("/sbin/service" + " " + argument2)
+        if not test:
+            subprocess.run(["sudo", "-u", sudo_user, "/sbin/service", argument2], check=True)
+    except subprocess.CalledProcessError as e:
+        raise RuntimeError(e.output)
+
+
+def state(unit, test=False) -> bool:
+    argument1 = "vpn" + "_" + "client" + unit + "_" + "state"
+    try:
+        pprint("/bin/nvram" + " " + "get" + " " + argument1)
+        if not test:
+            client_state = subprocess.run(["/bin/nvram", "get", argument1], check=True, stdout=subprocess.PIPE).stdout
+            code = client_state.decode("utf-8").strip()
+            if code == "1":
+                logger.success("Connecting...")
+            elif code == "2":
+                logger.success("Connected")
+            elif code == "-1":
+                return errno(unit, test)
+
+        return 0
+    except subprocess.CalledProcessError as e:
+        raise RuntimeError(e.output)
+
+
+def errno(unit, test=False) -> bool:
+    argument1 = "vpn" + "_" + "client" + unit + "_" + "errno"
+    try:
+        pprint("/bin/nvram" + " " + "get" + " " + argument1)
+        if not test:
+            client_errno = subprocess.run(["/bin/nvram", "get", argument1], check=True, stdout=subprocess.PIPE).stdout
+            code = client_errno.decode("utf-8").strip()
+            if code == "1":
+                logger.error("Error - IP conflict!")
+            elif code == "2":
+                logger.error("Error - Routing conflict!")
+            elif code == "4":
+                logger.error("Error - SSL/TLS issue!")
+            elif code == "5":
+                logger.error("Error - DH issue!")
+            elif code == "6":
+                logger.error("Error - Authentication failure!")
+            else:
+                logger.error("Error - check configuration!")
+
+        return 1
+    except subprocess.CalledProcessError as e:
+        raise RuntimeError(e.output)
+
+
+def pprint(msg, debug=False):
+    if debug:
+        logger.debug(msg)

openpyn/openpyn.py

@@ -17,6 +17,7 @@
 import zipfile
 from email.utils import parsedate
 from pathlib import Path
+from time import sleep
 from typing import List
 from typing import Set
 
@@ -410,6 +411,12 @@ def run(init: bool, server: str, country_code: str, country: str, area: str, tcp
             systemd.update_service(openpyn_options, run=True)
 
     elif kill:
+        if detected_os == "linux":
+            if asuswrt_os:
+                if nvram:
+                    asus.disconnect(nvram, test)
+                    sleep(2)
+                    return asus.state(nvram, test)
         kill_all()
 
     elif kill_flush:
@@ -534,7 +541,9 @@ def run(init: bool, server: str, country_code: str, country: str, area: str, tcp
                     check_config_files()
                     asus.run(aserver, nvram, openvpn_options, "All", "adaptive", "Strict", tcp, test)
                     logger.success("SAVED SERVER " + aserver + " ON PORT " + port + " TO NVRAM " + nvram)
-                    return 0
+                    asus.connect(nvram, test)
+                    sleep(2)
+                    return asus.state(nvram, test)
 
                 if test:
                     logger.success(
@@ -601,7 +610,9 @@ def run(init: bool, server: str, country_code: str, country: str, area: str, tcp
                 check_config_files()
                 asus.run(server, nvram, openvpn_options, "All", "adaptive", "Strict", tcp, test)
                 logger.success("SAVED SERVER " + server + " ON PORT " + port + " TO NVRAM " + nvram)
-                return 0
+                asus.connect(nvram, test)
+                sleep(2)
+                return asus.state(nvram, test)
 
             if test:
                 logger.success(