diff options
| author | K900 <me@0upti.me> | 2023-11-14 00:40:37 +0300 |
|---|---|---|
| committer | GitHub <noreply@github.com> | 2023-11-13 23:40:37 +0200 |
| commit | 5a633fdd8284dd1a2b6f3c95806f033ef4a4becf (patch) | |
| tree | b89f3660d3b8918484e6bc153003a84b95207045 /backend/src/plugin/sandboxed_plugin.py | |
| parent | 8ce4a7679e9f0abb67e85822fefe08237ec9d82e (diff) | |
| download | decky-loader-marios8543/async-plugin-method-requests.tar.gz decky-loader-marios8543/async-plugin-method-requests.zip | |
Packaging rework (#531)marios8543/async-plugin-method-requests
* fix: get rid of title view jank on latest beta
* Count the number of installs for each plugin (#557)
* Bump aiohttp from 3.8.4 to 3.8.5 in /backend (#558)
* fix: include Decky version in request for index.js
This avoids the If-Modified-Since logic in aiohttp and ensures Steam doesn't cache old JS,
even if the timestamps are normalized.
* fix: clean up shellcheck warnings in act runner script
* fix: gitignore settings/
* fix: ensure state directories exist when running without the installer
* feat: determine root directory correctly when running from in-tree
* fix: fix typo in CI script
* refactor: build a proper Python package with poetry
* refactor: move decky_plugin under the poetry structure
There's no need to special case it anymore, just treat it like any other Python module.
* sandboxed_plugin: better fix, attempt 2
---------
Co-authored-by: AAGaming <aagaming@riseup.net>
Co-authored-by: Party Wumpus <48649272+PartyWumpus@users.noreply.github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
Diffstat (limited to 'backend/src/plugin/sandboxed_plugin.py')
| -rw-r--r-- | backend/src/plugin/sandboxed_plugin.py | 138 |
1 files changed, 0 insertions, 138 deletions
diff --git a/backend/src/plugin/sandboxed_plugin.py b/backend/src/plugin/sandboxed_plugin.py deleted file mode 100644 index adf9f802..00000000 --- a/backend/src/plugin/sandboxed_plugin.py +++ /dev/null @@ -1,138 +0,0 @@ -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 -from sys import exit, path as syspath, modules as sysmodules -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 Any, Dict, 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 initialize(self, socket: LocalSocket): - self._socket = socket - - 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")) - - #TODO: FIX IN A LESS CURSED WAY - keys = [key.replace("src.", "") for key in sysmodules if key.startswith("src.")] - for key in keys: - sysmodules[key] = sysmodules["src"].__dict__[key] - - 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 - - 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"): - 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") - - 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) - - async def emit_message(self, message: Dict[Any, Any]): - await self._socket.write_single_line_server(dumps({ - "id": "0", - "payload": message - }))
\ No newline at end of file |
