summaryrefslogtreecommitdiff
path: root/backend/src/localplatform
diff options
context:
space:
mode:
Diffstat (limited to 'backend/src/localplatform')
-rw-r--r--backend/src/localplatform/localplatform.py52
-rw-r--r--backend/src/localplatform/localplatformlinux.py192
-rw-r--r--backend/src/localplatform/localplatformwin.py53
-rw-r--r--backend/src/localplatform/localsocket.py139
4 files changed, 436 insertions, 0 deletions
diff --git a/backend/src/localplatform/localplatform.py b/backend/src/localplatform/localplatform.py
new file mode 100644
index 00000000..028eff8f
--- /dev/null
+++ b/backend/src/localplatform/localplatform.py
@@ -0,0 +1,52 @@
+import platform, os
+
+ON_WINDOWS = platform.system() == "Windows"
+ON_LINUX = not ON_WINDOWS
+
+if ON_WINDOWS:
+ from .localplatformwin import *
+ from . import localplatformwin as localplatform
+else:
+ 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'''
+ 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:
+ return os.getenv("LIVE_RELOAD", "1") == "1"
+
+def get_keep_systemd_service() -> bool:
+ return os.getenv("KEEP_SYSTEMD_SERVICE", "0") == "1"
+
+def get_log_level() -> int:
+ return {"CRITICAL": 50, "ERROR": 40, "WARNING": 30, "INFO": 20, "DEBUG": 10}[
+ os.getenv("LOG_LEVEL", "INFO")
+ ]
+
+def get_selinux() -> bool:
+ if ON_LINUX:
+ from subprocess import check_output
+ try:
+ if (check_output("getenforce").decode("ascii").strip("\n") == "Enforcing"): return True
+ except FileNotFoundError:
+ pass
+ return False
diff --git a/backend/src/localplatform/localplatformlinux.py b/backend/src/localplatform/localplatformlinux.py
new file mode 100644
index 00000000..1ec3fc1a
--- /dev/null
+++ b/backend/src/localplatform/localplatformlinux.py
@@ -0,0 +1,192 @@
+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:
+ return pwd.getpwnam(_get_user()).pw_uid
+
+# Get the user hosting the plugin loader
+def _get_user() -> str:
+ return get_unprivileged_user()
+
+# 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) -> str:
+ return pwd.getpwuid(os.stat(file_path).st_uid).pw_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
+
+def chown(path : str, user : UserType = UserType.HOST_USER, recursive : bool = True) -> bool:
+ user_str = ""
+
+ if user == UserType.HOST_USER:
+ 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")
+
+ result = call(["chown", "-R", user_str, path] if recursive else ["chown", user_str, path])
+ return result == 0
+
+def chmod(path : str, permissions : int, recursive : bool = True) -> bool:
+ if _get_effective_user_id() != 0:
+ return True
+ result = call(["chmod", "-R", str(permissions), path] if recursive else ["chmod", str(permissions), path])
+ return result == 0
+
+def folder_owner(path : str) -> UserType|None:
+ user_owner = _get_user_owner(path)
+
+ if (user_owner == _get_user()):
+ return UserType.HOST_USER
+
+ elif (user_owner == _get_effective_user()):
+ return UserType.EFFECTIVE_USER
+
+ else:
+ return None
+
+def get_home_path(user : UserType = UserType.HOST_USER) -> str:
+ user_name = "root"
+
+ if user == UserType.HOST_USER:
+ 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_username() -> str:
+ return _get_user()
+
+def setgid(user : UserType = UserType.HOST_USER):
+ user_id = 0
+
+ if user == UserType.HOST_USER:
+ user_id = _get_user_group_id()
+ elif user == UserType.ROOT:
+ pass
+ else:
+ raise Exception("Unknown user type")
+
+ os.setgid(user_id)
+
+def setuid(user : UserType = UserType.HOST_USER):
+ user_id = 0
+
+ if user == UserType.HOST_USER:
+ user_id = _get_user_id()
+ elif user == UserType.ROOT:
+ pass
+ else:
+ raise Exception("Unknown user type")
+
+ os.setuid(user_id)
+
+async def service_active(service_name : str) -> bool:
+ res = run(["systemctl", "is-active", service_name], stdout=DEVNULL, stderr=DEVNULL)
+ return res.returncode == 0
+
+async def service_restart(service_name : str) -> bool:
+ call(["systemctl", "daemon-reload"])
+ cmd = ["systemctl", "restart", service_name]
+ res = run(cmd, stdout=PIPE, stderr=STDOUT)
+ return res.returncode == 0
+
+async def service_stop(service_name : str) -> bool:
+ cmd = ["systemctl", "stop", service_name]
+ res = run(cmd, stdout=PIPE, stderr=STDOUT)
+ return res.returncode == 0
+
+async def service_start(service_name : str) -> bool:
+ cmd = ["systemctl", "start", service_name]
+ res = run(cmd, stdout=PIPE, stderr=STDOUT)
+ 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 | None) -> str | None:
+ 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 path != None and 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
diff --git a/backend/src/localplatform/localplatformwin.py b/backend/src/localplatform/localplatformwin.py
new file mode 100644
index 00000000..4c4e9439
--- /dev/null
+++ b/backend/src/localplatform/localplatformwin.py
@@ -0,0 +1,53 @@
+from .customtypes import UserType
+import os, sys
+
+def chown(path : str, user : UserType = UserType.HOST_USER, recursive : bool = True) -> bool:
+ return True # Stubbed
+
+def chmod(path : str, permissions : int, recursive : bool = True) -> bool:
+ return True # Stubbed
+
+def folder_owner(path : str) -> UserType|None:
+ return UserType.HOST_USER # Stubbed
+
+def get_home_path(user : UserType = UserType.HOST_USER) -> str:
+ return os.path.expanduser("~") # Mostly stubbed
+
+def setgid(user : UserType = UserType.HOST_USER):
+ pass # Stubbed
+
+def setuid(user : UserType = UserType.HOST_USER):
+ pass # Stubbed
+
+async def service_active(service_name : str) -> bool:
+ return True # Stubbed
+
+async def service_stop(service_name : str) -> bool:
+ return True # Stubbed
+
+async def service_start(service_name : str) -> bool:
+ return True # Stubbed
+
+async def service_restart(service_name : str) -> bool:
+ if service_name == "plugin_loader":
+ sys.exit(42)
+
+ return True # Stubbed
+
+def get_username() -> str:
+ 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())
diff --git a/backend/src/localplatform/localsocket.py b/backend/src/localplatform/localsocket.py
new file mode 100644
index 00000000..f38fe5e7
--- /dev/null
+++ b/backend/src/localplatform/localsocket.py
@@ -0,0 +1,139 @@
+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: 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.
+ Method should be async
+ '''
+ self.socket_addr = f"/tmp/plugin_socket_{time.time()}"
+ self.on_new_message = on_new_message
+ self.socket = None
+ self.reader = None
+ self.writer = None
+
+ async def setup_server(self):
+ self.socket = await asyncio.start_unix_server(self._listen_for_method_call, path=self.socket_addr, limit=BUFFER_LIMIT)
+
+ async def _open_socket_if_not_exists(self):
+ if not self.reader:
+ retries = 0
+ while retries < 10:
+ try:
+ self.reader, self.writer = await asyncio.open_unix_connection(self.socket_addr, limit=BUFFER_LIMIT)
+ return True
+ except:
+ await asyncio.sleep(2)
+ retries += 1
+ return False
+ else:
+ return True
+
+ async def get_socket_connection(self):
+ if not await self._open_socket_if_not_exists():
+ return None, None
+
+ return self.reader, self.writer
+
+ async def close_socket_connection(self):
+ if self.writer != None:
+ self.writer.close()
+
+ self.reader = None
+
+ async def read_single_line(self) -> str|None:
+ reader, _ = await self.get_socket_connection()
+
+ try:
+ assert reader
+ except AssertionError:
+ return
+
+ return await self._read_single_line(reader)
+
+ async def write_single_line(self, message : str):
+ _, writer = await self.get_socket_connection()
+
+ try:
+ assert writer
+ except AssertionError:
+ return
+
+ await self._write_single_line(writer, message)
+
+ 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)) # type: ignore
+ continue
+ except asyncio.IncompleteReadError as err:
+ line.extend(err.partial)
+ break
+ else:
+ break
+
+ return line.decode("utf-8")
+
+ 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: asyncio.StreamReader, writer: asyncio.StreamWriter):
+ while True:
+ line = await self._read_single_line(reader)
+
+ try:
+ res = await self.on_new_message(line)
+ except Exception:
+ return
+
+ if res != None:
+ await self._write_single_line(writer, res)
+
+class PortSocket (UnixSocket):
+ 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.
+ Method should be async
+ '''
+ super().__init__(on_new_message)
+ self.host = "127.0.0.1"
+ self.port = random.sample(range(40000, 60000), 1)[0]
+
+ async def setup_server(self):
+ self.socket = await asyncio.start_server(self._listen_for_method_call, host=self.host, port=self.port, limit=BUFFER_LIMIT)
+
+ async def _open_socket_if_not_exists(self):
+ if not self.reader:
+ retries = 0
+ while retries < 10:
+ try:
+ self.reader, self.writer = await asyncio.open_connection(host=self.host, port=self.port, limit=BUFFER_LIMIT)
+ return True
+ except:
+ await asyncio.sleep(2)
+ retries += 1
+ return False
+ else:
+ return True
+
+if ON_WINDOWS:
+ class LocalSocket (PortSocket): # type: ignore
+ pass
+else:
+ class LocalSocket (UnixSocket):
+ pass \ No newline at end of file
generated by cgit v1.2.3 (git 2.25.1) at 2025-12-16 13:08:43 +0000