dashboardwireguard-dashboardwireguardwg-managervpnsite-to-siteobfuscationwireguard-vpn-setupwireguard-vpnwireguard-tunnel
You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
73 lines
2.4 KiB
73 lines
2.4 KiB
import abc
|
|
from pathlib import Path
|
|
import subprocess
|
|
import shlex
|
|
|
|
|
|
class NotInstalledError(Exception):
|
|
pass
|
|
|
|
|
|
class BaseObfuscation(abc.ABC):
|
|
|
|
def __init__(self, binary_name=None, binary_path=None, algorithm=None):
|
|
|
|
assert binary_name is not None or binary_path is not None
|
|
self.binary_name = binary_name if binary_name is not None else Path(binary_path).name
|
|
self.binary_path = binary_path if binary_path else ""
|
|
self.algorithm = algorithm
|
|
|
|
def ensure_installed(self):
|
|
|
|
# Attempt to find process by path
|
|
binary = Path(self.binary_path)
|
|
if not binary.is_file():
|
|
# Did not find by path, attempt to find using which
|
|
proc_which = subprocess.Popen(["which", self.binary_name], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
|
data = [x.decode().strip() for x in proc_which.communicate() if x != b''][0]
|
|
|
|
if proc_which.returncode != 0:
|
|
raise NotInstalledError("Could not find binary '%s'" % data)
|
|
|
|
self.binary_path = data
|
|
|
|
def execute(self, *args, kill_first=False, override_command=None, stream=False, prefix=""):
|
|
if prefix != "":
|
|
prefix += ": "
|
|
|
|
if kill_first:
|
|
# TODO try to delete by full name as we dont want to kill other processes.
|
|
pattern = self.binary_name
|
|
self.execute(*[pattern], override_command="pkill")
|
|
#pattern = self.binary_path + " " + ' '.join(args)
|
|
#print(pattern)
|
|
#kill_output, kill_code = self.execute(*[pattern], override_command="pkill")
|
|
|
|
command = override_command if override_command is not None else self.binary_path
|
|
proc_which = subprocess.Popen([command] + list(args), stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
|
|
|
if not stream:
|
|
raw_data = proc_which.communicate()
|
|
|
|
data = [x.decode().strip() for x in raw_data if x != b'']
|
|
if len(data) == 0:
|
|
data = ""
|
|
else:
|
|
data = data[0]
|
|
return data, proc_which.returncode
|
|
|
|
else:
|
|
while True:
|
|
output = proc_which.stdout.readline()
|
|
if output == '' and proc_which.poll() is not None:
|
|
break
|
|
if output:
|
|
print(prefix + output.strip().decode())
|
|
rc = proc_which.poll()
|
|
return rc
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|