summaryrefslogtreecommitdiff
path: root/backend
diff options
context:
space:
mode:
Diffstat (limited to 'backend')
-rw-r--r--backend/main.py195
-rw-r--r--backend/pyrightconfig.json3
-rw-r--r--backend/requirements.txt5
-rw-r--r--backend/src/browser.py (renamed from backend/browser.py)64
-rw-r--r--backend/src/customtypes.py (renamed from backend/customtypes.py)0
-rw-r--r--backend/src/helpers.py (renamed from backend/helpers.py)59
-rw-r--r--backend/src/injector.py (renamed from backend/injector.py)84
-rw-r--r--backend/src/legacy/library.js (renamed from backend/legacy/library.js)0
-rw-r--r--backend/src/loader.py (renamed from backend/loader.py)76
-rw-r--r--backend/src/localplatform.py (renamed from backend/localplatform.py)8
-rw-r--r--backend/src/localplatformlinux.py (renamed from backend/localplatformlinux.py)18
-rw-r--r--backend/src/localplatformwin.py (renamed from backend/localplatformwin.py)2
-rw-r--r--backend/src/localsocket.py (renamed from backend/localsocket.py)39
-rw-r--r--backend/src/main.py192
-rw-r--r--backend/src/plugin.py (renamed from backend/plugin.py)30
-rw-r--r--backend/src/settings.py (renamed from backend/settings.py)17
-rw-r--r--backend/src/updater.py (renamed from backend/updater.py)40
-rw-r--r--backend/src/utilities.py (renamed from backend/utilities.py)92
18 files changed, 506 insertions, 418 deletions
diff --git a/backend/main.py b/backend/main.py
index b2e3e74a..46a0671a 100644
--- a/backend/main.py
+++ b/backend/main.py
@@ -1,193 +1,4 @@
-# Change PyInstaller files permissions
-import sys
-from localplatform import (chmod, chown, service_stop, service_start,
- ON_WINDOWS, get_log_level, get_live_reload,
- get_server_port, get_server_host, get_chown_plugin_path,
- get_unprivileged_user, get_unprivileged_path,
- get_privileged_path)
-if hasattr(sys, '_MEIPASS'):
- chmod(sys._MEIPASS, 755)
-# Full imports
-from asyncio import new_event_loop, set_event_loop, sleep
-from json import dumps, loads
-from logging import DEBUG, INFO, basicConfig, getLogger
-from os import getenv, path
-from traceback import format_exc
-import multiprocessing
-
-import aiohttp_cors
-# Partial imports
-from aiohttp import client_exceptions, WSMsgType
-from aiohttp.web import Application, Response, get, run_app, static
-from aiohttp_jinja2 import setup as jinja_setup
-
-# local modules
-from browser import PluginBrowser
-from helpers import (REMOTE_DEBUGGER_UNIT, csrf_middleware, get_csrf_token,
- mkdir_as_user, get_system_pythonpaths, get_effective_user_id)
-
-from injector import get_gamepadui_tab, Tab, get_tabs, close_old_tabs
-from loader import Loader
-from settings import SettingsManager
-from updater import Updater
-from utilities import Utilities
-from customtypes import UserType
-
-
-basicConfig(
- level=get_log_level(),
- format="[%(module)s][%(levelname)s]: %(message)s"
-)
-
-logger = getLogger("Main")
-plugin_path = path.join(get_privileged_path(), "plugins")
-
-def chown_plugin_dir():
- if not path.exists(plugin_path): # For safety, create the folder before attempting to do anything with it
- mkdir_as_user(plugin_path)
-
- if not chown(plugin_path, UserType.HOST_USER) or not chmod(plugin_path, 555):
- logger.error(f"chown/chmod exited with a non-zero exit code")
-
-if get_chown_plugin_path() == True:
- chown_plugin_dir()
-
-class PluginManager:
- def __init__(self, loop) -> None:
- self.loop = loop
- self.web_app = Application()
- self.web_app.middlewares.append(csrf_middleware)
- self.cors = aiohttp_cors.setup(self.web_app, defaults={
- "https://steamloopback.host": aiohttp_cors.ResourceOptions(
- expose_headers="*",
- allow_headers="*",
- allow_credentials=True
- )
- })
- self.plugin_loader = Loader(self.web_app, plugin_path, self.loop, get_live_reload())
- self.settings = SettingsManager("loader", path.join(get_privileged_path(), "settings"))
- self.plugin_browser = PluginBrowser(plugin_path, self.plugin_loader.plugins, self.plugin_loader, self.settings)
- self.utilities = Utilities(self)
- self.updater = Updater(self)
-
- jinja_setup(self.web_app)
-
- async def startup(_):
- if self.settings.getSetting("cef_forward", False):
- self.loop.create_task(service_start(REMOTE_DEBUGGER_UNIT))
- else:
- self.loop.create_task(service_stop(REMOTE_DEBUGGER_UNIT))
- self.loop.create_task(self.loader_reinjector())
- self.loop.create_task(self.load_plugins())
-
- self.web_app.on_startup.append(startup)
-
- self.loop.set_exception_handler(self.exception_handler)
- self.web_app.add_routes([get("/auth/token", self.get_auth_token)])
-
- for route in list(self.web_app.router.routes()):
- self.cors.add(route)
- self.web_app.add_routes([static("/static", path.join(path.dirname(__file__), 'static'))])
- self.web_app.add_routes([static("/legacy", path.join(path.dirname(__file__), 'legacy'))])
-
- def exception_handler(self, loop, context):
- if context["message"] == "Unclosed connection":
- return
- loop.default_exception_handler(context)
-
- async def get_auth_token(self, request):
- return Response(text=get_csrf_token())
-
- async def load_plugins(self):
- # await self.wait_for_server()
- logger.debug("Loading plugins")
- self.plugin_loader.import_plugins()
- # await inject_to_tab("SP", "window.syncDeckyPlugins();")
- if self.settings.getSetting("pluginOrder", None) == None:
- self.settings.setSetting("pluginOrder", list(self.plugin_loader.plugins.keys()))
- logger.debug("Did not find pluginOrder setting, set it to default")
-
- async def loader_reinjector(self):
- while True:
- tab = None
- nf = False
- dc = False
- while not tab:
- try:
- tab = await get_gamepadui_tab()
- except (client_exceptions.ClientConnectorError, client_exceptions.ServerDisconnectedError):
- if not dc:
- logger.debug("Couldn't connect to debugger, waiting...")
- dc = True
- pass
- except ValueError:
- if not nf:
- logger.debug("Couldn't find GamepadUI tab, waiting...")
- nf = True
- pass
- if not tab:
- await sleep(5)
- await tab.open_websocket()
- await tab.enable()
- await self.inject_javascript(tab, True)
- try:
- async for msg in tab.listen_for_message():
- # this gets spammed a lot
- if msg.get("method", None) != "Page.navigatedWithinDocument":
- logger.debug("Page event: " + str(msg.get("method", None)))
- if msg.get("method", None) == "Page.domContentEventFired":
- if not await tab.has_global_var("deckyHasLoaded", False):
- await self.inject_javascript(tab)
- if msg.get("method", None) == "Inspector.detached":
- logger.info("CEF has requested that we detach.")
- await tab.close_websocket()
- break
- # If this is a forceful disconnect the loop will just stop without any failure message. In this case, injector.py will handle this for us so we don't need to close the socket.
- # This is because of https://github.com/aio-libs/aiohttp/blob/3ee7091b40a1bc58a8d7846e7878a77640e96996/aiohttp/client_ws.py#L321
- logger.info("CEF has disconnected...")
- # At this point the loop starts again and we connect to the freshly started Steam client once it is ready.
- except Exception as e:
- logger.error("Exception while reading page events " + format_exc())
- await tab.close_websocket()
- pass
- # while True:
- # await sleep(5)
- # if not await tab.has_global_var("deckyHasLoaded", False):
- # logger.info("Plugin loader isn't present in Steam anymore, reinjecting...")
- # await self.inject_javascript(tab)
-
- async def inject_javascript(self, tab: Tab, first=False, request=None):
- logger.info("Loading Decky frontend!")
- try:
- if first:
- if await tab.has_global_var("deckyHasLoaded", False):
- await close_old_tabs()
- await tab.evaluate_js("try{if (window.deckyHasLoaded){setTimeout(() => location.reload(), 100)}else{window.deckyHasLoaded = true;(async()=>{try{while(!window.SP_REACT){await new Promise(r => setTimeout(r, 10))};await import('http://localhost:1337/frontend/index.js')}catch(e){console.error(e)};})();}}catch(e){console.error(e)}", False, False, False)
- except:
- logger.info("Failed to inject JavaScript into tab\n" + format_exc())
- pass
-
- def run(self):
- return run_app(self.web_app, host=get_server_host(), port=get_server_port(), loop=self.loop, access_log=None)
-
+# This file is needed to make the relative imports in src/ work properly.
if __name__ == "__main__":
- if ON_WINDOWS:
- # Fix windows/flask not recognising that .js means 'application/javascript'
- import mimetypes
- mimetypes.add_type('application/javascript', '.js')
-
- # Required for multiprocessing support in frozen files
- multiprocessing.freeze_support()
- else:
- if get_effective_user_id() != 0:
- logger.warning(f"decky is running as an unprivileged user, this is not officially supported and may cause issues")
-
- # Append the loader's plugin path to the recognized python paths
- sys.path.append(path.join(path.dirname(__file__), "plugin"))
-
- # Append the system and user python paths
- sys.path.extend(get_system_pythonpaths())
-
- loop = new_event_loop()
- set_event_loop(loop)
- PluginManager(loop).run()
+ from src.main import main
+ main()
diff --git a/backend/pyrightconfig.json b/backend/pyrightconfig.json
new file mode 100644
index 00000000..9937f227
--- /dev/null
+++ b/backend/pyrightconfig.json
@@ -0,0 +1,3 @@
+{
+ "strict": ["*"]
+} \ No newline at end of file
diff --git a/backend/requirements.txt b/backend/requirements.txt
new file mode 100644
index 00000000..326a924c
--- /dev/null
+++ b/backend/requirements.txt
@@ -0,0 +1,5 @@
+aiohttp==3.8.4
+aiohttp-jinja2==1.5.1
+aiohttp_cors==0.7.0
+watchdog==2.1.7
+certifi==2023.7.22
diff --git a/backend/browser.py b/backend/src/browser.py
index ce9b3dd7..da8569be 100644
--- a/backend/browser.py
+++ b/backend/src/browser.py
@@ -4,53 +4,70 @@ import json
# from pprint import pformat
# Partial imports
-from aiohttp import ClientSession, web
-from asyncio import get_event_loop, sleep
-from concurrent.futures import ProcessPoolExecutor
+from aiohttp import ClientSession
+from asyncio import sleep
from hashlib import sha256
from io import BytesIO
from logging import getLogger
-from os import R_OK, W_OK, path, rename, listdir, access, mkdir
+from os import R_OK, W_OK, path, listdir, access, mkdir
from shutil import rmtree
from time import time
from zipfile import ZipFile
-from localplatform import chown, chmod
+from enum import IntEnum
+from typing import Dict, List, TypedDict
# Local modules
-from helpers import get_ssl_context, download_remote_binary_to_path
-from injector import get_gamepadui_tab
+from .localplatform import chown, chmod
+from .loader import Loader, Plugins
+from .helpers import get_ssl_context, download_remote_binary_to_path
+from .settings import SettingsManager
+from .injector import get_gamepadui_tab
logger = getLogger("Browser")
+class PluginInstallType(IntEnum):
+ INSTALL = 0
+ REINSTALL = 1
+ UPDATE = 2
+
+class PluginInstallRequest(TypedDict):
+ name: str
+ artifact: str
+ version: str
+ hash: str
+ install_type: PluginInstallType
+
class PluginInstallContext:
- def __init__(self, artifact, name, version, hash) -> None:
+ def __init__(self, artifact: str, name: str, version: str, hash: str) -> None:
self.artifact = artifact
self.name = name
self.version = version
self.hash = hash
class PluginBrowser:
- def __init__(self, plugin_path, plugins, loader, settings) -> None:
+ def __init__(self, plugin_path: str, plugins: Plugins, loader: Loader, settings: SettingsManager) -> None:
self.plugin_path = plugin_path
self.plugins = plugins
self.loader = loader
self.settings = settings
- self.install_requests = {}
+ self.install_requests: Dict[str, PluginInstallContext | List[PluginInstallContext]] = {}
- def _unzip_to_plugin_dir(self, zip, name, hash):
+ def _unzip_to_plugin_dir(self, zip: BytesIO, name: str, hash: str):
zip_hash = sha256(zip.getbuffer()).hexdigest()
if hash and (zip_hash != hash):
return False
zip_file = ZipFile(zip)
zip_file.extractall(self.plugin_path)
- plugin_dir = path.join(self.plugin_path, self.find_plugin_folder(name))
+ plugin_folder = self.find_plugin_folder(name)
+ assert plugin_folder is not None
+ plugin_dir = path.join(self.plugin_path, plugin_folder)
if not chown(plugin_dir) or not chmod(plugin_dir, 555):
logger.error(f"chown/chmod exited with a non-zero exit code")
return False
return True
- async def _download_remote_binaries_for_plugin_with_name(self, pluginBasePath):
+ async def _download_remote_binaries_for_plugin_with_name(self, pluginBasePath: str):
rv = False
try:
packageJsonPath = path.join(pluginBasePath, 'package.json')
@@ -91,7 +108,7 @@ class PluginBrowser:
return rv
"""Return the filename (only) for the specified plugin"""
- def find_plugin_folder(self, name):
+ def find_plugin_folder(self, name: str) -> str | None:
for folder in listdir(self.plugin_path):
try:
with open(path.join(self.plugin_path, folder, 'plugin.json'), "r", encoding="utf-8") as f:
@@ -102,11 +119,13 @@ class PluginBrowser:
except:
logger.debug(f"skipping {folder}")
- async def uninstall_plugin(self, name):
+ async def uninstall_plugin(self, name: str):
if self.loader.watcher:
self.loader.watcher.disabled = True
tab = await get_gamepadui_tab()
- plugin_dir = path.join(self.plugin_path, self.find_plugin_folder(name))
+ plugin_folder = self.find_plugin_folder(name)
+ assert plugin_folder is not None
+ plugin_dir = path.join(self.plugin_path, plugin_folder)
try:
logger.info("uninstalling " + name)
logger.info(" at dir " + plugin_dir)
@@ -133,7 +152,7 @@ class PluginBrowser:
if self.loader.watcher:
self.loader.watcher.disabled = False
- async def _install(self, artifact, name, version, hash):
+ async def _install(self, artifact: str, name: str, version: str, hash: str):
# Will be set later in code
res_zip = None
@@ -185,6 +204,7 @@ class PluginBrowser:
ret = self._unzip_to_plugin_dir(res_zip, name, hash)
if ret:
plugin_folder = self.find_plugin_folder(name)
+ assert plugin_folder is not None
plugin_dir = path.join(self.plugin_path, plugin_folder)
ret = await self._download_remote_binaries_for_plugin_with_name(plugin_dir)
if ret:
@@ -206,14 +226,14 @@ class PluginBrowser:
if self.loader.watcher:
self.loader.watcher.disabled = False
- async def request_plugin_install(self, artifact, name, version, hash, install_type):
+ async def request_plugin_install(self, artifact: str, name: str, version: str, hash: str, install_type: PluginInstallType):
request_id = str(time())
self.install_requests[request_id] = PluginInstallContext(artifact, name, version, hash)
tab = await get_gamepadui_tab()
await tab.open_websocket()
await tab.evaluate_js(f"DeckyPluginLoader.addPluginInstallPrompt('{name}', '{version}', '{request_id}', '{hash}', {install_type})")
- async def request_multiple_plugin_installs(self, requests):
+ async def request_multiple_plugin_installs(self, requests: List[PluginInstallRequest]):
request_id = str(time())
self.install_requests[request_id] = [PluginInstallContext(req['artifact'], req['name'], req['version'], req['hash']) for req in requests]
js_requests_parameter = ','.join([
@@ -224,17 +244,17 @@ class PluginBrowser:
await tab.open_websocket()
await tab.evaluate_js(f"DeckyPluginLoader.addMultiplePluginsInstallPrompt('{request_id}', [{js_requests_parameter}])")
- async def confirm_plugin_install(self, request_id):
+ async def confirm_plugin_install(self, request_id: str):
requestOrRequests = self.install_requests.pop(request_id)
if isinstance(requestOrRequests, list):
[await self._install(req.artifact, req.name, req.version, req.hash) for req in requestOrRequests]
else:
await self._install(requestOrRequests.artifact, requestOrRequests.name, requestOrRequests.version, requestOrRequests.hash)
- def cancel_plugin_install(self, request_id):
+ def cancel_plugin_install(self, request_id: str):
self.install_requests.pop(request_id)
- def cleanup_plugin_settings(self, name):
+ def cleanup_plugin_settings(self, name: str):
"""Removes any settings related to a plugin. Propably called when a plugin is uninstalled.
Args:
diff --git a/backend/customtypes.py b/backend/src/customtypes.py
index 84ebc235..84ebc235 100644
--- a/backend/customtypes.py
+++ b/backend/src/customtypes.py
diff --git a/backend/helpers.py b/backend/src/helpers.py
index a1877fb8..f8796bd8 100644
--- a/backend/helpers.py
+++ b/backend/src/helpers.py
@@ -2,16 +2,16 @@ import re
import ssl
import uuid
import os
-import sys
import subprocess
from hashlib import sha256
from io import BytesIO
import certifi
-from aiohttp.web import Response, middleware
+from aiohttp.web import Request, Response, middleware
+from aiohttp.typedefs import Handler
from aiohttp import ClientSession
-import localplatform
-from customtypes import UserType
+from . import localplatform
+from .customtypes import UserType
from logging import getLogger
REMOTE_DEBUGGER_UNIT = "steam-web-debug-portforward.service"
@@ -31,17 +31,17 @@ def get_csrf_token():
return csrf_token
@middleware
-async def csrf_middleware(request, handler):
+async def csrf_middleware(request: Request, handler: Handler):
if str(request.method) == "OPTIONS" or request.headers.get('Authentication') == csrf_token or str(request.rel_url) == "/auth/token" or str(request.rel_url).startswith("/plugins/load_main/") or str(request.rel_url).startswith("/static/") or str(request.rel_url).startswith("/legacy/") or str(request.rel_url).startswith("/steam_resource/") or str(request.rel_url).startswith("/frontend/") or assets_regex.match(str(request.rel_url)) or frontend_regex.match(str(request.rel_url)):
return await handler(request)
- return Response(text='Forbidden', status='403')
+ return Response(text='Forbidden', status=403)
# Get the default homebrew path unless a home_path is specified. home_path argument is deprecated
-def get_homebrew_path(home_path = None) -> str:
+def get_homebrew_path() -> str:
return localplatform.get_unprivileged_path()
# Recursively create path and chown as user
-def mkdir_as_user(path):
+def mkdir_as_user(path: str):
path = os.path.realpath(path)
os.makedirs(path, exist_ok=True)
localplatform.chown(path)
@@ -57,23 +57,18 @@ def get_loader_version() -> str:
# returns the appropriate system python paths
def get_system_pythonpaths() -> list[str]:
- extra_args = {}
-
- if localplatform.ON_LINUX:
- # run as normal normal user to also include user python paths
- extra_args["user"] = localplatform.localplatform._get_user_id()
- extra_args["env"] = {}
-
try:
+ # run as normal normal user if on linux to also include user python paths
proc = subprocess.run(["python3" if localplatform.ON_LINUX else "python", "-c", "import sys; print('\\n'.join(x for x in sys.path if x))"],
- capture_output=True, **extra_args)
+ # TODO make this less insane
+ capture_output=True, user=localplatform.localplatform._get_user_id() if localplatform.ON_LINUX else None, env={} if localplatform.ON_LINUX else None) # type: ignore
return [x.strip() for x in proc.stdout.decode().strip().split("\n")]
except Exception as e:
logger.warn(f"Failed to execute get_system_pythonpaths(): {str(e)}")
return []
# Download Remote Binaries to local Plugin
-async def download_remote_binary_to_path(url, binHash, path) -> bool:
+async def download_remote_binary_to_path(url: str, binHash: str, path: str) -> bool:
rv = False
try:
if os.access(os.path.dirname(path), os.W_OK):
@@ -110,46 +105,42 @@ def set_user_group() -> str:
# Get the user id hosting the plugin loader
def get_user_id() -> int:
- return localplatform.localplatform._get_user_id()
+ return localplatform.localplatform._get_user_id() # pyright: ignore [reportPrivateUsage]
# Get the user hosting the plugin loader
def get_user() -> str:
- return localplatform.localplatform._get_user()
+ return localplatform.localplatform._get_user() # pyright: ignore [reportPrivateUsage]
# Get the effective user id of the running process
def get_effective_user_id() -> int:
- return localplatform.localplatform._get_effective_user_id()
+ return localplatform.localplatform._get_effective_user_id() # pyright: ignore [reportPrivateUsage]
# Get the effective user of the running process
def get_effective_user() -> str:
- return localplatform.localplatform._get_effective_user()
+ return localplatform.localplatform._get_effective_user() # pyright: ignore [reportPrivateUsage]
# Get the effective user group id of the running process
def get_effective_user_group_id() -> int:
- return localplatform.localplatform._get_effective_user_group_id()
+ return localplatform.localplatform._get_effective_user_group_id() # pyright: ignore [reportPrivateUsage]
# Get the effective user group of the running process
def get_effective_user_group() -> str:
- return localplatform.localplatform._get_effective_user_group()
+ return localplatform.localplatform._get_effective_user_group() # pyright: ignore [reportPrivateUsage]
# Get the user owner of the given file path.
-def get_user_owner(file_path) -> str:
- return localplatform.localplatform._get_user_owner(file_path)
+def get_user_owner(file_path: str) -> str:
+ return localplatform.localplatform._get_user_owner(file_path) # pyright: ignore [reportPrivateUsage]
-# Get the user group of the given file path.
-def get_user_group(file_path) -> str:
- return localplatform.localplatform._get_user_group(file_path)
+# Get the user group of the given file path, or the user group hosting the plugin loader
+def get_user_group(file_path: str | None = None) -> str:
+ return localplatform.localplatform._get_user_group(file_path) # pyright: ignore [reportPrivateUsage]
# Get the group id of the user hosting the plugin loader
def get_user_group_id() -> int:
- return localplatform.localplatform._get_user_group_id()
-
-# Get the group of the user hosting the plugin loader
-def get_user_group() -> str:
- return localplatform.localplatform._get_user_group()
+ return localplatform.localplatform._get_user_group_id() # pyright: ignore [reportPrivateUsage]
# Get the default home path unless a user is specified
-def get_home_path(username = None) -> str:
+def get_home_path(username: str | None = None) -> str:
return localplatform.get_home_path(UserType.ROOT if username == "root" else UserType.HOST_USER)
async def is_systemd_unit_active(unit_name: str) -> bool:
diff --git a/backend/injector.py b/backend/src/injector.py
index e3414fee..a217f689 100644
--- a/backend/injector.py
+++ b/backend/src/injector.py
@@ -2,10 +2,9 @@
from asyncio import sleep
from logging import getLogger
-from traceback import format_exc
-from typing import List
+from typing import Any, Callable, List, TypedDict, Dict
-from aiohttp import ClientSession, WSMsgType
+from aiohttp import ClientSession
from aiohttp.client_exceptions import ClientConnectorError, ClientOSError
from asyncio.exceptions import TimeoutError
import uuid
@@ -14,35 +13,43 @@ BASE_ADDRESS = "http://localhost:8080"
logger = getLogger("Injector")
+class _TabResponse(TypedDict):
+ title: str
+ id: str
+ url: str
+ webSocketDebuggerUrl: str
class Tab:
cmd_id = 0
- def __init__(self, res) -> None:
- self.title = res["title"]
- self.id = res["id"]
- self.url = res["url"]
- self.ws_url = res["webSocketDebuggerUrl"]
+ def __init__(self, res: _TabResponse) -> None:
+ self.title: str = res["title"]
+ self.id: str = res["id"]
+ self.url: str = res["url"]
+ self.ws_url: str = res["webSocketDebuggerUrl"]
self.websocket = None
self.client = None
async def open_websocket(self):
self.client = ClientSession()
- self.websocket = await self.client.ws_connect(self.ws_url)
+ self.websocket = await self.client.ws_connect(self.ws_url) # type: ignore
async def close_websocket(self):
- await self.websocket.close()
- await self.client.close()
+ if self.websocket:
+ await self.websocket.close()
+ if self.client:
+ await self.client.close()
async def listen_for_message(self):
- async for message in self.websocket:
- data = message.json()
- yield data
- logger.warn(f"The Tab {self.title} socket has been disconnected while listening for messages.")
- await self.close_websocket()
+ if self.websocket:
+ async for message in self.websocket:
+ data = message.json()
+ yield data
+ logger.warn(f"The Tab {self.title} socket has been disconnected while listening for messages.")
+ await self.close_websocket()
- async def _send_devtools_cmd(self, dc, receive=True):
+ async def _send_devtools_cmd(self, dc: Dict[str, Any], receive: bool = True):
if self.websocket:
self.cmd_id += 1
dc["id"] = self.cmd_id
@@ -54,7 +61,7 @@ class Tab:
return None
raise RuntimeError("Websocket not opened")
- async def evaluate_js(self, js, run_async=False, manage_socket=True, get_result=True):
+ async def evaluate_js(self, js: str, run_async: bool | None = False, manage_socket: bool | None = True, get_result: bool = True):
try:
if manage_socket:
await self.open_websocket()
@@ -73,15 +80,16 @@ class Tab:
await self.close_websocket()
return res
- async def has_global_var(self, var_name, manage_socket=True):
+ async def has_global_var(self, var_name: str, manage_socket: bool = True):
res = await self.evaluate_js(f"window['{var_name}'] !== null && window['{var_name}'] !== undefined", False, manage_socket)
+ assert res is not None
if not "result" in res or not "result" in res["result"] or not "value" in res["result"]["result"]:
return False
return res["result"]["result"]["value"]
- async def close(self, manage_socket=True):
+ async def close(self, manage_socket: bool = True):
try:
if manage_socket:
await self.open_websocket()
@@ -111,7 +119,7 @@ class Tab:
"method": "Page.disable",
}, False)
- async def refresh(self, manage_socket=True):
+ async def refresh(self, manage_socket: bool = True):
try:
if manage_socket:
await self.open_websocket()
@@ -125,7 +133,7 @@ class Tab:
await self.close_websocket()
return
- async def reload_and_evaluate(self, js, manage_socket=True):
+ async def reload_and_evaluate(self, js: str, manage_socket: bool = True):
"""
Reloads the current tab, with JS to run on load via debugger
"""
@@ -153,11 +161,13 @@ class Tab:
}
}, True)
+ assert breakpoint_res is not None
+
logger.info(breakpoint_res)
# Page finishes loading when breakpoint hits
- for x in range(20):
+ for _ in range(20):
# this works around 1/5 of the time, so just send it 8 times.
# the js accounts for being injected multiple times allowing only one instance to run at a time anyway
await self._send_devtools_cmd({
@@ -176,7 +186,7 @@ class Tab:
}
}, False)
- for x in range(4):
+ for _ in range(4):
await self._send_devtools_cmd({
"method": "Debugger.resume"
}, False)
@@ -190,7 +200,7 @@ class Tab:
await self.close_websocket()
return
- async def add_script_to_evaluate_on_new_document(self, js, add_dom_wrapper=True, manage_socket=True, get_result=True):
+ async def add_script_to_evaluate_on_new_document(self, js: str, add_dom_wrapper: bool = True, manage_socket: bool = True, get_result: bool = True):
"""
How the underlying call functions is not particularly clear from the devtools docs, so stealing puppeteer's description:
@@ -253,7 +263,7 @@ class Tab:
await self.close_websocket()
return res
- async def remove_script_to_evaluate_on_new_document(self, script_id, manage_socket=True):
+ async def remove_script_to_evaluate_on_new_document(self, script_id: str, manage_socket: bool = True):
"""
Removes a script from a page that was added with `add_script_to_evaluate_on_new_document`
@@ -267,7 +277,7 @@ class Tab:
if manage_socket:
await self.open_websocket()
- res = await self._send_devtools_cmd({
+ await self._send_devtools_cmd({
"method": "Page.removeScriptToEvaluateOnNewDocument",
"params": {
"identifier": script_id
@@ -278,15 +288,16 @@ class Tab:
if manage_socket:
await self.close_websocket()
- async def has_element(self, element_name, manage_socket=True):
+ async def has_element(self, element_name: str, manage_socket: bool = True):
res = await self.evaluate_js(f"document.getElementById('{element_name}') != null", False, manage_socket)
+ assert res is not None
if not "result" in res or not "result" in res["result"] or not "value" in res["result"]["result"]:
return False
return res["result"]["result"]["value"]
- async def inject_css(self, style, manage_socket=True):
+ async def inject_css(self, style: str, manage_socket: bool = True):
try:
css_id = str(uuid.uuid4())
@@ -300,6 +311,8 @@ class Tab:
}})()
""", False, manage_socket)
+ assert result is not None
+
if "exceptionDetails" in result["result"]:
return {
"success": False,
@@ -316,7 +329,7 @@ class Tab:
"result": e
}
- async def remove_css(self, css_id, manage_socket=True):
+ async def remove_css(self, css_id: str, manage_socket: bool = True):
try:
result = await self.evaluate_js(
f"""
@@ -328,6 +341,8 @@ class Tab:
}})()
""", False, manage_socket)
+ assert result is not None
+
if "exceptionDetails" in result["result"]:
return {
"success": False,
@@ -343,8 +358,9 @@ class Tab:
"result": e
}
- async def get_steam_resource(self, url):
+ async def get_steam_resource(self, url: str):
res = await self.evaluate_js(f'(async function test() {{ return await (await fetch("{url}")).text() }})()', True)
+ assert res is not None
return res["result"]["result"]["value"]
def __repr__(self):
@@ -380,14 +396,14 @@ async def get_tabs() -> List[Tab]:
raise Exception(f"/json did not return 200. {await res.text()}")
-async def get_tab(tab_name) -> Tab:
+async def get_tab(tab_name: str) -> Tab:
tabs = await get_tabs()
tab = next((i for i in tabs if i.title == tab_name), None)
if not tab:
raise ValueError(f"Tab {tab_name} not found")
return tab
-async def get_tab_lambda(test) -> Tab:
+async def get_tab_lambda(test: Callable[[Tab], bool]) -> Tab:
tabs = await get_tabs()
tab = next((i for i in tabs if test(i)), None)
if not tab:
@@ -408,7 +424,7 @@ async def get_gamepadui_tab() -> Tab:
raise ValueError(f"GamepadUI Tab not found")
return tab
-async def inject_to_tab(tab_name, js, run_async=False):
+async def inject_to_tab(tab_name: str, js: str, run_async: bool = False):
tab = await get_tab(tab_name)
return await tab.evaluate_js(js, run_async)
diff --git a/backend/legacy/library.js b/backend/src/legacy/library.js
index 17f4e46f..17f4e46f 100644
--- a/backend/legacy/library.js
+++ b/backend/src/legacy/library.js
diff --git a/backend/loader.py b/backend/src/loader.py
index d07b1c08..e59cbcaf 100644
--- a/backend/loader.py
+++ b/backend/src/loader.py
@@ -1,34 +1,43 @@
-from asyncio import Queue, sleep
+from __future__ import annotations
+from asyncio import AbstractEventLoop, Queue, sleep
from json.decoder import JSONDecodeError
from logging import getLogger
from os import listdir, path
from pathlib import Path
from traceback import print_exc
+from typing import Any, Tuple
from aiohttp import web
from os.path import exists
-from watchdog.events import RegexMatchingEventHandler
-from watchdog.observers import Observer
+from watchdog.events import RegexMatchingEventHandler, DirCreatedEvent, DirModifiedEvent, FileCreatedEvent, FileModifiedEvent # type: ignore
+from watchdog.observers import Observer # type: ignore
-from injector import get_tab, get_gamepadui_tab
-from plugin import PluginWrapper
+from typing import TYPE_CHECKING
+if TYPE_CHECKING:
+ from .main import PluginManager
+
+from .injector import get_tab, get_gamepadui_tab
+from .plugin import PluginWrapper
+
+Plugins = dict[str, PluginWrapper]
+ReloadQueue = Queue[Tuple[str, str, bool | None] | Tuple[str, str]]
class FileChangeHandler(RegexMatchingEventHandler):
- def __init__(self, queue, plugin_path) -> None:
- super().__init__(regexes=[r'^.*?dist\/index\.js$', r'^.*?main\.py$'])
+ def __init__(self, queue: ReloadQueue, plugin_path: str) -> None:
+ super().__init__(regexes=[r'^.*?dist\/index\.js$', r'^.*?main\.py$']) # type: ignore
self.logger = getLogger("file-watcher")
self.plugin_path = plugin_path
self.queue = queue
self.disabled = True
- def maybe_reload(self, src_path):
+ def maybe_reload(self, src_path: str):
if self.disabled:
return
plugin_dir = Path(path.relpath(src_path, self.plugin_path)).parts[0]
if exists(path.join(self.plugin_path, plugin_dir, "plugin.json")):
self.queue.put_nowait((path.join(self.plugin_path, plugin_dir, "main.py"), plugin_dir, True))
- def on_created(self, event):
+ def on_created(self, event: DirCreatedEvent | FileCreatedEvent):
src_path = event.src_path
if "__pycache__" in src_path:
return
@@ -42,7 +51,7 @@ class FileChangeHandler(RegexMatchingEventHandler):
self.logger.debug(f"file created: {src_path}")
self.maybe_reload(src_path)
- def on_modified(self, event):
+ def on_modified(self, event: DirModifiedEvent | FileModifiedEvent):
src_path = event.src_path
if "__pycache__" in src_path:
return
@@ -57,25 +66,25 @@ class FileChangeHandler(RegexMatchingEventHandler):
self.maybe_reload(src_path)
class Loader:
- def __init__(self, server_instance, plugin_path, loop, live_reload=False) -> None:
+ def __init__(self, server_instance: PluginManager, plugin_path: str, loop: AbstractEventLoop, live_reload: bool = False) -> None:
self.loop = loop
self.logger = getLogger("Loader")
self.plugin_path = plugin_path
self.logger.info(f"plugin_path: {self.plugin_path}")
- self.plugins : dict[str, PluginWrapper] = {}
+ self.plugins: Plugins = {}
self.watcher = None
self.live_reload = live_reload
- self.reload_queue = Queue()
+ self.reload_queue: ReloadQueue = Queue()
self.loop.create_task(self.handle_reloads())
if live_reload:
self.observer = Observer()
self.watcher = FileChangeHandler(self.reload_queue, plugin_path)
- self.observer.schedule(self.watcher, self.plugin_path, recursive=True)
+ self.observer.schedule(self.watcher, self.plugin_path, recursive=True) # type: ignore
self.observer.start()
self.loop.create_task(self.enable_reload_wait())
- server_instance.add_routes([
+ server_instance.web_app.add_routes([
web.get("/frontend/{path:.*}", self.handle_frontend_assets),
web.get("/locales/{path:.*}", self.handle_frontend_locales),
web.get("/plugins", self.get_plugins),
@@ -93,40 +102,41 @@ class Loader:
async def enable_reload_wait(self):
if self.live_reload:
await sleep(10)
- self.logger.info("Hot reload enabled")
- self.watcher.disabled = False
+ if self.watcher:
+ self.logger.info("Hot reload enabled")
+ self.watcher.disabled = False
- async def handle_frontend_assets(self, request):
- file = path.join(path.dirname(__file__), "static", request.match_info["path"])
+ async def handle_frontend_assets(self, request: web.Request):
+ file = path.join(path.dirname(__file__), "..", "static", request.match_info["path"])
return web.FileResponse(file, headers={"Cache-Control": "no-cache"})
- async def handle_frontend_locales(self, request):
+ async def handle_frontend_locales(self, request: web.Request):
req_lang = request.match_info["path"]
- file = path.join(path.dirname(__file__), "locales", req_lang)
+ file = path.join(path.dirname(__file__), "..", "locales", req_lang)
if exists(file):
return web.FileResponse(file, headers={"Cache-Control": "no-cache", "Content-Type": "application/json"})
else:
self.logger.info(f"Language {req_lang} not available, returning an empty dictionary")
return web.json_response(data={}, headers={"Cache-Control": "no-cache"})
- async def get_plugins(self, request):
+ async def get_plugins(self, request: web.Request):
plugins = list(self.plugins.values())
return web.json_response([{"name": str(i) if not i.legacy else "$LEGACY_"+str(i), "version": i.version} for i in plugins])
- def handle_plugin_frontend_assets(self, request):
+ async def handle_plugin_frontend_assets(self, request: web.Request):
plugin = self.plugins[request.match_info["plugin_name"]]
file = path.join(self.plugin_path, plugin.plugin_directory, "dist/assets", request.match_info["path"])
return web.FileResponse(file, headers={"Cache-Control": "no-cache"})
- def handle_frontend_bundle(self, request):
+ async def handle_frontend_bundle(self, request: web.Request):
plugin = self.plugins[request.match_info["plugin_name"]]
with open(path.join(self.plugin_path, plugin.plugin_directory, "dist/index.js"), "r", encoding="utf-8") as bundle:
return web.Response(text=bundle.read(), content_type="application/javascript")
- def import_plugin(self, file, plugin_directory, refresh=False, batch=False):
+ def import_plugin(self, file: str, plugin_directory: str, refresh: bool | None = False, batch: bool | None = False):
try:
plugin = PluginWrapper(file, plugin_directory, self.plugin_path)
if plugin.name in self.plugins:
@@ -146,7 +156,7 @@ class Loader:
self.logger.error(f"Could not load {file}. {e}")
print_exc()
- async def dispatch_plugin(self, name, version):
+ async def dispatch_plugin(self, name: str, version: str | None):
gpui_tab = await get_gamepadui_tab()
await gpui_tab.evaluate_js(f"window.importDeckyPlugin('{name}', '{version}')")
@@ -161,15 +171,15 @@ class Loader:
async def handle_reloads(self):
while True:
args = await self.reload_queue.get()
- self.import_plugin(*args)
+ self.import_plugin(*args) # type: ignore
- async def handle_plugin_method_call(self, request):
+ async def handle_plugin_method_call(self, request: web.Request):
res = {}
plugin = self.plugins[request.match_info["plugin_name"]]
method_name = request.match_info["method_name"]
try:
method_info = await request.json()
- args = method_info["args"]
+ args: Any = method_info["args"]
except JSONDecodeError:
args = {}
try:
@@ -189,7 +199,7 @@ class Loader:
can introduce it more smoothly and give people the chance to sample the new features even
without plugin support. They will be removed once legacy plugins are no longer relevant.
"""
- async def load_plugin_main_view(self, request):
+ async def load_plugin_main_view(self, request: web.Request):
plugin = self.plugins[request.match_info["name"]]
with open(path.join(self.plugin_path, plugin.plugin_directory, plugin.main_view_html), "r", encoding="utf-8") as template:
template_data = template.read()
@@ -201,7 +211,7 @@ class Loader:
"""
return web.Response(text=ret, content_type="text/html")
- async def handle_sub_route(self, request):
+ async def handle_sub_route(self, request: web.Request):
plugin = self.plugins[request.match_info["name"]]
route_path = request.match_info["path"]
self.logger.info(path)
@@ -212,14 +222,14 @@ class Loader:
return web.Response(text=ret)
- async def get_steam_resource(self, request):
+ async def get_steam_resource(self, request: web.Request):
tab = await get_tab("SP")
try:
return web.Response(text=await tab.get_steam_resource(f"https://steamloopback.host/{request.match_info['path']}"), content_type="text/html")
except Exception as e:
return web.Response(text=str(e), status=400)
- async def handle_backend_reload_request(self, request):
+ async def handle_backend_reload_request(self, request: web.Request):
plugin_name : str = request.match_info["plugin_name"]
plugin = self.plugins[plugin_name]
diff --git a/backend/localplatform.py b/backend/src/localplatform.py
index 43043ad0..028eff8f 100644
--- a/backend/localplatform.py
+++ b/backend/src/localplatform.py
@@ -4,11 +4,11 @@ ON_WINDOWS = platform.system() == "Windows"
ON_LINUX = not ON_WINDOWS
if ON_WINDOWS:
- from localplatformwin import *
- import localplatformwin as localplatform
+ from .localplatformwin import *
+ from . import localplatformwin as localplatform
else:
- from localplatformlinux import *
- import localplatformlinux as localplatform
+ from .localplatformlinux import *
+ from . import localplatformlinux as localplatform
def get_privileged_path() -> str:
'''Get path accessible by elevated user. Holds plugins, decky loader and decky loader configs'''
diff --git a/backend/localplatformlinux.py b/backend/src/localplatformlinux.py
index 811db8a6..bde2caac 100644
--- a/backend/localplatformlinux.py
+++ b/backend/src/localplatformlinux.py
@@ -1,6 +1,6 @@
import os, pwd, grp, sys, logging
from subprocess import call, run, DEVNULL, PIPE, STDOUT
-from customtypes import UserType
+from .customtypes import UserType
logger = logging.getLogger("localplatform")
@@ -29,21 +29,17 @@ def _get_effective_user_group() -> str:
return grp.getgrgid(_get_effective_user_group_id()).gr_name
# Get the user owner of the given file path.
-def _get_user_owner(file_path) -> str:
+def _get_user_owner(file_path: str) -> str:
return pwd.getpwuid(os.stat(file_path).st_uid).pw_name
-# Get the user group of the given file path.
-def _get_user_group(file_path) -> str:
- return grp.getgrgid(os.stat(file_path).st_gid).gr_name
+# Get the user group of the given file path, or the user group hosting the plugin loader
+def _get_user_group(file_path: str | None = None) -> str:
+ return grp.getgrgid(os.stat(file_path).st_gid if file_path is not None else _get_user_group_id()).gr_name
# Get the group id of the user hosting the plugin loader
def _get_user_group_id() -> int:
return pwd.getpwuid(_get_user_id()).pw_gid
-# Get the group of the user hosting the plugin loader
-def _get_user_group() -> str:
- return grp.getgrgid(_get_user_group_id()).gr_name
-
def chown(path : str, user : UserType = UserType.HOST_USER, recursive : bool = True) -> bool:
user_str = ""
@@ -146,7 +142,7 @@ def get_privileged_path() -> str:
return path
-def _parent_dir(path : str) -> str:
+def _parent_dir(path : str | None) -> str | None:
if path == None:
return None
@@ -166,7 +162,7 @@ def get_unprivileged_path() -> str:
# Expected path of loader binary is /home/deck/homebrew/service/PluginLoader
path = _parent_dir(_parent_dir(os.path.realpath(sys.argv[0])))
- if not os.path.exists(path):
+ if path != None and not os.path.exists(path):
path = None
if path == None:
diff --git a/backend/localplatformwin.py b/backend/src/localplatformwin.py
index b6bee330..4c4e9439 100644
--- a/backend/localplatformwin.py
+++ b/backend/src/localplatformwin.py
@@ -1,4 +1,4 @@
-from customtypes import UserType
+from .customtypes import UserType
import os, sys
def chown(path : str, user : UserType = UserType.HOST_USER, recursive : bool = True) -> bool:
diff --git a/backend/localsocket.py b/backend/src/localsocket.py
index ef0e3933..f38fe5e7 100644
--- a/backend/localsocket.py
+++ b/backend/src/localsocket.py
@@ -1,10 +1,13 @@
-import asyncio, time, random
-from localplatform import ON_WINDOWS
+import asyncio, time
+from typing import Awaitable, Callable
+import random
+
+from .localplatform import ON_WINDOWS
BUFFER_LIMIT = 2 ** 20 # 1 MiB
class UnixSocket:
- def __init__(self, on_new_message):
+ def __init__(self, on_new_message: Callable[[str], Awaitable[str|None]]):
'''
on_new_message takes 1 string argument.
It's return value gets used, if not None, to write data to the socket.
@@ -46,28 +49,32 @@ class UnixSocket:
self.reader = None
async def read_single_line(self) -> str|None:
- reader, writer = await self.get_socket_connection()
+ reader, _ = await self.get_socket_connection()
- if self.reader == None:
- return None
+ try:
+ assert reader
+ except AssertionError:
+ return
return await self._read_single_line(reader)
async def write_single_line(self, message : str):
- reader, writer = await self.get_socket_connection()
+ _, writer = await self.get_socket_connection()
- if self.writer == None:
- return;
+ try:
+ assert writer
+ except AssertionError:
+ return
await self._write_single_line(writer, message)
- async def _read_single_line(self, reader) -> str:
+ async def _read_single_line(self, reader: asyncio.StreamReader) -> str:
line = bytearray()
while True:
try:
line.extend(await reader.readuntil())
except asyncio.LimitOverrunError:
- line.extend(await reader.read(reader._limit))
+ line.extend(await reader.read(reader._limit)) # type: ignore
continue
except asyncio.IncompleteReadError as err:
line.extend(err.partial)
@@ -77,27 +84,27 @@ class UnixSocket:
return line.decode("utf-8")
- async def _write_single_line(self, writer, message : str):
+ async def _write_single_line(self, writer: asyncio.StreamWriter, message : str):
if not message.endswith("\n"):
message += "\n"
writer.write(message.encode("utf-8"))
await writer.drain()
- async def _listen_for_method_call(self, reader, writer):
+ async def _listen_for_method_call(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
while True:
line = await self._read_single_line(reader)
try:
res = await self.on_new_message(line)
- except Exception as e:
+ except Exception:
return
if res != None:
await self._write_single_line(writer, res)
class PortSocket (UnixSocket):
- def __init__(self, on_new_message):
+ def __init__(self, on_new_message: Callable[[str], Awaitable[str|None]]):
'''
on_new_message takes 1 string argument.
It's return value gets used, if not None, to write data to the socket.
@@ -125,7 +132,7 @@ class PortSocket (UnixSocket):
return True
if ON_WINDOWS:
- class LocalSocket (PortSocket):
+ class LocalSocket (PortSocket): # type: ignore
pass
else:
class LocalSocket (UnixSocket):
diff --git a/backend/src/main.py b/backend/src/main.py
new file mode 100644
index 00000000..83a4b997
--- /dev/null
+++ b/backend/src/main.py
@@ -0,0 +1,192 @@
+# Change PyInstaller files permissions
+import sys
+from typing import Dict
+from .localplatform import (chmod, chown, service_stop, service_start,
+ ON_WINDOWS, get_log_level, get_live_reload,
+ get_server_port, get_server_host, get_chown_plugin_path,
+ get_privileged_path)
+if hasattr(sys, '_MEIPASS'):
+ chmod(sys._MEIPASS, 755) # type: ignore
+# Full imports
+from asyncio import AbstractEventLoop, new_event_loop, set_event_loop, sleep
+from logging import basicConfig, getLogger
+from os import path
+from traceback import format_exc
+import multiprocessing
+
+import aiohttp_cors # type: ignore
+# Partial imports
+from aiohttp import client_exceptions
+from aiohttp.web import Application, Response, Request, get, run_app, static # type: ignore
+from aiohttp_jinja2 import setup as jinja_setup
+
+# local modules
+from .browser import PluginBrowser
+from .helpers import (REMOTE_DEBUGGER_UNIT, csrf_middleware, get_csrf_token,
+ mkdir_as_user, get_system_pythonpaths, get_effective_user_id)
+
+from .injector import get_gamepadui_tab, Tab, close_old_tabs
+from .loader import Loader
+from .settings import SettingsManager
+from .updater import Updater
+from .utilities import Utilities
+from .customtypes import UserType
+
+
+basicConfig(
+ level=get_log_level(),
+ format="[%(module)s][%(levelname)s]: %(message)s"
+)
+
+logger = getLogger("Main")
+plugin_path = path.join(get_privileged_path(), "plugins")
+
+def chown_plugin_dir():
+ if not path.exists(plugin_path): # For safety, create the folder before attempting to do anything with it
+ mkdir_as_user(plugin_path)
+
+ if not chown(plugin_path, UserType.HOST_USER) or not chmod(plugin_path, 555):
+ logger.error(f"chown/chmod exited with a non-zero exit code")
+
+if get_chown_plugin_path() == True:
+ chown_plugin_dir()
+
+class PluginManager:
+ def __init__(self, loop: AbstractEventLoop) -> None:
+ self.loop = loop
+ self.web_app = Application()
+ self.web_app.middlewares.append(csrf_middleware)
+ self.cors = aiohttp_cors.setup(self.web_app, defaults={
+ "https://steamloopback.host": aiohttp_cors.ResourceOptions(
+ expose_headers="*",
+ allow_headers="*",
+ allow_credentials=True
+ )
+ })
+ self.plugin_loader = Loader(self, plugin_path, self.loop, get_live_reload())
+ self.settings = SettingsManager("loader", path.join(get_privileged_path(), "settings"))
+ self.plugin_browser = PluginBrowser(plugin_path, self.plugin_loader.plugins, self.plugin_loader, self.settings)
+ self.utilities = Utilities(self)
+ self.updater = Updater(self)
+
+ jinja_setup(self.web_app)
+
+ async def startup(_: Application):
+ if self.settings.getSetting("cef_forward", False):
+ self.loop.create_task(service_start(REMOTE_DEBUGGER_UNIT))
+ else:
+ self.loop.create_task(service_stop(REMOTE_DEBUGGER_UNIT))
+ self.loop.create_task(self.loader_reinjector())
+ self.loop.create_task(self.load_plugins())
+
+ self.web_app.on_startup.append(startup)
+
+ self.loop.set_exception_handler(self.exception_handler)
+ self.web_app.add_routes([get("/auth/token", self.get_auth_token)])
+
+ for route in list(self.web_app.router.routes()):
+ self.cors.add(route) # type: ignore
+ self.web_app.add_routes([static("/static", path.join(path.dirname(__file__), '..', 'static'))])
+ self.web_app.add_routes([static("/legacy", path.join(path.dirname(__file__), 'legacy'))])
+
+ def exception_handler(self, loop: AbstractEventLoop, context: Dict[str, str]):
+ if context["message"] == "Unclosed connection":
+ return
+ loop.default_exception_handler(context)
+
+ async def get_auth_token(self, request: Request):
+ return Response(text=get_csrf_token())
+
+ async def load_plugins(self):
+ # await self.wait_for_server()
+ logger.debug("Loading plugins")
+ self.plugin_loader.import_plugins()
+ # await inject_to_tab("SP", "window.syncDeckyPlugins();")
+ if self.settings.getSetting("pluginOrder", None) == None:
+ self.settings.setSetting("pluginOrder", list(self.plugin_loader.plugins.keys()))
+ logger.debug("Did not find pluginOrder setting, set it to default")
+
+ async def loader_reinjector(self):
+ while True:
+ tab = None
+ nf = False
+ dc = False
+ while not tab:
+ try:
+ tab = await get_gamepadui_tab()
+ except (client_exceptions.ClientConnectorError, client_exceptions.ServerDisconnectedError):
+ if not dc:
+ logger.debug("Couldn't connect to debugger, waiting...")
+ dc = True
+ pass
+ except ValueError:
+ if not nf:
+ logger.debug("Couldn't find GamepadUI tab, waiting...")
+ nf = True
+ pass
+ if not tab:
+ await sleep(5)
+ await tab.open_websocket()
+ await tab.enable()
+ await self.inject_javascript(tab, True)
+ try:
+ async for msg in tab.listen_for_message():
+ # this gets spammed a lot
+ if msg.get("method", None) != "Page.navigatedWithinDocument":
+ logger.debug("Page event: " + str(msg.get("method", None)))
+ if msg.get("method", None) == "Page.domContentEventFired":
+ if not await tab.has_global_var("deckyHasLoaded", False):
+ await self.inject_javascript(tab)
+ if msg.get("method", None) == "Inspector.detached":
+ logger.info("CEF has requested that we detach.")
+ await tab.close_websocket()
+ break
+ # If this is a forceful disconnect the loop will just stop without any failure message. In this case, injector.py will handle this for us so we don't need to close the socket.
+ # This is because of https://github.com/aio-libs/aiohttp/blob/3ee7091b40a1bc58a8d7846e7878a77640e96996/aiohttp/client_ws.py#L321
+ logger.info("CEF has disconnected...")
+ # At this point the loop starts again and we connect to the freshly started Steam client once it is ready.
+ except Exception:
+ logger.error("Exception while reading page events " + format_exc())
+ await tab.close_websocket()
+ pass
+ # while True:
+ # await sleep(5)
+ # if not await tab.has_global_var("deckyHasLoaded", False):
+ # logger.info("Plugin loader isn't present in Steam anymore, reinjecting...")
+ # await self.inject_javascript(tab)
+
+ async def inject_javascript(self, tab: Tab, first: bool=False, request: Request|None=None):
+ logger.info("Loading Decky frontend!")
+ try:
+ if first:
+ if await tab.has_global_var("deckyHasLoaded", False):
+ await close_old_tabs()
+ await tab.evaluate_js("try{if (window.deckyHasLoaded){setTimeout(() => location.reload(), 100)}else{window.deckyHasLoaded = true;(async()=>{try{while(!window.SP_REACT){await new Promise(r => setTimeout(r, 10))};await import('http://localhost:1337/frontend/index.js')}catch(e){console.error(e)};})();}}catch(e){console.error(e)}", False, False, False)
+ except:
+ logger.info("Failed to inject JavaScript into tab\n" + format_exc())
+ pass
+
+ def run(self):
+ return run_app(self.web_app, host=get_server_host(), port=get_server_port(), loop=self.loop, access_log=None)
+
+def main():
+ if ON_WINDOWS:
+ # Fix windows/flask not recognising that .js means 'application/javascript'
+ import mimetypes
+ mimetypes.add_type('application/javascript', '.js')
+
+ # Required for multiprocessing support in frozen files
+ multiprocessing.freeze_support()
+ else:
+ if get_effective_user_id() != 0:
+ logger.warning(f"decky is running as an unprivileged user, this is not officially supported and may cause issues")
+
+ # Append the loader's plugin path to the recognized python paths
+ sys.path.append(path.join(path.dirname(__file__), "plugin"))
+
+ # Append the system and user python paths
+ sys.path.extend(get_system_pythonpaths())
+
+ loop = new_event_loop()
+ set_event_loop(loop)
+ PluginManager(loop).run()
diff --git a/backend/plugin.py b/backend/src/plugin.py
index 026a6b09..163bb9b6 100644
--- a/backend/plugin.py
+++ b/backend/src/plugin.py
@@ -1,7 +1,6 @@
import multiprocessing
from asyncio import (Lock, get_event_loop, new_event_loop,
set_event_loop, sleep)
-from concurrent.futures import ProcessPoolExecutor
from importlib.util import module_from_spec, spec_from_file_location
from json import dumps, load, loads
from logging import getLogger
@@ -9,19 +8,19 @@ from traceback import format_exc
from os import path, environ
from signal import SIGINT, signal
from sys import exit, path as syspath
-from time import time
-from localsocket import LocalSocket
-from localplatform import setgid, setuid, get_username, get_home_path
-from customtypes import UserType
-import helpers
+from typing import Any, Dict
+from .localsocket import LocalSocket
+from .localplatform import setgid, setuid, get_username, get_home_path
+from .customtypes import UserType
+from . import helpers
class PluginWrapper:
- def __init__(self, file, plugin_directory, plugin_path) -> None:
+ 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.method_call_lock = Lock()
- self.socket = LocalSocket(self._on_new_message)
+ self.socket: LocalSocket = LocalSocket(self._on_new_message)
self.version = None
@@ -73,14 +72,17 @@ class PluginWrapper:
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
- environ["DECKY_PLUGIN_VERSION"] = self.version
+ 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
@@ -118,7 +120,8 @@ class PluginWrapper:
get_event_loop().close()
raise Exception("Closing message listener")
- d = {"res": None, "success": True}
+ # TODO there is definitely a better way to type this
+ d: Dict[str, Any] = {"res": None, "success": True}
try:
d["res"] = await getattr(self.Plugin, data["method"])(self.Plugin, **data["args"])
except Exception as e:
@@ -137,17 +140,18 @@ class PluginWrapper:
if self.passive:
return
- async def _(self):
+ async def _(self: PluginWrapper):
await self.socket.write_single_line(dumps({ "stop": True }, ensure_ascii=False))
await self.socket.close_socket_connection()
get_event_loop().create_task(_(self))
- async def execute_method(self, method_name, kwargs):
+ 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)")
async with self.method_call_lock:
- reader, writer = await self.socket.get_socket_connection()
+ # reader, writer =
+ await self.socket.get_socket_connection()
await self.socket.write_single_line(dumps({ "method": method_name, "args": kwargs }, ensure_ascii=False))
diff --git a/backend/settings.py b/backend/src/settings.py
index c00e6a82..a9ab3daa 100644
--- a/backend/settings.py
+++ b/backend/src/settings.py
@@ -1,13 +1,14 @@
from json import dump, load
from os import mkdir, path, listdir, rename
-from localplatform import chown, folder_owner, get_chown_plugin_path
-from customtypes import UserType
+from typing import Any, Dict
+from .localplatform import chown, folder_owner, get_chown_plugin_path
+from .customtypes import UserType
-from helpers import get_homebrew_path
+from .helpers import get_homebrew_path
class SettingsManager:
- def __init__(self, name, settings_directory = None) -> None:
+ def __init__(self, name: str, settings_directory: str | None = None) -> None:
wrong_dir = get_homebrew_path()
if settings_directory == None:
settings_directory = path.join(wrong_dir, "settings")
@@ -31,11 +32,11 @@ class SettingsManager:
if folder_owner(settings_directory) != expected_user:
chown(settings_directory, expected_user, False)
- self.settings = {}
+ self.settings: Dict[str, Any] = {}
try:
open(self.path, "x", encoding="utf-8")
- except FileExistsError as e:
+ except FileExistsError as _:
self.read()
pass
@@ -51,9 +52,9 @@ class SettingsManager:
with open(self.path, "w+", encoding="utf-8") as file:
dump(self.settings, file, indent=4, ensure_ascii=False)
- def getSetting(self, key, default=None):
+ def getSetting(self, key: str, default: Any = None) -> Any:
return self.settings.get(key, default)
- def setSetting(self, key, value):
+ def setSetting(self, key: str, value: Any) -> Any:
self.settings[key] = value
self.commit()
diff --git a/backend/updater.py b/backend/src/updater.py
index 6b38dd25..d28e67b0 100644
--- a/backend/updater.py
+++ b/backend/src/updater.py
@@ -1,23 +1,33 @@
+from __future__ import annotations
import os
import shutil
-import uuid
from asyncio import sleep
-from ensurepip import version
from json.decoder import JSONDecodeError
from logging import getLogger
from os import getcwd, path, remove
-from localplatform import chmod, service_restart, ON_LINUX, get_keep_systemd_service, get_selinux
+from typing import TYPE_CHECKING, List, TypedDict
+if TYPE_CHECKING:
+ from .main import PluginManager
+from .localplatform import chmod, service_restart, ON_LINUX, get_keep_systemd_service, get_selinux
from aiohttp import ClientSession, web
-import helpers
-from injector import get_gamepadui_tab, inject_to_tab
-from settings import SettingsManager
+from . import helpers
+from .injector import get_gamepadui_tab
+from .settings import SettingsManager
logger = getLogger("Updater")
+class RemoteVerAsset(TypedDict):
+ name: str
+ browser_download_url: str
+class RemoteVer(TypedDict):
+ tag_name: str
+ prerelease: bool
+ assets: List[RemoteVerAsset]
+
class Updater:
- def __init__(self, context) -> None:
+ def __init__(self, context: PluginManager) -> None:
self.context = context
self.settings = self.context.settings
# Exposes updater methods to frontend
@@ -28,8 +38,8 @@ class Updater:
"do_restart": self.do_restart,
"check_for_updates": self.check_for_updates
}
- self.remoteVer = None
- self.allRemoteVers = None
+ self.remoteVer: RemoteVer | None = None
+ self.allRemoteVers: List[RemoteVer] = []
self.localVer = helpers.get_loader_version()
try:
@@ -44,7 +54,7 @@ class Updater:
])
context.loop.create_task(self.version_reloader())
- async def _handle_server_method_call(self, request):
+ async def _handle_server_method_call(self, request: web.Request):
method_name = request.match_info["method_name"]
try:
args = await request.json()
@@ -52,7 +62,7 @@ class Updater:
args = {}
res = {}
try:
- r = await self.updater_methods[method_name](**args)
+ r = await self.updater_methods[method_name](**args) # type: ignore
res["result"] = r
res["success"] = True
except Exception as e:
@@ -105,7 +115,7 @@ class Updater:
selectedBranch = self.get_branch(self.context.settings)
async with ClientSession() as web:
async with web.request("GET", "https://api.github.com/repos/SteamDeckHomebrew/decky-loader/releases", ssl=helpers.get_ssl_context()) as res:
- remoteVersions = await res.json()
+ remoteVersions: List[RemoteVer] = await res.json()
if selectedBranch == 0:
logger.debug("release type: release")
remoteVersions = list(filter(lambda ver: ver["tag_name"].startswith("v") and not ver["prerelease"] and not ver["tag_name"].find("-pre") > 0 and ver["tag_name"], remoteVersions))
@@ -142,6 +152,12 @@ class Updater:
async def do_update(self):
logger.debug("Starting update.")
+ try:
+ assert self.remoteVer
+ except AssertionError:
+ logger.error("Unable to update as remoteVer is missing")
+ return
+
version = self.remoteVer["tag_name"]
download_url = None
download_filename = "PluginLoader" if ON_LINUX else "PluginLoader.exe"
diff --git a/backend/utilities.py b/backend/src/utilities.py
index bcb35578..b0e23b88 100644
--- a/backend/utilities.py
+++ b/backend/src/utilities.py
@@ -1,26 +1,36 @@
+from __future__ import annotations
+from os import stat_result
import uuid
-import os
from json.decoder import JSONDecodeError
from os.path import splitext
import re
from traceback import format_exc
-from stat import FILE_ATTRIBUTE_HIDDEN
+from stat import FILE_ATTRIBUTE_HIDDEN # type: ignore
-from asyncio import sleep, start_server, gather, open_connection
+from asyncio import StreamReader, StreamWriter, start_server, gather, open_connection
from aiohttp import ClientSession, web
+from typing import TYPE_CHECKING, Callable, Coroutine, Dict, Any, List, TypedDict
from logging import getLogger
-from injector import inject_to_tab, get_gamepadui_tab, close_old_tabs, get_tab
from pathlib import Path
-from localplatform import ON_WINDOWS
-import helpers
-import subprocess
-from localplatform import service_stop, service_start, get_home_path, get_username
+
+from .browser import PluginInstallRequest, PluginInstallType
+if TYPE_CHECKING:
+ from .main import PluginManager
+from .injector import inject_to_tab, get_gamepadui_tab, close_old_tabs, get_tab
+from .localplatform import ON_WINDOWS
+from . import helpers
+from .localplatform import service_stop, service_start, get_home_path, get_username
+
+class FilePickerObj(TypedDict):
+ file: Path
+ filest: stat_result
+ is_dir: bool
class Utilities:
- def __init__(self, context) -> None:
+ def __init__(self, context: PluginManager) -> None:
self.context = context
- self.util_methods = {
+ self.util_methods: Dict[str, Callable[..., Coroutine[Any, Any, Any]]] = {
"ping": self.ping,
"http_request": self.http_request,
"install_plugin": self.install_plugin,
@@ -53,7 +63,7 @@ class Utilities:
web.post("/methods/{method_name}", self._handle_server_method_call)
])
- async def _handle_server_method_call(self, request):
+ async def _handle_server_method_call(self, request: web.Request):
method_name = request.match_info["method_name"]
try:
args = await request.json()
@@ -69,7 +79,7 @@ class Utilities:
res["success"] = False
return web.json_response(res)
- async def install_plugin(self, artifact="", name="No name", version="dev", hash=False, install_type=0):
+ async def install_plugin(self, artifact: str="", name: str="No name", version: str="dev", hash: str="", install_type: PluginInstallType=PluginInstallType.INSTALL):
return await self.context.plugin_browser.request_plugin_install(
artifact=artifact,
name=name,
@@ -78,21 +88,21 @@ class Utilities:
install_type=install_type
)
- async def install_plugins(self, requests):
+ async def install_plugins(self, requests: List[PluginInstallRequest]):
return await self.context.plugin_browser.request_multiple_plugin_installs(
requests=requests
)
- async def confirm_plugin_install(self, request_id):
+ async def confirm_plugin_install(self, request_id: str):
return await self.context.plugin_browser.confirm_plugin_install(request_id)
- def cancel_plugin_install(self, request_id):
+ async def cancel_plugin_install(self, request_id: str):
return self.context.plugin_browser.cancel_plugin_install(request_id)
- async def uninstall_plugin(self, name):
+ async def uninstall_plugin(self, name: str):
return await self.context.plugin_browser.uninstall_plugin(name)
- async def http_request(self, method="", url="", **kwargs):
+ async def http_request(self, method: str="", url: str="", **kwargs: Any):
async with ClientSession() as web:
res = await web.request(method, url, ssl=helpers.get_ssl_context(), **kwargs)
text = await res.text()
@@ -102,12 +112,13 @@ class Utilities:
"body": text
}
- async def ping(self, **kwargs):
+ async def ping(self, **kwargs: Any):
return "pong"
- async def execute_in_tab(self, tab, run_async, code):
+ async def execute_in_tab(self, tab: str, run_async: bool, code: str):
try:
result = await inject_to_tab(tab, code, run_async)
+ assert result
if "exceptionDetails" in result["result"]:
return {
"success": False,
@@ -124,7 +135,7 @@ class Utilities:
"result": e
}
- async def inject_css_into_tab(self, tab, style):
+ async def inject_css_into_tab(self, tab: str, style: str):
try:
css_id = str(uuid.uuid4())
@@ -138,7 +149,7 @@ class Utilities:
}})()
""", False)
- if "exceptionDetails" in result["result"]:
+ if result and "exceptionDetails" in result["result"]:
return {
"success": False,
"result": result["result"]
@@ -154,7 +165,7 @@ class Utilities:
"result": e
}
- async def remove_css_from_tab(self, tab, css_id):
+ async def remove_css_from_tab(self, tab: str, css_id: str):
try:
result = await inject_to_tab(tab,
f"""
@@ -166,7 +177,7 @@ class Utilities:
}})()
""", False)
- if "exceptionDetails" in result["result"]:
+ if result and "exceptionDetails" in result["result"]:
return {
"success": False,
"result": result
@@ -181,10 +192,10 @@ class Utilities:
"result": e
}
- async def get_setting(self, key, default):
+ async def get_setting(self, key: str, default: Any):
return self.context.settings.getSetting(key, default)
- async def set_setting(self, key, value):
+ async def set_setting(self, key: str, value: Any):
return self.context.settings.setSetting(key, value)
async def allow_remote_debugging(self):
@@ -209,17 +220,18 @@ class Utilities:
if path == None:
path = get_home_path()
- path = Path(path).resolve()
+ path_obj = Path(path).resolve()
- files, folders = [], []
+ files: List[FilePickerObj] = []
+ folders: List[FilePickerObj] = []
#Resolving all files/folders in the requested directory
- for file in path.iterdir():
+ for file in path_obj.iterdir():
if file.exists():
filest = file.stat()
is_hidden = file.name.startswith('.')
if ON_WINDOWS and not is_hidden:
- is_hidden = bool(filest.st_file_attributes & FILE_ATTRIBUTE_HIDDEN)
+ is_hidden = bool(filest.st_file_attributes & FILE_ATTRIBUTE_HIDDEN) # type: ignore
if include_folders and file.is_dir():
if (is_hidden and include_hidden) or not is_hidden:
folders.append({"file": file, "filest": filest, "is_dir": True})
@@ -233,9 +245,9 @@ class Utilities:
if filter_for is not None:
try:
if re.compile(filter_for):
- files = filter(lambda file: re.search(filter_for, file.name) != None, files)
+ files = list(filter(lambda file: re.search(filter_for, file["file"].name) != None, files))
except re.error:
- files = filter(lambda file: file.name.find(filter_for) != -1, files)
+ files = list(filter(lambda file: file["file"].name.find(filter_for) != -1, files))
# Ordering logic
ord_arg = order_by.split("_")
@@ -255,6 +267,9 @@ class Utilities:
files.sort(key=lambda x: x['filest'].st_size, reverse = not rev)
# Folders has no file size, order by name instead
folders.sort(key=lambda x: x['file'].name.casefold())
+ case _:
+ files.sort(key=lambda x: x['file'].name.casefold(), reverse = rev)
+ folders.sort(key=lambda x: x['file'].name.casefold(), reverse = rev)
#Constructing the final file list, folders first
all = [{
@@ -274,14 +289,14 @@ class Utilities:
# Based on https://stackoverflow.com/a/46422554/13174603
- def start_rdt_proxy(self, ip, port):
- async def pipe(reader, writer):
+ def start_rdt_proxy(self, ip: str, port: int):
+ async def pipe(reader: StreamReader, writer: StreamWriter):
try:
while not reader.at_eof():
writer.write(await reader.read(2048))
finally:
writer.close()
- async def handle_client(local_reader, local_writer):
+ async def handle_client(local_reader: StreamReader, local_writer: StreamWriter):
try:
remote_reader, remote_writer = await open_connection(
ip, port)
@@ -295,9 +310,10 @@ class Utilities:
self.rdt_proxy_task = self.context.loop.create_task(self.rdt_proxy_server)
def stop_rdt_proxy(self):
- if self.rdt_proxy_server:
+ if self.rdt_proxy_server != None:
self.rdt_proxy_server.close()
- self.rdt_proxy_task.cancel()
+ if self.rdt_proxy_task:
+ self.rdt_proxy_task.cancel()
async def _enable_rdt(self):
# TODO un-hardcode port
@@ -347,11 +363,11 @@ class Utilities:
await tab.evaluate_js("location.reload();", False, True, False)
self.logger.info("React DevTools disabled")
- async def get_user_info(self) -> dict:
+ async def get_user_info(self) -> Dict[str, str]:
return {
"username": get_username(),
"path": get_home_path()
}
- async def get_tab_id(self, name):
+ async def get_tab_id(self, name: str):
return (await get_tab(name)).id
generated by cgit v1.2.3 (git 2.25.1) at 2025-12-15 17:08:39 +0000