Feat/configurable paths (#404)

backend/helpers.py

@@ -36,9 +36,9 @@ async def csrf_middleware(request, handler):
         return await handler(request)
     return Response(text='Forbidden', status='403')
 
-# Get the default homebrew path unless a home_path is specified
+# Get the default homebrew path unless a home_path is specified. home_path argument is deprecated
 def get_homebrew_path(home_path = None) -> str:
-    return os.path.join(home_path if home_path != None else localplatform.get_home_path(), "homebrew")
+    return localplatform.get_unprivileged_path()
 
 # Recursively create path and chown as user
 def mkdir_as_user(path):

backend/localplatform.py

@@ -1,4 +1,4 @@
-import platform
+import platform, os
 
 ON_WINDOWS = platform.system() == "Windows"
 ON_LINUX = not ON_WINDOWS
@@ -8,4 +8,33 @@ if ON_WINDOWS:
     import localplatformwin as localplatform
 else:
     from localplatformlinux import *
-    import localplatformlinux as localplatform
+    import localplatformlinux as localplatform
+
+def get_privileged_path() -> str:
+    '''Get path accessible by elevated user. Holds plugins, decky loader and decky loader configs'''
+    return localplatform.get_privileged_path()
+
+def get_unprivileged_path() -> str:
+    '''Get path accessible by non-elevated user. Holds plugin configuration, plugin data and plugin logs. Externally referred to as the 'Homebrew' directory'''
+    return localplatform.get_unprivileged_path()
+
+def get_unprivileged_user() -> str:
+    '''Get user that should own files made in unprivileged path'''
+    return localplatform.get_unprivileged_user()
+
+def get_chown_plugin_path() -> bool:
+    return os.getenv("CHOWN_PLUGIN_PATH", "1") == "1"
+
+def get_server_host() -> str:
+    return os.getenv("SERVER_HOST", "127.0.0.1")
+
+def get_server_port() -> int:
+    return int(os.getenv("SERVER_PORT", "1337"))
+
+def get_live_reload() -> bool:
+    os.getenv("LIVE_RELOAD", "1") == "1"
+
+def get_log_level() -> int:
+    return {"CRITICAL": 50, "ERROR": 40, "WARNING": 30, "INFO": 20, "DEBUG": 10}[
+        os.getenv("LOG_LEVEL", "INFO")
+    ]

backend/localplatformlinux.py

@@ -1,19 +1,16 @@
-import os, pwd, grp, sys
+import os, pwd, grp, sys, logging
 from subprocess import call, run, DEVNULL, PIPE, STDOUT
 from customtypes import UserType
 
+logger = logging.getLogger("localplatform")
+
 # Get the user id hosting the plugin loader
 def _get_user_id() -> int:
-    proc_path = os.path.realpath(sys.argv[0])
-    pws = sorted(pwd.getpwall(), reverse=True, key=lambda pw: len(pw.pw_dir))
-    for pw in pws:
-        if proc_path.startswith(os.path.realpath(pw.pw_dir)):
-            return pw.pw_uid
-    raise PermissionError("The plugin loader does not seem to be hosted by any known user.")
+    return pwd.getpwnam(_get_user()).pw_uid
 
 # Get the user hosting the plugin loader
 def _get_user() -> str:
-    return pwd.getpwuid(_get_user_id()).pw_name
+    return get_unprivileged_user()
 
 # Get the effective user id of the running process
 def _get_effective_user_id() -> int:
@@ -50,10 +47,12 @@ def _get_user_group() -> str:
 def chown(path : str,  user : UserType = UserType.HOST_USER, recursive : bool = True) -> bool:
     user_str = ""
 
-    if (user == UserType.HOST_USER):
+    if user == UserType.HOST_USER:
         user_str = _get_user()+":"+_get_user_group()
-    elif (user == UserType.EFFECTIVE_USER):
+    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")
 
@@ -135,4 +134,61 @@ async def service_stop(service_name : str) -> bool:
 async def service_start(service_name : str) -> bool:
     cmd = ["systemctl", "start", service_name]
     res = run(cmd, stdout=PIPE, stderr=STDOUT)
-    return res.returncode == 0
+    return res.returncode == 0
+
+def get_privileged_path() -> str:
+    path = os.getenv("PRIVILEGED_PATH")
+
+    if path == None:
+        path = get_unprivileged_path()
+
+    return path
+
+def _parent_dir(path : str) -> str:
+    if path == None:
+        return None
+
+    if path.endswith('/'):
+        path = path[:-1]
+    
+    return os.path.dirname(path)
+
+def get_unprivileged_path() -> str:
+    path = os.getenv("UNPRIVILEGED_PATH")
+    
+    if path == None:
+        path = _parent_dir(os.getenv("PLUGIN_PATH"))
+    
+    if path == None:
+        logger.debug("Unprivileged path is not properly configured. Making something up!")
+        # 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):
+            path = None
+
+    if path == None:
+        logger.warn("Unprivileged path is not properly configured. Defaulting to /home/deck/homebrew")
+        path = "/home/deck/homebrew" # We give up
+    
+    return path
+
+
+def get_unprivileged_user() -> str:
+    user = os.getenv("UNPRIVILEGED_USER")
+
+    if user == None:
+        # Lets hope we can extract it from the unprivileged dir
+        dir = os.path.realpath(get_unprivileged_path())
+
+        pws = sorted(pwd.getpwall(), reverse=True, key=lambda pw: len(pw.pw_dir))
+        for pw in pws:
+            if dir.startswith(os.path.realpath(pw.pw_dir)):
+                user = pw.pw_name
+                break
+    
+    if user == None:
+        logger.warn("Unprivileged user is not properly configured. Defaulting to 'deck'")
+        user = 'deck'
+
+    return user

backend/localplatformwin.py

@@ -35,4 +35,19 @@ async def service_restart(service_name : str) -> bool:
     return True # Stubbed
 
 def get_username() -> str:
-    return os.getlogin()
+    return os.getlogin()
+
+def get_privileged_path() -> str:
+    '''On windows, privileged_path is equal to unprivileged_path'''
+    return get_unprivileged_path()
+
+def get_unprivileged_path() -> str:
+    path = os.getenv("UNPRIVILEGED_PATH")
+
+    if path == None:
+        path = os.getenv("PRIVILEGED_PATH", os.path.join(os.path.expanduser("~"), "homebrew"))
+
+    return path
+
+def get_unprivileged_user() -> str:
+    return os.getenv("UNPRIVILEGED_USER", os.getlogin())

backend/main.py

@@ -1,6 +1,10 @@
 # Change PyInstaller files permissions
 import sys
-from localplatform import chmod, chown, service_stop, service_start, ON_WINDOWS
+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
@@ -20,7 +24,7 @@ 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,
-                     get_homebrew_path, mkdir_as_user, get_system_pythonpaths)
+                     mkdir_as_user, get_system_pythonpaths)
                      
 from injector import get_gamepadui_tab, Tab, get_tabs, close_old_tabs
 from loader import Loader
@@ -29,33 +33,23 @@ from updater import Updater
 from utilities import Utilities
 from customtypes import UserType
 
-HOMEBREW_PATH = get_homebrew_path()
-CONFIG = {
-    "plugin_path": getenv("PLUGIN_PATH", path.join(HOMEBREW_PATH, "plugins")),
-    "chown_plugin_path": getenv("CHOWN_PLUGIN_PATH", "1") == "1",
-    "server_host": getenv("SERVER_HOST", "127.0.0.1"),
-    "server_port": int(getenv("SERVER_PORT", "1337")),
-    "live_reload": getenv("LIVE_RELOAD", "1") == "1",
-    "log_level": {"CRITICAL": 50, "ERROR": 40, "WARNING": 30, "INFO": 20, "DEBUG": 10}[
-        getenv("LOG_LEVEL", "INFO")
-    ],
-}
 
 basicConfig(
-    level=CONFIG["log_level"],
+    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(CONFIG["plugin_path"]): # For safety, create the folder before attempting to do anything with it
-        mkdir_as_user(CONFIG["plugin_path"])
+    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(CONFIG["plugin_path"], UserType.HOST_USER) or not chmod(CONFIG["plugin_path"], 555):
+    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 CONFIG["chown_plugin_path"] == True:
+if get_chown_plugin_path() == True:
     chown_plugin_dir()
 
 class PluginManager:
@@ -70,9 +64,9 @@ class PluginManager:
                 allow_credentials=True
             )
         })
-        self.plugin_loader = Loader(self.web_app, CONFIG["plugin_path"], self.loop, CONFIG["live_reload"])
-        self.settings = SettingsManager("loader", path.join(HOMEBREW_PATH, "settings"))
-        self.plugin_browser = PluginBrowser(CONFIG["plugin_path"], self.plugin_loader.plugins, self.plugin_loader, self.settings)
+        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)
 
@@ -174,7 +168,7 @@ class PluginManager:
             pass
 
     def run(self):
-        return run_app(self.web_app, host=CONFIG["server_host"], port=CONFIG["server_port"], loop=self.loop, access_log=None)
+        return run_app(self.web_app, host=get_server_host(), port=get_server_port(), loop=self.loop, access_log=None)
 
 if __name__ == "__main__":
     if ON_WINDOWS:

backend/settings.py

@@ -1,6 +1,6 @@
 from json import dump, load
 from os import mkdir, path, listdir, rename
-from localplatform import chown, folder_owner
+from localplatform import chown, folder_owner, get_chown_plugin_path
 from customtypes import UserType
 
 from helpers import get_homebrew_path
@@ -17,7 +17,6 @@ class SettingsManager:
         #Create the folder with the correct permission
         if not path.exists(settings_directory):
             mkdir(settings_directory)
-            chown(settings_directory)
 
         #Copy all old settings file in the root directory to the correct folder
         for file in listdir(wrong_dir):
@@ -28,9 +27,9 @@ class SettingsManager:
 
 
         #If the owner of the settings directory is not the user, then set it as the user:
-
-        if folder_owner(settings_directory) != UserType.HOST_USER:
-            chown(settings_directory, UserType.HOST_USER, False)
+        expected_user = UserType.HOST_USER if get_chown_plugin_path() else UserType.ROOT
+        if folder_owner(settings_directory) != expected_user:
+            chown(settings_directory, expected_user, False)
 
         self.settings = {}
 

dist/plugin_loader-prerelease.service

@@ -9,7 +9,8 @@ Restart=always
 ExecStart=${HOMEBREW_FOLDER}/services/PluginLoader
 WorkingDirectory=${HOMEBREW_FOLDER}/services
 KillSignal=SIGKILL
-Environment=PLUGIN_PATH=${HOMEBREW_FOLDER}/plugins
+Environment=UNPRIVILEGED_PATH=${HOMEBREW_FOLDER}
+Environment=PRIVILEGED_PATH=${HOMEBREW_FOLDER}
 Environment=LOG_LEVEL=DEBUG
 [Install]
 WantedBy=multi-user.target

dist/plugin_loader-release.service

@@ -9,7 +9,8 @@ Restart=always
 ExecStart=${HOMEBREW_FOLDER}/services/PluginLoader
 WorkingDirectory=${HOMEBREW_FOLDER}/services
 KillSignal=SIGKILL
-Environment=PLUGIN_PATH=${HOMEBREW_FOLDER}/plugins
+Environment=UNPRIVILEGED_PATH=${HOMEBREW_FOLDER}
+Environment=PRIVILEGED_PATH=${HOMEBREW_FOLDER}
 Environment=LOG_LEVEL=INFO
 [Install]
 WantedBy=multi-user.target