diff options
Diffstat (limited to 'py_modules/lsfg_vk/wrapper_service.py')
| -rw-r--r-- | py_modules/lsfg_vk/wrapper_service.py | 210 |
1 files changed, 45 insertions, 165 deletions
diff --git a/py_modules/lsfg_vk/wrapper_service.py b/py_modules/lsfg_vk/wrapper_service.py index 980f7ae..807a1c3 100644 --- a/py_modules/lsfg_vk/wrapper_service.py +++ b/py_modules/lsfg_vk/wrapper_service.py @@ -6,7 +6,6 @@ import json import re import shlex import threading -from pathlib import Path from typing import Any, Dict, Optional, Tuple from .base_service import BaseService @@ -43,6 +42,7 @@ class WrapperService(BaseService): "__GLX_VENDOR_LIBRARY_NAME", "GALLIUM_DRIVER", "DXVK_FRAME_RATE", + "LSFGVK_FLATPAK", ) def __init__(self, logger=None): @@ -92,44 +92,28 @@ class WrapperService(BaseService): if not isinstance(raw, dict): raise ValueError("Workaround transport must be an object") kind = raw.get("kind") - if kind == "host": - return {"kind": "host"} - if kind == "flatpak": - app_id = raw.get("flatpakAppId") - if ( - not isinstance(app_id, str) - or not re.fullmatch( - r"^[A-Za-z0-9][A-Za-z0-9-]*(?:\.[A-Za-z0-9][A-Za-z0-9-]*)+$", - app_id, - ) - ): - raise ValueError("Flatpak transport requires a valid application ID") - return {"kind": "flatpak", "flatpakAppId": app_id} + if kind in ("host", "flatpak"): + # Flatpak app IDs are deliberately not part of plugin state. The + # only special target is the direct /usr/bin/flatpak shortcut. + return {"kind": kind} raise ValueError("Workaround transport must be host or flatpak") @classmethod def _validate_entry(cls, raw: Any) -> Dict[str, Any]: if not isinstance(raw, dict): raise ValueError("Workaround AppID entry must be an object") + transport = cls._validate_transport(raw.get("transport")) entry = { "state": cls._validate_state(raw.get("state")), "command_token_added": raw.get("command_token_added", False), - # Version 1 entries had no transport field. They are preserved as - # host entries until the shortcut is explicitly repaired with the - # backend's classified transport. - "transport": cls._validate_transport(raw.get("transport")), } if type(entry["command_token_added"]) is not bool: raise ValueError("command_token_added must be a boolean") - if entry["transport"]["kind"] == "flatpak" and "shortcut_exe" in raw and raw["shortcut_exe"] is not None: - shortcut_exe = raw["shortcut_exe"] - if ( - not isinstance(shortcut_exe, str) - or not shortcut_exe.startswith("/") - or "\x00" in shortcut_exe - or not shortcut_exe.strip() - ): - raise ValueError("shortcut_exe must be an absolute executable path") + if transport["kind"] == "flatpak": + shortcut_exe = raw.get("shortcut_exe") + if shortcut_exe != "/usr/bin/flatpak": + raise ValueError("Flatpak workaround state must save /usr/bin/flatpak") + entry["transport"] = transport entry["shortcut_exe"] = shortcut_exe return entry @@ -160,11 +144,11 @@ class WrapperService(BaseService): if self.sidecar_path.is_symlink() or not self.sidecar_path.is_file(): raise RuntimeError("Workaround state path is not a regular file") try: - raw = json.loads(self.sidecar_path.read_text(encoding="utf-8")) + content = self.sidecar_path.read_text(encoding="utf-8") + raw = json.loads(content) except (OSError, json.JSONDecodeError) as error: raise RuntimeError(f"Could not read workaround state: {error}") from error - document = self._validate_document(raw) - return document, True, self.sidecar_path.read_text(encoding="utf-8") + return self._validate_document(raw), True, content def _wrapper_marker(self) -> bool: if self.wrapper_path.is_symlink() or not self.wrapper_path.exists(): @@ -190,20 +174,19 @@ class WrapperService(BaseService): def _shell(value: str) -> str: return shlex.quote(value) - @staticmethod - def _direct_flatpak_tokens(value: str) -> Optional[list[str]]: - """Parse the supported full executable form: /usr/bin/flatpak run APP.""" - try: - tokens = shlex.split(value, posix=True) - except ValueError: - return None - if len(tokens) >= 3 and Path(tokens[0]).name == "flatpak" and tokens[1] == "run": - return tokens - return None - - @classmethod - def _state_lines(cls, state: Dict[str, Any], shortcut_exe: Optional[str]) -> list[str]: - lines = [" unset " + " ".join(cls.MANAGED_ENV_KEYS)] + def _state_lines(self, state: Dict[str, Any], shortcut_exe: Optional[str]) -> list[str]: + lines = [ + " unset " + " ".join(self.MANAGED_ENV_KEYS), + ' SteamAppId="$appid"', + " export SteamAppId", + f" LSFGVK_CONFIG={self._shell(str(self.config_file_path))}", + " export LSFGVK_CONFIG", + ] + if shortcut_exe: + lines.extend([ + " LSFGVK_FLATPAK=1", + " export LSFGVK_FLATPAK", + ]) if state["disableGamescopeWsi"]: lines.extend([" ENABLE_GAMESCOPE_WSI=0", " export ENABLE_GAMESCOPE_WSI"]) if state["disableHdr"]: @@ -235,68 +218,9 @@ class WrapperService(BaseService): " fi", " export DXVK_CONFIG", ]) - lines.append(f" shortcut_exe={cls._shell(shortcut_exe or '')}") + lines.append(f" shortcut_exe={self._shell(shortcut_exe or '')}") return lines - def _dll_directory(self) -> Path: - if self.config_file_path.exists(): - try: - content = self.config_file_path.read_text(encoding="utf-8") - match = re.search( - r'(?m)^[ \t]*dll[ \t]*=[ \t]*"((?:\\.|[^"\\])*)"', - content, - ) - if match: - configured_dll = json.loads('"' + match.group(1) + '"') - if configured_dll: - return Path(configured_dll).parent - except Exception: - pass - return self.user_home / ".local/share/Steam/steamapps/common/Lossless Scaling" - - def _flatpak_args(self, state: Dict[str, Any]) -> list[str]: - config_dir = str(self.config_dir) - config_file = str(self.config_file_path) - dll_dir = str(self._dll_directory()) - args = [ - self._shell(f"--filesystem={config_dir}:rw"), - self._shell(f"--filesystem={dll_dir}:ro"), - self._shell(f"--env=LSFGVK_CONFIG={config_file}"), - '"--env=LSFGVK_FLATPAK=1"', - '"--env=SteamAppId=$appid"', - '"--unset-env=DISABLE_LSFGVK" "--unset-env=DISABLE_LSFG"', - '"--unset-env=DISABLE_GAMESCOPE_WSI"', - '"--unset-env=ENABLE_GAMESCOPE_WSI"' if not state["disableGamescopeWsi"] else - '"--env=ENABLE_GAMESCOPE_WSI=0"', - '"--unset-env=DXVK_HDR"' if not state["disableHdr"] else - '"--env=DXVK_HDR=0"', - '"--unset-env=SteamDeck"' if not state["disableSteamdeckMode"] else - '"--env=SteamDeck=0"', - '"--unset-env=DISABLE_VKBASALT" "--unset-env=ENABLE_VKBASALT"', - ] - if state["disableVkbasalt"]: - args.append('"--env=DISABLE_VKBASALT=1"') - args.extend([ - '"--unset-env=MESA_LOADER_DRIVER_OVERRIDE" "--unset-env=__GLX_VENDOR_LIBRARY_NAME" "--unset-env=GALLIUM_DRIVER"', - ]) - if state["enableZink"]: - args.extend([ - '"--env=__GLX_VENDOR_LIBRARY_NAME=mesa"', - '"--env=MESA_LOADER_DRIVER_OVERRIDE=zink"', - '"--env=GALLIUM_DRIVER=zink"', - ]) - args.extend([ - '"--unset-env=DXVK_FRAME_RATE"', - ]) - static_args = " ".join(args) - return [ - ' if [ -n "${DXVK_CONFIG+x}" ]; then', - f' set -- "$flatpak_command" {static_args} "--env=DXVK_CONFIG=$DXVK_CONFIG" "$@"', - " else", - f' set -- "$flatpak_command" {static_args} "$@"', - " fi", - ] - def _render_wrapper(self, document: Dict[str, Any]) -> str: lines = [ "#!/bin/sh", @@ -332,55 +256,9 @@ class WrapperService(BaseService): "esac", "", 'if [ -n "$shortcut_exe" ]; then', - ]) - # The arguments are emitted per branch below so the values are static and - # the wrapper never needs a JSON parser or another helper executable. - lines.append(' case "$appid" in') - for appid in sorted(document["apps"], key=lambda value: int(value)): - entry = document["apps"][appid] - transport = entry.get("transport", {"kind": "host"}) - if transport.get("kind") != "flatpak": - continue - shortcut_exe = entry.get("shortcut_exe", "") - direct_flatpak_tokens = self._direct_flatpak_tokens(shortcut_exe) - if direct_flatpak_tokens is None and Path(shortcut_exe).name != "flatpak": - raise ValueError( - f"Flatpak target {appid} does not use a direct flatpak executable" - ) - lines.append(f" {appid})") - lines.extend([ - *( - [ - f" shortcut_exe={self._shell(direct_flatpak_tokens[0])}", - " set -- " - + " ".join(self._shell(token) for token in direct_flatpak_tokens[1:]) - + ' "$@"', - ] - if direct_flatpak_tokens - else [] - ), - ' if [ "${1-}" != "run" ]; then', - ' echo "lsfg-vk: Flatpak shortcut must use direct flatpak run transport" >&2', - " exit 64", - " fi", - ' flatpak_command="$1"', - " shift", - " flatpak_target=", - ' for flatpak_arg in "$@"; do', - ' case "$flatpak_arg" in', - ' -*) ;;', - ' *) flatpak_target="$flatpak_arg"; break ;;', - " esac", - " done", - f' if [ "$flatpak_target" != {self._shell(transport["flatpakAppId"])} ]; then', - ' echo "lsfg-vk: Flatpak shortcut application ID changed externally" >&2', - " exit 64", - " fi", - ]) - lines.extend(self._flatpak_args(entry["state"])) - lines.append(" ;;") - lines.extend([ - " esac", + ' if [ "${1-}" = "$shortcut_exe" ]; then', + " shift", + " fi", ' exec "$shortcut_exe" "$@"', "fi", 'exec "$@"', @@ -396,7 +274,11 @@ class WrapperService(BaseService): old_sidecar_exists = self.sidecar_path.exists() old_sidecar = self.sidecar_path.read_text(encoding="utf-8") if old_sidecar_exists else None old_wrapper_exists = self.wrapper_path.exists() or self.wrapper_path.is_symlink() - old_wrapper = self.wrapper_path.read_text(encoding="utf-8") if old_wrapper_exists and not self.wrapper_path.is_symlink() else None + old_wrapper = ( + self.wrapper_path.read_text(encoding="utf-8") + if old_wrapper_exists and not self.wrapper_path.is_symlink() + else None + ) try: self._write_document(document) self._write_file(self.wrapper_path, self._render_wrapper(document), 0o755) @@ -465,24 +347,22 @@ class WrapperService(BaseService): document, _, _ = self._read_document() previous_entry = document["apps"].get(normalized) selected_transport = self._validate_transport( - transport - if transport is not None - else ( - previous_entry.get("transport") - if previous_entry - else None + transport if transport is not None else ( + previous_entry.get("transport") if previous_entry else None ) ) entry: Dict[str, Any] = { "state": validated_state, "command_token_added": bool(command_token_added), - "transport": selected_transport, } if selected_transport["kind"] == "flatpak": - if shortcut_exe is not None: - entry = self._validate_entry({**entry, "shortcut_exe": shortcut_exe}) - elif previous_entry and "shortcut_exe" in previous_entry: - entry["shortcut_exe"] = previous_entry["shortcut_exe"] + selected_shortcut = shortcut_exe or ( + previous_entry.get("shortcut_exe") if previous_entry else None + ) + if selected_shortcut != "/usr/bin/flatpak": + raise ValueError("Flatpak workaround state must save /usr/bin/flatpak") + entry["transport"] = selected_transport + entry["shortcut_exe"] = selected_shortcut document["apps"][normalized] = entry self._write_pair(document) return self._response(document, normalized) |