From 07c8ddc0b298894f3da77fe3f8aca1b18189b79b Mon Sep 17 00:00:00 2001 From: marios8543 Date: Tue, 17 Oct 2023 23:52:18 +0300 Subject: Experimental support for async method calls --- backend/src/plugin/method_call_request.py | 29 +++++++ backend/src/plugin/plugin.py | 55 +++++++++++++ backend/src/plugin/sandboxed_plugin.py | 131 ++++++++++++++++++++++++++++++ 3 files changed, 215 insertions(+) create mode 100644 backend/src/plugin/method_call_request.py create mode 100644 backend/src/plugin/plugin.py create mode 100644 backend/src/plugin/sandboxed_plugin.py (limited to 'backend/src/plugin') diff --git a/backend/src/plugin/method_call_request.py b/backend/src/plugin/method_call_request.py new file mode 100644 index 00000000..8d93a6cc --- /dev/null +++ b/backend/src/plugin/method_call_request.py @@ -0,0 +1,29 @@ +from typing import Any, TypedDict +from uuid import uuid4 +from asyncio import Event + +class SocketResponseDict(TypedDict): + id: str + success: bool + res: Any + +class MethodCallResponse: + def __init__(self, success: bool, result: Any) -> None: + self.success = success + self.result = result + +class MethodCallRequest: + def __init__(self) -> None: + self.id = str(uuid4()) + self.event = Event() + self.response: MethodCallResponse + + def set_result(self, dc: SocketResponseDict): + self.response = MethodCallResponse(dc["success"], dc["res"]) + self.event.set() + + async def wait_for_result(self): + await self.event.wait() + if not self.response.success: + raise Exception(self.response.result) + return self.response \ No newline at end of file diff --git a/backend/src/plugin/plugin.py b/backend/src/plugin/plugin.py new file mode 100644 index 00000000..b7d0a44a --- /dev/null +++ b/backend/src/plugin/plugin.py @@ -0,0 +1,55 @@ +from json import dumps, load, loads +from logging import getLogger +from os import path + +from .sandboxed_plugin import SandboxedPlugin +from .method_call_request import MethodCallRequest +from ..localplatform.localsocket import LocalSocket + +from typing import Any, Dict + +class PluginWrapper: + def __init__(self, file: str, plugin_directory: str, plugin_path: str) -> None: + self.file = file + self.plugin_path = plugin_path + self.plugin_directory = plugin_directory + + self.version = None + + json = load(open(path.join(plugin_path, plugin_directory, "plugin.json"), "r", encoding="utf-8")) + if path.isfile(path.join(plugin_path, plugin_directory, "package.json")): + package_json = load(open(path.join(plugin_path, plugin_directory, "package.json"), "r", encoding="utf-8")) + self.version = package_json["version"] + + self.name = json["name"] + self.author = json["author"] + self.flags = json["flags"] + self.passive = not path.isfile(self.file) + + self.log = getLogger("plugin") + self.method_call_requests: Dict[str, MethodCallRequest] = {} + self.sandboxed_plugin = SandboxedPlugin(self.name, self.passive, self.flags, self.file, self.plugin_directory, self.plugin_path, self.version, self.author) + #TODO: Maybe somehow make LocalSocket not require on_new_message to make this more clear + self.socket = LocalSocket(self.sandboxed_plugin.on_new_message) + self.sandboxed_plugin.start(self.socket) + + def __str__(self) -> str: + return self.name + + async def response_listener(self): + while True: + line = await self.socket.read_single_line() + if line != None: + res = loads(line) + self.method_call_requests.pop(res["id"]).set_result(res) + + async def execute_method(self, method_name: str, kwargs: Dict[Any, Any]): + if self.passive: + raise RuntimeError("This plugin is passive (aka does not implement main.py)") + + request = MethodCallRequest() + await self.socket.get_socket_connection() + await self.socket.write_single_line(dumps({ "method": method_name, "args": kwargs, "id": request.id }, ensure_ascii=False)) + self.method_call_requests[request.id] = request + + return await request.wait_for_result() \ No newline at end of file diff --git a/backend/src/plugin/sandboxed_plugin.py b/backend/src/plugin/sandboxed_plugin.py new file mode 100644 index 00000000..7c96da01 --- /dev/null +++ b/backend/src/plugin/sandboxed_plugin.py @@ -0,0 +1,131 @@ +from os import path, environ +from signal import SIGINT, signal +from importlib.util import module_from_spec, spec_from_file_location +from json import dumps, loads +from logging import getLogger +import multiprocessing +from sys import exit, path as syspath +from traceback import format_exc +from asyncio import (get_event_loop, new_event_loop, + set_event_loop, sleep) + +from .method_call_request import SocketResponseDict +from ..localplatform.localsocket import LocalSocket +from ..localplatform.localplatform import setgid, setuid, get_username, get_home_path +from ..customtypes import UserType +from .. import helpers + +from typing import List + +class SandboxedPlugin: + def __init__(self, + name: str, + passive: bool, + flags: List[str], + file: str, + plugin_directory: str, + plugin_path: str, + version: str|None, + author: str) -> None: + self.name = name + self.passive = passive + self.flags = flags + self.file = file + self.plugin_path = plugin_path + self.plugin_directory = plugin_directory + self.version = version + self.author = author + + self.log = getLogger("plugin") + + def _init(self, socket: LocalSocket): + try: + signal(SIGINT, lambda s, f: exit(0)) + + set_event_loop(new_event_loop()) + if self.passive: + return + setgid(UserType.ROOT if "root" in self.flags else UserType.HOST_USER) + setuid(UserType.ROOT if "root" in self.flags else UserType.HOST_USER) + # export a bunch of environment variables to help plugin developers + environ["HOME"] = get_home_path(UserType.ROOT if "root" in self.flags else UserType.HOST_USER) + environ["USER"] = "root" if "root" in self.flags else get_username() + environ["DECKY_VERSION"] = helpers.get_loader_version() + environ["DECKY_USER"] = get_username() + environ["DECKY_USER_HOME"] = helpers.get_home_path() + environ["DECKY_HOME"] = helpers.get_homebrew_path() + environ["DECKY_PLUGIN_SETTINGS_DIR"] = path.join(environ["DECKY_HOME"], "settings", self.plugin_directory) + helpers.mkdir_as_user(path.join(environ["DECKY_HOME"], "settings")) + helpers.mkdir_as_user(environ["DECKY_PLUGIN_SETTINGS_DIR"]) + environ["DECKY_PLUGIN_RUNTIME_DIR"] = path.join(environ["DECKY_HOME"], "data", self.plugin_directory) + helpers.mkdir_as_user(path.join(environ["DECKY_HOME"], "data")) + helpers.mkdir_as_user(environ["DECKY_PLUGIN_RUNTIME_DIR"]) + environ["DECKY_PLUGIN_LOG_DIR"] = path.join(environ["DECKY_HOME"], "logs", self.plugin_directory) + helpers.mkdir_as_user(path.join(environ["DECKY_HOME"], "logs")) + helpers.mkdir_as_user(environ["DECKY_PLUGIN_LOG_DIR"]) + environ["DECKY_PLUGIN_DIR"] = path.join(self.plugin_path, self.plugin_directory) + environ["DECKY_PLUGIN_NAME"] = self.name + if self.version: + environ["DECKY_PLUGIN_VERSION"] = self.version + environ["DECKY_PLUGIN_AUTHOR"] = self.author + + # append the plugin's `py_modules` to the recognized python paths + syspath.append(path.join(environ["DECKY_PLUGIN_DIR"], "py_modules")) + + spec = spec_from_file_location("_", self.file) + assert spec is not None + module = module_from_spec(spec) + assert spec.loader is not None + spec.loader.exec_module(module) + self.Plugin = module.Plugin + + if hasattr(self.Plugin, "_migration"): + get_event_loop().run_until_complete(self.Plugin._migration(self.Plugin)) + if hasattr(self.Plugin, "_main"): + get_event_loop().create_task(self.Plugin._main(self.Plugin)) + get_event_loop().create_task(socket.setup_server()) + get_event_loop().run_forever() + except: + self.log.error("Failed to start " + self.name + "!\n" + format_exc()) + exit(0) + + async def _unload(self): + try: + self.log.info("Attempting to unload with plugin " + self.name + "'s \"_unload\" function.\n") + if hasattr(self.Plugin, "_unload"): + await self.Plugin._unload(self.Plugin) + self.log.info("Unloaded " + self.name + "\n") + else: + self.log.info("Could not find \"_unload\" in " + self.name + "'s main.py" + "\n") + except: + self.log.error("Failed to unload " + self.name + "!\n" + format_exc()) + exit(0) + + async def on_new_message(self, message : str) -> str|None: + data = loads(message) + + if "stop" in data: + self.log.info("Calling Loader unload function.") + await self._unload() + get_event_loop().stop() + while get_event_loop().is_running(): + await sleep(0) + get_event_loop().close() + raise Exception("Closing message listener") + + # TODO there is definitely a better way to type this + d: SocketResponseDict = {"res": None, "success": True, "id": data["id"]} + try: + d["res"] = await getattr(self.Plugin, data["method"])(self.Plugin, **data["args"]) + except Exception as e: + d["res"] = str(e) + d["success"] = False + finally: + return dumps(d, ensure_ascii=False) + + + def start(self, socket: LocalSocket): + if self.passive: + return self + multiprocessing.Process(target=self._init, args=[socket]).start() + return self \ No newline at end of file -- cgit v1.2.3 From 315b2f9cda6def0c93bc7a0fa302415d2f972c85 Mon Sep 17 00:00:00 2001 From: marios8543 Date: Wed, 18 Oct 2023 00:04:14 +0300 Subject: Run response_listener task --- backend/src/plugin/plugin.py | 18 +++++++++++++++--- 1 file changed, 15 insertions(+), 3 deletions(-) (limited to 'backend/src/plugin') diff --git a/backend/src/plugin/plugin.py b/backend/src/plugin/plugin.py index b7d0a44a..d5f79d42 100644 --- a/backend/src/plugin/plugin.py +++ b/backend/src/plugin/plugin.py @@ -1,3 +1,4 @@ +from asyncio import Task, create_task from json import dumps, load, loads from logging import getLogger from os import path @@ -31,12 +32,12 @@ class PluginWrapper: self.sandboxed_plugin = SandboxedPlugin(self.name, self.passive, self.flags, self.file, self.plugin_directory, self.plugin_path, self.version, self.author) #TODO: Maybe somehow make LocalSocket not require on_new_message to make this more clear self.socket = LocalSocket(self.sandboxed_plugin.on_new_message) - self.sandboxed_plugin.start(self.socket) + self.listener_task: Task[Any] def __str__(self) -> str: return self.name - async def response_listener(self): + async def _response_listener(self): while True: line = await self.socket.read_single_line() if line != None: @@ -52,4 +53,15 @@ class PluginWrapper: await self.socket.write_single_line(dumps({ "method": method_name, "args": kwargs, "id": request.id }, ensure_ascii=False)) self.method_call_requests[request.id] = request - return await request.wait_for_result() \ No newline at end of file + return await request.wait_for_result() + + async def start(self): + self.sandboxed_plugin.start(self.socket) + self.listener_task = create_task(self._response_listener()) + + async def stop(self): + self.listener_task.cancel() + async def _(self: PluginWrapper): + await self.socket.write_single_line(dumps({ "stop": True }, ensure_ascii=False)) + await self.socket.close_socket_connection() + create_task(_(self)) \ No newline at end of file -- cgit v1.2.3 From d9ba637cd9c44bc7c7ec4e8381f4f05fae8e4244 Mon Sep 17 00:00:00 2001 From: marios8543 Date: Wed, 18 Oct 2023 14:45:36 +0300 Subject: Add message emit mechanism --- backend/src/plugin/plugin.py | 41 ++++++++++++++++++++++------------ backend/src/plugin/sandboxed_plugin.py | 22 +++++++++--------- 2 files changed, 39 insertions(+), 24 deletions(-) (limited to 'backend/src/plugin') diff --git a/backend/src/plugin/plugin.py b/backend/src/plugin/plugin.py index d5f79d42..a1b43ba9 100644 --- a/backend/src/plugin/plugin.py +++ b/backend/src/plugin/plugin.py @@ -2,12 +2,13 @@ from asyncio import Task, create_task from json import dumps, load, loads from logging import getLogger from os import path +from multiprocessing import Process from .sandboxed_plugin import SandboxedPlugin from .method_call_request import MethodCallRequest from ..localplatform.localsocket import LocalSocket -from typing import Any, Dict +from typing import Any, Callable, Coroutine, Dict class PluginWrapper: def __init__(self, file: str, plugin_directory: str, plugin_path: str) -> None: @@ -28,40 +29,52 @@ class PluginWrapper: self.passive = not path.isfile(self.file) self.log = getLogger("plugin") - self.method_call_requests: Dict[str, MethodCallRequest] = {} + self.sandboxed_plugin = SandboxedPlugin(self.name, self.passive, self.flags, self.file, self.plugin_directory, self.plugin_path, self.version, self.author) - #TODO: Maybe somehow make LocalSocket not require on_new_message to make this more clear - self.socket = LocalSocket(self.sandboxed_plugin.on_new_message) - self.listener_task: Task[Any] + #TODO: Maybe make LocalSocket not require on_new_message to make this cleaner + self._socket = LocalSocket(self.sandboxed_plugin.on_new_message) + self._listener_task: Task[Any] + self._method_call_requests: Dict[str, MethodCallRequest] = {} + + self.emitted_message_callback: Callable[[Dict[Any, Any]], Coroutine[Any, Any, Any]] def __str__(self) -> str: return self.name async def _response_listener(self): while True: - line = await self.socket.read_single_line() + line = await self._socket.read_single_line() if line != None: res = loads(line) - self.method_call_requests.pop(res["id"]).set_result(res) + if res["id"] == 0: + create_task(self.emitted_message_callback(res["payload"])) + return + self._method_call_requests.pop(res["id"]).set_result(res) + + async def set_emitted_message_callback(self, callback: Callable[[Dict[Any, Any]], Coroutine[Any, Any, Any]]): + self.emitted_message_callback = callback async def execute_method(self, method_name: str, kwargs: Dict[Any, Any]): if self.passive: raise RuntimeError("This plugin is passive (aka does not implement main.py)") request = MethodCallRequest() - await self.socket.get_socket_connection() - await self.socket.write_single_line(dumps({ "method": method_name, "args": kwargs, "id": request.id }, ensure_ascii=False)) - self.method_call_requests[request.id] = request + await self._socket.get_socket_connection() + await self._socket.write_single_line(dumps({ "method": method_name, "args": kwargs, "id": request.id }, ensure_ascii=False)) + self._method_call_requests[request.id] = request return await request.wait_for_result() async def start(self): - self.sandboxed_plugin.start(self.socket) + if self.passive: + return self + Process(target=self.sandboxed_plugin.initialize, args=[self._socket]).start() self.listener_task = create_task(self._response_listener()) + return self async def stop(self): - self.listener_task.cancel() + self._listener_task.cancel() async def _(self: PluginWrapper): - await self.socket.write_single_line(dumps({ "stop": True }, ensure_ascii=False)) - await self.socket.close_socket_connection() + await self._socket.write_single_line(dumps({ "stop": True }, ensure_ascii=False)) + await self._socket.close_socket_connection() create_task(_(self)) \ No newline at end of file diff --git a/backend/src/plugin/sandboxed_plugin.py b/backend/src/plugin/sandboxed_plugin.py index 7c96da01..972a535a 100644 --- a/backend/src/plugin/sandboxed_plugin.py +++ b/backend/src/plugin/sandboxed_plugin.py @@ -3,7 +3,6 @@ from signal import SIGINT, signal from importlib.util import module_from_spec, spec_from_file_location from json import dumps, loads from logging import getLogger -import multiprocessing from sys import exit, path as syspath from traceback import format_exc from asyncio import (get_event_loop, new_event_loop, @@ -15,7 +14,7 @@ from ..localplatform.localplatform import setgid, setuid, get_username, get_home from ..customtypes import UserType from .. import helpers -from typing import List +from typing import Any, Dict, List class SandboxedPlugin: def __init__(self, @@ -38,7 +37,9 @@ class SandboxedPlugin: self.log = getLogger("plugin") - def _init(self, socket: LocalSocket): + def initialize(self, socket: LocalSocket): + self._socket = socket + try: signal(SIGINT, lambda s, f: exit(0)) @@ -79,6 +80,9 @@ class SandboxedPlugin: spec.loader.exec_module(module) self.Plugin = module.Plugin + setattr(self.Plugin, "emit_message", self.emit_message) + #TODO: Find how to put emit_message on global namespace so it doesn't pollute Plugin + if hasattr(self.Plugin, "_migration"): get_event_loop().run_until_complete(self.Plugin._migration(self.Plugin)) if hasattr(self.Plugin, "_main"): @@ -113,7 +117,6 @@ class SandboxedPlugin: get_event_loop().close() raise Exception("Closing message listener") - # TODO there is definitely a better way to type this d: SocketResponseDict = {"res": None, "success": True, "id": data["id"]} try: d["res"] = await getattr(self.Plugin, data["method"])(self.Plugin, **data["args"]) @@ -123,9 +126,8 @@ class SandboxedPlugin: finally: return dumps(d, ensure_ascii=False) - - def start(self, socket: LocalSocket): - if self.passive: - return self - multiprocessing.Process(target=self._init, args=[socket]).start() - return self \ No newline at end of file + async def emit_message(self, message: Dict[Any, Any]): + await self._socket.write_single_line(dumps({ + "id": 0, + "payload": message + })) \ No newline at end of file -- cgit v1.2.3 From 934b1b35ad2ab2743a9d55de21eea73a54df72b9 Mon Sep 17 00:00:00 2001 From: marios8543 Date: Wed, 18 Oct 2023 15:34:25 +0300 Subject: fix start/stop methods --- backend/src/plugin/plugin.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) (limited to 'backend/src/plugin') diff --git a/backend/src/plugin/plugin.py b/backend/src/plugin/plugin.py index a1b43ba9..a0f926cd 100644 --- a/backend/src/plugin/plugin.py +++ b/backend/src/plugin/plugin.py @@ -65,14 +65,14 @@ class PluginWrapper: return await request.wait_for_result() - async def start(self): + def start(self): if self.passive: return self Process(target=self.sandboxed_plugin.initialize, args=[self._socket]).start() self.listener_task = create_task(self._response_listener()) return self - async def stop(self): + def stop(self): self._listener_task.cancel() async def _(self: PluginWrapper): await self._socket.write_single_line(dumps({ "stop": True }, ensure_ascii=False)) -- cgit v1.2.3 From 47e9708a209c9f48159880b90a710ea26ec09a29 Mon Sep 17 00:00:00 2001 From: marios8543 Date: Wed, 18 Oct 2023 17:21:57 +0300 Subject: fix static/lang file fetch and method call --- backend/src/plugin/method_call_request.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'backend/src/plugin') diff --git a/backend/src/plugin/method_call_request.py b/backend/src/plugin/method_call_request.py index 8d93a6cc..cebe34f8 100644 --- a/backend/src/plugin/method_call_request.py +++ b/backend/src/plugin/method_call_request.py @@ -26,4 +26,4 @@ class MethodCallRequest: await self.event.wait() if not self.response.success: raise Exception(self.response.result) - return self.response \ No newline at end of file + return self.response.result \ No newline at end of file -- cgit v1.2.3 From feabb582b2def5ad437a57dc9600d398d32fcfab Mon Sep 17 00:00:00 2001 From: marios8543 Date: Wed, 18 Oct 2023 19:38:07 +0300 Subject: run method calls asynchronously --- backend/src/plugin/plugin.py | 22 +++++++++++++--------- backend/src/plugin/sandboxed_plugin.py | 2 +- 2 files changed, 14 insertions(+), 10 deletions(-) (limited to 'backend/src/plugin') diff --git a/backend/src/plugin/plugin.py b/backend/src/plugin/plugin.py index a0f926cd..befd1569 100644 --- a/backend/src/plugin/plugin.py +++ b/backend/src/plugin/plugin.py @@ -26,6 +26,7 @@ class PluginWrapper: self.name = json["name"] self.author = json["author"] self.flags = json["flags"] + self.passive = not path.isfile(self.file) self.log = getLogger("plugin") @@ -43,15 +44,18 @@ class PluginWrapper: async def _response_listener(self): while True: - line = await self._socket.read_single_line() - if line != None: - res = loads(line) - if res["id"] == 0: - create_task(self.emitted_message_callback(res["payload"])) - return - self._method_call_requests.pop(res["id"]).set_result(res) + try: + line = await self._socket.read_single_line() + if line != None: + res = loads(line) + if res["id"] == "0": + create_task(self.emitted_message_callback(res["payload"])) + return + self._method_call_requests.pop(res["id"]).set_result(res) + except: + pass - async def set_emitted_message_callback(self, callback: Callable[[Dict[Any, Any]], Coroutine[Any, Any, Any]]): + def set_emitted_message_callback(self, callback: Callable[[Dict[Any, Any]], Coroutine[Any, Any, Any]]): self.emitted_message_callback = callback async def execute_method(self, method_name: str, kwargs: Dict[Any, Any]): @@ -69,7 +73,7 @@ class PluginWrapper: if self.passive: return self Process(target=self.sandboxed_plugin.initialize, args=[self._socket]).start() - self.listener_task = create_task(self._response_listener()) + self._listener_task = create_task(self._response_listener()) return self def stop(self): diff --git a/backend/src/plugin/sandboxed_plugin.py b/backend/src/plugin/sandboxed_plugin.py index 972a535a..8540bebc 100644 --- a/backend/src/plugin/sandboxed_plugin.py +++ b/backend/src/plugin/sandboxed_plugin.py @@ -128,6 +128,6 @@ class SandboxedPlugin: async def emit_message(self, message: Dict[Any, Any]): await self._socket.write_single_line(dumps({ - "id": 0, + "id": "0", "payload": message })) \ No newline at end of file -- cgit v1.2.3 From 28ca7b5c904ab67fab4147a340f2d78ca9a6e473 Mon Sep 17 00:00:00 2001 From: marios8543 Date: Wed, 18 Oct 2023 21:04:51 +0300 Subject: fix emit_message mechanism --- backend/src/plugin/plugin.py | 4 ++-- backend/src/plugin/sandboxed_plugin.py | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) (limited to 'backend/src/plugin') diff --git a/backend/src/plugin/plugin.py b/backend/src/plugin/plugin.py index befd1569..6c338106 100644 --- a/backend/src/plugin/plugin.py +++ b/backend/src/plugin/plugin.py @@ -50,8 +50,8 @@ class PluginWrapper: res = loads(line) if res["id"] == "0": create_task(self.emitted_message_callback(res["payload"])) - return - self._method_call_requests.pop(res["id"]).set_result(res) + else: + self._method_call_requests.pop(res["id"]).set_result(res) except: pass diff --git a/backend/src/plugin/sandboxed_plugin.py b/backend/src/plugin/sandboxed_plugin.py index 8540bebc..d44794fa 100644 --- a/backend/src/plugin/sandboxed_plugin.py +++ b/backend/src/plugin/sandboxed_plugin.py @@ -127,7 +127,7 @@ class SandboxedPlugin: return dumps(d, ensure_ascii=False) async def emit_message(self, message: Dict[Any, Any]): - await self._socket.write_single_line(dumps({ + await self._socket.write_single_line_server(dumps({ "id": "0", "payload": message })) \ No newline at end of file -- cgit v1.2.3