diff options
| -rw-r--r-- | backend/helpers.py | 111 | ||||
| -rw-r--r-- | backend/main.py | 14 | ||||
| -rw-r--r-- | backend/plugin.py | 25 | ||||
| -rw-r--r-- | backend/settings.py | 10 | ||||
| -rw-r--r-- | backend/updater.py | 6 |
5 files changed, 98 insertions, 68 deletions
diff --git a/backend/helpers.py b/backend/helpers.py index b97352bd..7cab512b 100644 --- a/backend/helpers.py +++ b/backend/helpers.py @@ -5,6 +5,7 @@ import ssl import subprocess import uuid import os +import sys from subprocess import check_output from time import sleep from hashlib import sha256 @@ -19,8 +20,6 @@ REMOTE_DEBUGGER_UNIT = "steam-web-debug-portforward.service" # global vars csrf_token = str(uuid.uuid4()) ssl_ctx = ssl.create_default_context(cafile=certifi.where()) -user = None -group = None assets_regex = re.compile("^/plugins/.*/assets/.*") frontend_regex = re.compile("^/frontend/.*") @@ -37,65 +36,87 @@ async def csrf_middleware(request, handler): return await handler(request) return Response(text='Forbidden', status='403') -# Get the user by checking for the first logged in user. As this is run -# by systemd at startup the process is likely to start before the user -# logs in, so we will wait here until they are available. Note that -# other methods such as getenv wont work as there was no $SUDO_USER to -# start the systemd service. +# Deprecated def set_user(): - global user - cmd = "who | awk '{print $1}' | sort | head -1" - while user == None: - name = check_output(cmd, shell=True).decode().strip() - if name not in [None, '']: - user = name - sleep(0.1) - -# Get the global user. get_user must be called first. + pass + +# 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.") + +# Get the user hosting the plugin loader def get_user() -> str: - global user - if user == None: - raise ValueError("helpers.get_user method called before user variable was set. Run helpers.set_user first.") - return user + return pwd.getpwuid(get_user_id()).pw_name -#Get the user owner of the given file path. +# Get the effective user id of the running process +def get_effective_user_id() -> int: + return os.geteuid() + +# Get the effective user of the running process +def get_effective_user() -> str: + return pwd.getpwuid(get_effective_user_id()).pw_name + +# Get the effective user group id of the running process +def get_effective_user_group_id() -> int: + return os.getegid() + +# Get the effective user group of the running process +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: - return pwd.getpwuid(os.stat(file_path).st_uid)[0] + return pwd.getpwuid(os.stat(file_path).st_uid).pw_name -#Get the user group of the given file path. +# 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)[0] + return grp.getgrgid(os.stat(file_path).st_gid).gr_name -# Set the global user group. get_user must be called first +# Deprecated def set_user_group() -> str: - global group - global user - if user == None: - raise ValueError("helpers.set_user_dir method called before user variable was set. Run helpers.set_user first.") - if group == None: - group = check_output(["id", "-g", "-n", user]).decode().strip() - -# Get the group of the global user. set_user_group must be called first. + return get_user_group() + +# 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: - global group - if group == None: - raise ValueError("helpers.get_user_group method called before group variable was set. Run helpers.set_user_group first.") - return group + return grp.getgrgid(get_user_group_id()).gr_name # Get the default home path unless a user is specified def get_home_path(username = None) -> str: if username == None: - raise ValueError("Username not defined, no home path can be found.") - else: - return str("/home/"+username) + username = get_user() + return pwd.getpwnam(username).pw_dir -# Get the default homebrew path unless a user is specified +# Get the default homebrew path unless a home_path is specified def get_homebrew_path(home_path = None) -> str: if home_path == None: - raise ValueError("Home path not defined, homebrew dir cannot be determined.") - else: - return str(home_path+"/homebrew") - # return str(home_path+"/homebrew") + home_path = get_home_path() + return os.path.join(home_path, "homebrew") + +# Recursively create path and chown as user +def mkdir_as_user(path): + path = os.path.realpath(path) + os.makedirs(path, exist_ok=True) + chown_path = get_home_path() + parts = os.path.relpath(path, chown_path).split(os.sep) + uid = get_user_id() + gid = get_user_group_id() + for p in parts: + chown_path = os.path.join(chown_path, p) + os.chown(chown_path, uid, gid) + +# Fetches the version of loader +def get_loader_version() -> str: + with open(os.path.join(os.path.dirname(sys.argv[0]), ".loader.version"), "r", encoding="utf-8") as version_file: + return version_file.readline().replace("\n", "") # Download Remote Binaries to local Plugin async def download_remote_binary_to_path(url, binHash, path) -> bool: diff --git a/backend/main.py b/backend/main.py index c48ad752..a2ac008a 100644 --- a/backend/main.py +++ b/backend/main.py @@ -19,8 +19,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_home_path, get_homebrew_path, get_user, - get_user_group, set_user, set_user_group, + get_home_path, get_homebrew_path, get_user, get_user_group, stop_systemd_unit, start_systemd_unit) from injector import get_gamepadui_tab, Tab, get_tabs, close_old_tabs from loader import Loader @@ -28,18 +27,11 @@ from settings import SettingsManager from updater import Updater from utilities import Utilities -# Ensure USER and GROUP vars are set first. -# TODO: This isn't the best way to do this but supports the current -# implementation. All the config load and environment setting eventually be -# moved into init or a config/loader method. -set_user() -set_user_group() USER = get_user() GROUP = get_user_group() -HOME_PATH = "/home/"+USER -HOMEBREW_PATH = HOME_PATH+"/homebrew" +HOMEBREW_PATH = get_homebrew_path() CONFIG = { - "plugin_path": getenv("PLUGIN_PATH", HOMEBREW_PATH+"/plugins"), + "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")), diff --git a/backend/plugin.py b/backend/plugin.py index e21d5bde..df0efe16 100644 --- a/backend/plugin.py +++ b/backend/plugin.py @@ -7,10 +7,12 @@ from importlib.util import module_from_spec, spec_from_file_location from json import dumps, load, loads from logging import getLogger from traceback import format_exc -from os import path, setgid, setuid +from os import path, setgid, setuid, environ from signal import SIGINT, signal from sys import exit from time import time +import helpers +from updater import Updater multiprocessing.set_start_method("fork") @@ -19,6 +21,7 @@ BUFFER_LIMIT = 2 ** 20 # 1 MiB class PluginWrapper: def __init__(self, file, plugin_directory, plugin_path) -> None: self.file = file + self.plugin_path = plugin_path self.plugin_directory = plugin_directory self.reader = None self.writer = None @@ -56,8 +59,24 @@ class PluginWrapper: set_event_loop(new_event_loop()) if self.passive: return - setgid(0 if "root" in self.flags else 1000) - setuid(0 if "root" in self.flags else 1000) + setgid(0 if "root" in self.flags else helpers.get_user_group_id()) + setuid(0 if "root" in self.flags else helpers.get_user_id()) + # export a bunch of environment variables to help plugin developers + environ["HOME"] = helpers.get_home_path("root" if "root" in self.flags else helpers.get_user()) + environ["USER"] = "root" if "root" in self.flags else helpers.get_user() + environ["DECKY_VERSION"] = helpers.get_loader_version() + environ["DECKY_USER"] = helpers.get_user() + environ["DECKY_HOME"] = helpers.get_homebrew_path() + environ["DECKY_PLUGIN_SETTINGS_DIR"] = path.join(environ["DECKY_HOME"], "settings", self.plugin_directory) + helpers.mkdir_as_user(environ["DECKY_PLUGIN_SETTINGS_DIR"]) + environ["DECKY_PLUGIN_RUNTIME_DIR"] = path.join(environ["DECKY_HOME"], "data", self.plugin_directory) + helpers.mkdir_as_user(environ["DECKY_PLUGIN_RUNTIME_DIR"]) + environ["DECKY_PLUGIN_LOG_DIR"] = path.join(environ["DECKY_HOME"], "logs", self.plugin_directory) + helpers.mkdir_as_user(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 + environ["DECKY_PLUGIN_AUTHOR"] = self.author spec = spec_from_file_location("_", self.file) module = module_from_spec(spec) spec.loader.exec_module(module) diff --git a/backend/settings.py b/backend/settings.py index 6dedcbbe..64b04c60 100644 --- a/backend/settings.py +++ b/backend/settings.py @@ -2,14 +2,14 @@ from json import dump, load from os import mkdir, path, listdir, rename from shutil import chown -from helpers import get_home_path, get_homebrew_path, get_user, set_user, get_user_owner +from helpers import get_home_path, get_homebrew_path, get_user, get_user_group, get_user_owner class SettingsManager: def __init__(self, name, settings_directory = None) -> None: - set_user() USER = get_user() - wrong_dir = get_homebrew_path(get_home_path(USER)) + GROUP = get_user_group() + wrong_dir = get_homebrew_path() if settings_directory == None: settings_directory = path.join(wrong_dir, "settings") @@ -18,7 +18,7 @@ class SettingsManager: #Create the folder with the correct permission if not path.exists(settings_directory): mkdir(settings_directory) - chown(settings_directory, USER, USER) + chown(settings_directory, USER, GROUP) #Copy all old settings file in the root directory to the correct folder for file in listdir(wrong_dir): @@ -30,7 +30,7 @@ class SettingsManager: #If the owner of the settings directory is not the user, then set it as the user: if get_user_owner(settings_directory) != USER: - chown(settings_directory, USER, USER) + chown(settings_directory, USER, GROUP) self.settings = {} diff --git a/backend/updater.py b/backend/updater.py index 15a93e8a..14fd2070 100644 --- a/backend/updater.py +++ b/backend/updater.py @@ -31,9 +31,7 @@ class Updater: self.remoteVer = None self.allRemoteVers = None try: - logger.info(getcwd()) - with open(path.join(getcwd(), ".loader.version"), "r", encoding="utf-8") as version_file: - self.localVer = version_file.readline().replace("\n", "") + self.localVer = helpers.get_loader_version() except: self.localVer = False @@ -161,7 +159,7 @@ class Updater: logger.error(f"Error at %s", exc_info=e) with open(path.join(getcwd(), "plugin_loader.service"), "r", encoding="utf-8") as service_file: service_data = service_file.read() - service_data = service_data.replace("${HOMEBREW_FOLDER}", "/home/"+helpers.get_user()+"/homebrew") + service_data = service_data.replace("${HOMEBREW_FOLDER}", helpers.get_homebrew_path()) with open(path.join(getcwd(), "plugin_loader.service"), "w", encoding="utf-8") as service_file: service_file.write(service_data) |