Merge commit from fork

* fix incorrect permissions on plugin directories

* chown plugin dirs too

* fix the stupid

* cleanup useless comments

Author: AAGaming00

backend/decky_loader/browser.py

@@ -18,9 +18,10 @@
 from typing import Dict, List, TypedDict
 
 # Local modules
-from .localplatform.localplatform import chown, chmod
+from .localplatform.localplatform import chown, chmod, get_chown_plugin_path
 from .loader import Loader, Plugins
 from .helpers import get_ssl_context, download_remote_binary_to_path
+from .enums import UserType
 from .settings import SettingsManager
 
 logger = getLogger("Browser")
@@ -60,13 +61,6 @@ def _unzip_to_plugin_dir(self, zip: BytesIO, name: str, hash: str):
             return False
         zip_file = ZipFile(zip)
         zip_file.extractall(self.plugin_path)
-        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: str):
@@ -101,8 +95,6 @@ async def _download_remote_binaries_for_plugin_with_name(self, pluginBasePath: s
                                 rv = False
                                 raise Exception(f"Error Downloading Remote Binary {binName}@{binURL} with hash {binHash} to {path.join(pluginBinPath, binName)}")
 
-                        chown(self.plugin_path)
-                        chmod(pluginBasePath, 555)
                     else:
                         rv = True
                         logger.info(f"No Remote Binaries to Download")
@@ -124,6 +116,25 @@ def find_plugin_folder(self, name: str) -> str | None:
                     return folder
             except:
                 logger.debug(f"skipping {folder}")
+    
+    def set_plugin_dir_permissions(self, plugin_dir: str) -> bool:
+        plugin_json_path = path.join(plugin_dir, 'plugin.json')
+        logger.debug(f"Checking plugin.json at {plugin_json_path}")
+
+        root_plugin = False
+
+        if access(plugin_json_path, R_OK):
+            with open(plugin_json_path, "r", encoding="utf-8") as f:
+                plugin_json = json.load(f)
+                if "flags" in plugin_json and "root" in plugin_json["flags"]:
+                    root_plugin = True
+
+        logger.debug("root_plugin %d, dir %s", root_plugin, plugin_dir)
+        if get_chown_plugin_path():
+            return chown(plugin_dir, UserType.EFFECTIVE_USER if root_plugin else UserType.HOST_USER, True) and chown(plugin_dir, UserType.EFFECTIVE_USER, False) and chmod(plugin_dir, 755) and chown(plugin_json_path, UserType.EFFECTIVE_USER, False) and chmod(plugin_json_path, 755)
+        else:
+            logger.debug("chown disabled by environment")
+            return True
 
     async def uninstall_plugin(self, name: str):
         if self.loader.watcher:
@@ -266,6 +277,7 @@ async def _install(self, artifact: str, name: str, version: str, hash: str):
             plugin_dir = path.join(self.plugin_path, plugin_folder)
             await self.loader.ws.emit("loader/plugin_download_info", 95, "Store.download_progress_info.download_remote")
             ret = await self._download_remote_binaries_for_plugin_with_name(plugin_dir)
+            chown_ret = self.set_plugin_dir_permissions(plugin_dir)
             if ret:
                 logger.info(f"Installed {name} (Version: {version})")
                 if name in self.loader.plugins:
@@ -278,6 +290,9 @@ async def _install(self, artifact: str, name: str, version: str, hash: str):
                     self.settings.setSetting("pluginOrder", current_plugin_order)
                     logger.debug("Plugin %s was added to the pluginOrder setting", name)
                 await self.loader.import_plugin(path.join(plugin_dir, "main.py"), plugin_folder)
+            elif not chown_ret:
+                logger.error("Could not chown plugin")
+                return
             else:
                 logger.error("Could not download remote binaries")
                 return

backend/decky_loader/enums.py

@@ -1,9 +1,8 @@
 from enum import IntEnum
 
 class UserType(IntEnum):
-    HOST_USER = 1
-    EFFECTIVE_USER = 2
-    ROOT = 3
+    HOST_USER = 1 # usually deck
+    EFFECTIVE_USER = 2 # usually root
 
 class PluginLoadType(IntEnum):
     LEGACY_EVAL_IIFE = 0 # legacy, uses legacy serverAPI

backend/decky_loader/helpers.py

@@ -181,7 +181,8 @@ def get_user_group_id() -> int:
 
 # Get the default home path unless a user is specified
 def get_home_path(username: str | None = None) -> str:
-    return localplatform.get_home_path(UserType.ROOT if username == "root" else UserType.HOST_USER)
+    # TODO hardcoded root is kinda a hack
+    return localplatform.get_home_path(UserType.EFFECTIVE_USER if username == "root" else UserType.HOST_USER)
 
 async def is_systemd_unit_active(unit_name: str) -> bool:
     return await localplatform.service_active(unit_name)

backend/decky_loader/localplatform/localplatformlinux.py

@@ -59,8 +59,6 @@ def chown(path : str,  user : UserType = UserType.HOST_USER, recursive : bool =
         user_str = _get_user()+":"+_get_user_group()
     elif user == UserType.EFFECTIVE_USER:
         user_str = _get_effective_user()+":"+_get_effective_user_group()
-    elif user == UserType.ROOT:
-        user_str = "root:root"
     else:
         raise Exception("Unknown User Type")
 
@@ -87,7 +85,7 @@ def chmod(path : str, permissions : int, recursive : bool = True) -> bool:
 
     return True
 
-def folder_owner(path : str) -> UserType|None:
+def file_owner(path : str) -> UserType|None:
     user_owner = _get_user_owner(path)
 
     if (user_owner == _get_user()):
@@ -106,13 +104,14 @@ def get_home_path(user : UserType = UserType.HOST_USER) -> str:
         user_name = _get_user()
     elif user == UserType.EFFECTIVE_USER:
         user_name = _get_effective_user()
-    elif user == UserType.ROOT:
-        pass
     else:
         raise Exception("Unknown User Type")
 
     return pwd.getpwnam(user_name).pw_dir
 
+def get_effective_username() -> str:
+    return _get_effective_user()
+
 def get_username() -> str:
     return _get_user()
 
@@ -121,8 +120,8 @@ def setgid(user : UserType = UserType.HOST_USER):
 
     if user == UserType.HOST_USER:
         user_id = _get_user_group_id()
-    elif user == UserType.ROOT:
-        pass
+    elif user == UserType.EFFECTIVE_USER:
+        pass # we already are
     else:
         raise Exception("Unknown user type")
 
@@ -133,8 +132,8 @@ def setuid(user : UserType = UserType.HOST_USER):
 
     if user == UserType.HOST_USER:
         user_id = _get_user_id()
-    elif user == UserType.ROOT:
-        pass
+    elif user == UserType.EFFECTIVE_USER:
+        pass # we already are
     else:
         raise Exception("Unknown user type")
 

backend/decky_loader/localplatform/localplatformwin.py

@@ -7,7 +7,7 @@ def chown(path : str,  user : UserType = UserType.HOST_USER, recursive : bool =
 def chmod(path : str, permissions : int, recursive : bool = True) -> bool:
     return True # Stubbed
 
-def folder_owner(path : str) -> UserType|None:
+def file_owner(path : str) -> UserType|None:
     return UserType.HOST_USER # Stubbed
 
 def get_home_path(user : UserType = UserType.HOST_USER) -> str:
@@ -34,6 +34,9 @@ async def service_restart(service_name : str, block : bool = True) -> bool:
 
     return True # Stubbed
 
+def get_effective_username() -> str:
+    return os.getlogin()
+
 def get_username() -> str:
     return os.getlogin()
 

backend/decky_loader/main.py

@@ -50,7 +50,7 @@ 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):
+    if not chown(plugin_path, UserType.EFFECTIVE_USER, False) or not chmod(plugin_path, 755, False):
         logger.error(f"chown/chmod exited with a non-zero exit code")
 
 if get_chown_plugin_path() == True:

backend/decky_loader/plugin/plugin.py

@@ -8,7 +8,8 @@
 
 from .sandboxed_plugin import SandboxedPlugin
 from .messages import MethodCallRequest, SocketMessageType
-from ..enums import PluginLoadType
+from ..enums import PluginLoadType, UserType
+from ..localplatform.localplatform import file_owner, chown, chmod, get_chown_plugin_path
 from ..localplatform.localsocket import LocalSocket
 from ..helpers import get_homebrew_path, mkdir_as_user
 
@@ -26,9 +27,12 @@ def __init__(self, file: str, plugin_directory: str, plugin_path: str, emit_call
 
         self.load_type = PluginLoadType.LEGACY_EVAL_IIFE.value
 
-        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"))
+        plugin_dir_path = path.join(plugin_path, plugin_directory)
+        plugin_json_path = path.join(plugin_dir_path, "plugin.json")
+
+        json = load(open(plugin_json_path, "r", encoding="utf-8"))
+        if path.isfile(path.join(plugin_dir_path, "package.json")):
+            package_json = load(open(path.join(plugin_dir_path, "package.json"), "r", encoding="utf-8"))
             self.version = package_json["version"]
             if ("type" in package_json and package_json["type"] == "module"):
                 self.load_type = PluginLoadType.ESMODULE_V1.value
@@ -42,6 +46,17 @@ def __init__(self, file: str, plugin_directory: str, plugin_path: str, emit_call
 
         self.log = getLogger("plugin")
 
+        if get_chown_plugin_path():
+            # ensure plugin folder ownership
+            if file_owner(plugin_dir_path) != UserType.EFFECTIVE_USER:
+                chown(plugin_dir_path, UserType.EFFECTIVE_USER if "root" in self.flags else UserType.HOST_USER, True)
+                chown(plugin_dir_path, UserType.EFFECTIVE_USER, False)
+                chmod(plugin_dir_path, 755, True)
+            # fix plugin.json permissions
+            if file_owner(plugin_json_path) != UserType.EFFECTIVE_USER:
+                chown(plugin_json_path, UserType.EFFECTIVE_USER, False)
+                chmod(plugin_json_path, 755, False)
+
         self.sandboxed_plugin = SandboxedPlugin(self.name, self.passive, self.flags, self.file, self.plugin_directory, self.plugin_path, self.version, self.author, self.api_version)
         self.proc: Process | None = None
         self._socket = LocalSocket()

backend/decky_loader/plugin/sandboxed_plugin.py

@@ -13,7 +13,7 @@
 from ..localplatform.localsocket import LocalSocket
 from ..localplatform.localplatform import setgid, setuid, get_username, get_home_path, ON_LINUX
 from ..enums import UserType
-from .. import helpers, settings, injector # pyright: ignore [reportUnusedImport]
+from .. import helpers
 
 from typing import List, TypeVar, Any
 
@@ -61,10 +61,10 @@ def initialize(self, socket: LocalSocket):
             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)
+            setgid(UserType.EFFECTIVE_USER if "root" in self.flags else UserType.HOST_USER)
+            setuid(UserType.EFFECTIVE_USER 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["HOME"] = get_home_path(UserType.EFFECTIVE_USER 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()

backend/decky_loader/settings.py

@@ -1,7 +1,7 @@
 from json import dump, load
 from os import mkdir, path, listdir, rename
 from typing import Any, Dict
-from .localplatform.localplatform import chown, folder_owner, get_chown_plugin_path
+from .localplatform.localplatform import chown, file_owner, get_chown_plugin_path
 from .enums import UserType
 
 from .helpers import get_homebrew_path
@@ -28,8 +28,8 @@ def __init__(self, name: str, settings_directory: str | None = None) -> None:
 
 
         #If the owner of the settings directory is not the user, then set it as the user:
-        expected_user = UserType.HOST_USER if get_chown_plugin_path() else UserType.ROOT
-        if folder_owner(settings_directory) != expected_user:
+        expected_user = UserType.HOST_USER if get_chown_plugin_path() else UserType.EFFECTIVE_USER
+        if file_owner(settings_directory) != expected_user:
             chown(settings_directory, expected_user, False)
 
         self.settings: Dict[str, Any] = {}

backend/decky_loader/wsrouter.py

@@ -7,7 +7,7 @@
 
 from enum import IntEnum
 
-from typing import Callable, Coroutine, Dict, Any, cast, TypeVar
+from typing import Callable, Coroutine, Dict, Any, cast
 
 from traceback import format_exc
 
@@ -29,8 +29,6 @@ class WSMessageExtra(WSMessage):
 
 # see wsrouter.ts for typings
 
-DataType = TypeVar("DataType")
-
 Route = Callable[..., Coroutine[Any, Any, Any]]
 
 class WSRouter: