summaryrefslogtreecommitdiff
path: root/wrappers/python/eduvpncommon
diff options
context:
space:
mode:
authorjwijenbergh <jeroenwijenbergh@protonmail.com>2022-05-03 16:33:37 +0200
committerjwijenbergh <jeroenwijenbergh@protonmail.com>2022-05-03 16:33:37 +0200
commit657776055cd07c1f9279e982fbfef88dca1ca71b (patch)
tree4952be2ac2f999e74446601c5b138566a909e5e2 /wrappers/python/eduvpncommon
parenta94316159904e37ada2b932757a2594ef796efe0 (diff)
Python: Place py files in src
Diffstat (limited to 'wrappers/python/eduvpncommon')
-rw-r--r--wrappers/python/eduvpncommon/__init__.py60
-rw-r--r--wrappers/python/eduvpncommon/main.py123
2 files changed, 0 insertions, 183 deletions
diff --git a/wrappers/python/eduvpncommon/__init__.py b/wrappers/python/eduvpncommon/__init__.py
deleted file mode 100644
index 1df305b..0000000
--- a/wrappers/python/eduvpncommon/__init__.py
+++ /dev/null
@@ -1,60 +0,0 @@
-from ctypes import *
-from collections import defaultdict
-import pathlib
-import platform
-
-_lib_prefixes = defaultdict(
- lambda: "lib",
- {
- "windows": "",
- },
-)
-
-_lib_suffixes = defaultdict(
- lambda: ".so",
- {
- "windows": ".dll",
- "darwin": ".dylib",
- },
-)
-
-_os = platform.system().lower()
-
-_libname = "eduvpn_common"
-_libfile = f"{_lib_prefixes[_os]}{_libname}{_lib_suffixes[_os]}"
-# Library should have been copied to the lib/ folder
-lib = cdll.LoadLibrary(str(pathlib.Path(__file__).parent / "lib" / _libfile))
-
-
-class DataError(Structure):
- _fields_ = [("data", c_void_p), ("error", c_void_p)]
-
-
-VPNStateChange = CFUNCTYPE(None, c_char_p, c_char_p, c_char_p)
-
-# Exposed functions
-lib.Connect.argtypes, lib.Connect.restype = [c_char_p, c_char_p], DataError
-lib.Deregister.argtypes, lib.Deregister.restype = [c_char_p], c_void_p
-lib.Register.argtypes, lib.Register.restype = [c_char_p, c_char_p, VPNStateChange, c_int], c_void_p
-lib.GetOrganizationsList.argtypes, lib.GetOrganizationsList.restype = [c_char_p], DataError
-lib.GetServersList.argtypes, lib.GetServersList.restype = [c_char_p], DataError
-lib.CancelOAuth.argtypes, lib.CancelOAuth.restype = [c_char_p], c_void_p
-lib.SetProfileID.argtypes, lib.SetProfileID.restype = [c_char_p, c_char_p], c_void_p
-# We have to use c_void_p instead of c_char_p to free it properly
-# See https://stackoverflow.com/questions/13445568/python-ctypes-how-to-free-memory-getting-invalid-pointer-error
-lib.FreeString.argtypes, lib.FreeString.restype = [c_void_p], None
-
-
-def GetPtrString(ptr):
- if ptr:
- string = cast(ptr, c_char_p).value
- lib.FreeString(ptr)
- if string:
- return string.decode()
- return ""
-
-
-def GetDataError(data_error):
- data = GetPtrString(data_error.data)
- error = GetPtrString(data_error.error)
- return data, error
diff --git a/wrappers/python/eduvpncommon/main.py b/wrappers/python/eduvpncommon/main.py
deleted file mode 100644
index 5474ade..0000000
--- a/wrappers/python/eduvpncommon/main.py
+++ /dev/null
@@ -1,123 +0,0 @@
-from . import lib, VPNStateChange, GetDataError, GetPtrString
-from ctypes import *
-from enum import Enum
-
-
-class StateType(Enum):
- Enter = 1
- Leave = 2
-
-
-# Registers the python app with the Go code
-# name: The name of the app to be registered
-# state_callback: The callback to trigger whenever a state is changed
-def Register(name, config_directory, state_callback, debug):
- name_bytes = name.encode("utf-8")
- dir_bytes = config_directory.encode("utf-8")
- ptr_err = lib.Register(name_bytes, dir_bytes, state_callback, debug)
- err_string = GetPtrString(ptr_err)
- return err_string
-
-def CancelOAuth(name):
- name_bytes = name.encode("utf-8")
- ptr_err = lib.CancelOAuth(name_bytes)
- err_string = GetPtrString(ptr_err)
- return err_string
-
-def Deregister(name):
- name_bytes = name.encode("utf-8")
- ptr_err = lib.Deregister(name_bytes)
- err_string = GetPtrString(ptr_err)
- return err_string
-
-def GetDiscoServers(name):
- name_bytes = name.encode("utf-8")
- servers, serversErr = GetDataError(lib.GetServersList(name_bytes))
- organizations, organizationsErr = GetDataError(lib.GetOrganizationsList(name_bytes))
- return servers, serversErr, organizations, organizationsErr
-
-
-def Connect(name, url):
- name_bytes = name.encode("utf-8")
- url_bytes = url.encode("utf-8")
- data_error = lib.Connect(name_bytes, url_bytes)
- return GetDataError(data_error)
-
-
-# This has to be global as otherwise the callback is not alive
-callback_function = None
-
-
-def register_callback(eduvpn):
- global callback_function
- callback_function = VPNStateChange(
- lambda old_state, new_state, data: eduvpn.callback(
- old_state.decode(), new_state.decode(), data.decode()
- )
- )
-
-
-def SetProfileID(name, profile_id) -> str:
- name_bytes = name.encode("utf-8")
- profile_bytes = profile_id.encode("utf-8")
- error_string = lib.SetProfileID(name_bytes, profile_bytes)
- return GetPtrString(error_string)
-
-
-class EduVPN(object):
- def __init__(self, name, config_directory):
- self.event_handler = EventHandler()
- self.name = name
- self.config_directory = config_directory
- register_callback(self)
-
- def cancel_oauth(self) -> str:
- return CancelOAuth(self.name)
-
- def deregister(self) -> str:
- return Deregister(self.name)
-
- def register(self, debug=False) -> bool:
- return Register(self.name, self.config_directory, callback_function, debug) == ""
-
- def get_disco(self):
- return GetDiscoServers(self.name)
-
- def connect(self, url):
- return Connect(self.name, url)
-
- @property
- def event(self):
- return self.event_handler
-
- def callback(self, old_state, new_state, data):
- self.event.run(old_state, new_state, data)
-
- def set_profile(self, profile_id) -> str:
- return SetProfileID(self.name, profile_id)
-
-
-class EventHandler(object):
- def __init__(self):
- self.handlers = {}
-
- def on(self, state, state_type):
- def wrapped_f(func):
- if (state, state_type) not in self.handlers:
- self.handlers[(state, state_type)] = []
- self.handlers[(state, state_type)].append(func)
- return func
-
- return wrapped_f
-
- def run_state(self, state, state_type, data):
- if (state, state_type) not in self.handlers:
- return
- for func in self.handlers[(state, state_type)]:
- func(data)
-
- def run(self, old_state, new_state, data):
- if old_state == new_state:
- return
- self.run_state(old_state, StateType.Leave, data)
- self.run_state(new_state, StateType.Enter, data)