diff options
Diffstat (limited to 'backend/helpers.py')
| -rw-r--r-- | backend/helpers.py | 171 |
1 files changed, 83 insertions, 88 deletions
diff --git a/backend/helpers.py b/backend/helpers.py index af07b759..668a252d 100644 --- a/backend/helpers.py +++ b/backend/helpers.py @@ -1,19 +1,18 @@ -import grp -import pwd import re import ssl -import subprocess import uuid import os import sys -from subprocess import check_output -from time import sleep +import subprocess from hashlib import sha256 from io import BytesIO import certifi from aiohttp.web import Response, middleware from aiohttp import ClientSession +import localplatform +from customtypes import UserType +from logging import getLogger REMOTE_DEBUGGER_UNIT = "steam-web-debug-portforward.service" @@ -23,6 +22,7 @@ ssl_ctx = ssl.create_default_context(cafile=certifi.where()) assets_regex = re.compile("^/plugins/.*/assets/.*") frontend_regex = re.compile("^/frontend/.*") +logger = getLogger("Main") def get_ssl_context(): return ssl_ctx @@ -36,97 +36,41 @@ async def csrf_middleware(request, handler): return await handler(request) return Response(text='Forbidden', status='403') -# Deprecated -def set_user(): - 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: - return pwd.getpwuid(get_user_id()).pw_name - -# 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).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 - -# Deprecated -def set_user_group() -> str: - 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: - 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: - username = get_user() - return pwd.getpwnam(username).pw_dir - # Get the default homebrew path unless a home_path is specified def get_homebrew_path(home_path = None) -> str: - if home_path == None: - home_path = get_home_path() - return os.path.join(home_path, "homebrew") + return os.path.join(home_path if home_path != None else localplatform.get_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) + localplatform.chown(path) # Fetches the version of loader def get_loader_version() -> str: try: with open(os.path.join(os.getcwd(), ".loader.version"), "r", encoding="utf-8") as version_file: return version_file.readline().strip() - except: + except Exception as e: + logger.warn(f"Failed to execute get_loader_version(): {str(e)}") return "unknown" # returns the appropriate system python paths def get_system_pythonpaths() -> list[str]: - # run as normal normal user to also include user python paths - proc = subprocess.run(["python3", "-c", "import sys; print(':'.join(x for x in sys.path if x))"], - user=get_user_id(), env={}, capture_output=True) - return proc.stdout.decode().strip().split(":") + 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: + 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) + 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: @@ -152,16 +96,67 @@ async def download_remote_binary_to_path(url, binHash, path) -> bool: return rv -async def is_systemd_unit_active(unit_name: str) -> bool: - res = subprocess.run(["systemctl", "is-active", unit_name], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) - return res.returncode == 0 +# Deprecated +def set_user(): + pass + +# Deprecated +def set_user_group() -> str: + return get_user_group() + +######### +# Below is legacy code, provided for backwards compatibility. This will break on windows +######### + +# Get the user id hosting the plugin loader +def get_user_id() -> int: + return localplatform.localplatform._get_user_id() + +# Get the user hosting the plugin loader +def get_user() -> str: + return localplatform.localplatform._get_user() + +# Get the effective user id of the running process +def get_effective_user_id() -> int: + return localplatform.localplatform._get_effective_user_id() + +# Get the effective user of the running process +def get_effective_user() -> str: + return localplatform.localplatform._get_effective_user() + +# 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() + +# Get the effective user group of the running process +def get_effective_user_group() -> str: + return localplatform.localplatform._get_effective_user_group() + +# Get the user owner of the given file path. +def get_user_owner(file_path) -> str: + return localplatform.localplatform._get_user_owner(file_path) + +# Get the user group of the given file path. +def get_user_group(file_path) -> str: + return localplatform.localplatform._get_user_group(file_path) -async def stop_systemd_unit(unit_name: str) -> subprocess.CompletedProcess: - cmd = ["systemctl", "stop", unit_name] +# Get the group id of the user hosting the plugin loader +def get_user_group_id() -> int: + return localplatform.localplatform._get_user_group_id() - return subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) +# Get the group of the user hosting the plugin loader +def get_user_group() -> str: + return localplatform.localplatform._get_user_group() + +# Get the default home path unless a user is specified +def get_home_path(username = 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: + return await localplatform.service_active(unit_name) -async def start_systemd_unit(unit_name: str) -> subprocess.CompletedProcess: - cmd = ["systemctl", "start", unit_name] +async def stop_systemd_unit(unit_name: str) -> bool: + return await localplatform.service_stop(unit_name) - return subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) +async def start_systemd_unit(unit_name: str) -> bool: + return await localplatform.service_start(unit_name)
\ No newline at end of file |