From 485ebb748185391e28764f48c39322b3cbdac65c Mon Sep 17 00:00:00 2001 From: xXJSONDeruloXx Date: Thu, 10 Sep 2026 16:26:17 -0400 Subject: refactor: remove Flatpak support --- py_modules/lsfg_vk/constants.py | 3 - py_modules/lsfg_vk/flatpak_service.py | 431 ---------------------------------- py_modules/lsfg_vk/plugin.py | 49 +--- py_modules/lsfg_vk/steam_service.py | 47 ---- py_modules/lsfg_vk/wrapper_service.py | 193 +-------------- 5 files changed, 8 insertions(+), 715 deletions(-) delete mode 100644 py_modules/lsfg_vk/flatpak_service.py (limited to 'py_modules/lsfg_vk') diff --git a/py_modules/lsfg_vk/constants.py b/py_modules/lsfg_vk/constants.py index 960d230..45ac4c5 100644 --- a/py_modules/lsfg_vk/constants.py +++ b/py_modules/lsfg_vk/constants.py @@ -16,9 +16,6 @@ CLI_FILENAME = "lsfg-vk-cli" UI_FILENAME = "lsfg-vk-ui" UI_DESKTOP_FILENAME = "gay.pancake.lsfg-vk-ui.desktop" UI_ICON_FILENAME = "gay.pancake.lsfg-vk-ui.png" -FLATPAK_23_08_FILENAME = "org.freedesktop.Platform.VulkanLayer.lsfg_vk_23.08.flatpak" -FLATPAK_24_08_FILENAME = "org.freedesktop.Platform.VulkanLayer.lsfg_vk_24.08.flatpak" -FLATPAK_25_08_FILENAME = "org.freedesktop.Platform.VulkanLayer.lsfg_vk_25.08.flatpak" STEAM_LOSSLESS_SCALING_APP_ID = "993090" STEAM_LOSSLESS_SCALING_BRANCH = "lsfg-vk" diff --git a/py_modules/lsfg_vk/flatpak_service.py b/py_modules/lsfg_vk/flatpak_service.py deleted file mode 100644 index f486b74..0000000 --- a/py_modules/lsfg_vk/flatpak_service.py +++ /dev/null @@ -1,431 +0,0 @@ -"""Flatpak runtime support for classified Steam targets.""" - -from __future__ import annotations - -import json -import os -import pwd -import re -import shutil -import subprocess -import threading -from pathlib import Path -from typing import Dict, Optional, Set - -from .base_service import BaseService -from .constants import ( - BIN_DIR, - FLATPAK_23_08_FILENAME, - FLATPAK_24_08_FILENAME, - FLATPAK_25_08_FILENAME, -) - - -class FlatpakService(BaseService): - EXTENSION_ID = "org.freedesktop.Platform.VulkanLayer.lsfgvk" - SUPPORTED_RUNTIMES = ("23.08", "24.08", "25.08") - DERIVED_RUNTIME_IDS = {"org.gnome.Platform", "org.kde.Platform"} - RUNTIME_METADATA_SECTION = "Extension org.freedesktop.Platform.GL" - OWNERSHIP_FILENAME = "flatpak_extensions.json" - OWNERSHIP_VERSION = 1 - APP_ID_PATTERN = re.compile( - r"^[A-Za-z0-9][A-Za-z0-9-]*(?:\.[A-Za-z0-9][A-Za-z0-9-]*)+$" - ) - - def __init__(self, logger=None): - super().__init__(logger) - self.flatpak_command: Optional[str] = None - self._lock = threading.RLock() - - @property - def ownership_path(self) -> Path: - return self.config_dir / self.OWNERSHIP_FILENAME - - def _clean_env(self) -> Dict[str, str]: - env = os.environ.copy() - env.pop("LD_LIBRARY_PATH", None) - env["HOME"] = str(self.user_home) - path = [entry for entry in env.get("PATH", "").split(":") if entry] - for entry in ("/usr/bin", "/usr/local/bin", "/bin"): - if entry not in path: - path.insert(0, entry) - env["PATH"] = ":".join(path) - return env - - def check_flatpak_available(self) -> bool: - env = self._clean_env() - self.flatpak_command = shutil.which("flatpak", path=env["PATH"]) - return self.flatpak_command is not None - - def _run_flatpak_command(self, args, **kwargs): - if self.flatpak_command is None and not self.check_flatpak_available(): - raise FileNotFoundError("Flatpak command not available") - env = self._clean_env() - command = [self.flatpak_command, *args] - try: - user = pwd.getpwuid(self.user_home.stat().st_uid) - except (KeyError, OSError) as error: - raise RuntimeError(f"Unable to resolve Flatpak user for {self.user_home}") from error - if os.geteuid() != user.pw_uid: - runuser = shutil.which("runuser", path=env["PATH"]) - if runuser is None: - raise FileNotFoundError("runuser command not available") - command = [runuser, "--user", user.pw_name, "--", *command] - return subprocess.run(command, env=env, **kwargs) - - @classmethod - def _validate_app_id(cls, app_id: str) -> str: - if not isinstance(app_id, str) or not cls.APP_ID_PATTERN.fullmatch(app_id): - raise ValueError("Invalid Flatpak application ID") - return app_id - - @classmethod - def _validate_runtime(cls, branch: str) -> str: - if branch not in cls.SUPPORTED_RUNTIMES: - raise ValueError( - f"Unsupported Flatpak runtime branch {branch}; supported branches are " - + ", ".join(cls.SUPPORTED_RUNTIMES) - ) - return branch - - @classmethod - def runtime_branch_from_ref(cls, runtime_ref: str) -> str: - parts = runtime_ref.strip().split("/") if isinstance(runtime_ref, str) else [] - if len(parts) != 3 or parts[0] != "org.freedesktop.Platform": - raise ValueError(f"Unsupported Flatpak runtime reference: {runtime_ref}") - return cls._validate_runtime(parts[2]) - - @classmethod - def runtime_branch_from_metadata(cls, metadata: str) -> str: - section = None - versions = [] - for raw_line in metadata.splitlines() if isinstance(metadata, str) else []: - line = raw_line.strip() - if line.startswith("[") and line.endswith("]"): - section = line[1:-1].strip() - continue - if section != cls.RUNTIME_METADATA_SECTION: - continue - key, separator, value = line.partition("=") - if separator and key.strip() == "versions": - versions.extend(part.strip() for part in value.split(";")) - for value in versions: - for branch in cls.SUPPORTED_RUNTIMES: - if value == branch or value.startswith(f"{branch}-"): - return branch - raise ValueError("Could not determine a supported Freedesktop base runtime from Flatpak metadata") - - @classmethod - def _extension_ref(cls, branch: str) -> str: - return f"{cls.EXTENSION_ID}/x86_64/{cls._validate_runtime(branch)}" - - def _bundled_extension_path(self, branch: str) -> Path: - filename = { - "23.08": FLATPAK_23_08_FILENAME, - "24.08": FLATPAK_24_08_FILENAME, - "25.08": FLATPAK_25_08_FILENAME, - }[self._validate_runtime(branch)] - return Path(__file__).resolve().parent.parent.parent / BIN_DIR / filename - - def _installed_extension_branches(self, scope: Optional[str] = None) -> Set[str]: - scopes = ("user", "system") if scope is None else (scope,) - if any(item not in ("user", "system") for item in scopes): - raise ValueError("Flatpak installation scope must be user or system") - installed = set() - for item in scopes: - result = self._run_flatpak_command( - ["list", f"--{item}", "--runtime", "--columns=application,arch,branch"], - capture_output=True, - text=True, - check=True, - ) - for line in result.stdout.splitlines(): - fields = line.split("\t") if "\t" in line else line.split() - if len(fields) >= 3 and fields[0] == self.EXTENSION_ID and fields[1] == "x86_64": - installed.add(fields[2]) - return installed - - def _owned_branches(self) -> Set[str]: - path = self.ownership_path - if not path.exists() and not path.is_symlink(): - return set() - if path.is_symlink() or not path.is_file(): - raise RuntimeError("Flatpak ownership metadata is not a regular file") - try: - data = json.loads(path.read_text(encoding="utf-8")) - branches = data.get("plugin_owned_branches") - if data.get("version") != self.OWNERSHIP_VERSION or not isinstance(branches, list): - raise ValueError("invalid ownership metadata") - owned = {self._validate_runtime(branch) for branch in branches} - if len(owned) != len(branches): - raise ValueError("invalid ownership metadata") - return owned - except (OSError, json.JSONDecodeError, TypeError, ValueError) as error: - raise RuntimeError(f"Could not trust Flatpak ownership metadata: {error}") from error - - def _write_owned_branches(self, branches: Set[str]) -> None: - if not branches: - self.ownership_path.unlink(missing_ok=True) - return - self._write_file( - self.ownership_path, - json.dumps( - { - "version": self.OWNERSHIP_VERSION, - "plugin_owned_branches": sorted(branches), - }, - indent=2, - ) + "\n", - ) - - def get_extension_status(self): - try: - available = self.check_flatpak_available() - installed = self._installed_extension_branches() if available else set() - return self._success_response( - dict, - "Flatpak runtime extension status retrieved" if available else "Flatpak is not available", - available=available, - extension_id=self.EXTENSION_ID, - supported_branches=list(self.SUPPORTED_RUNTIMES), - installed_branches=sorted(installed), - ) - except Exception as error: - return self._error_response( - dict, - str(error), - available=False, - extension_id=self.EXTENSION_ID, - supported_branches=list(self.SUPPORTED_RUNTIMES), - installed_branches=[], - ) - - get_flatpak_support_status = get_extension_status - - def _resolve_runtime(self, app_id: str): - self._validate_app_id(app_id) - if not self.check_flatpak_available(): - raise FileNotFoundError("Flatpak is not available on this system") - result = self._run_flatpak_command( - ["info", "--show-runtime", app_id], - capture_output=True, - text=True, - ) - if result.returncode != 0: - raise OSError(result.stderr.strip() or f"Could not inspect Flatpak app {app_id}") - runtime = result.stdout.strip().splitlines()[0] if result.stdout.strip() else "" - parts = runtime.split("/") - if len(parts) != 3: - raise ValueError(f"Unsupported Flatpak runtime reference: {runtime}") - if parts[0] == "org.freedesktop.Platform": - branch = self._validate_runtime(parts[2]) - elif parts[0] in self.DERIVED_RUNTIME_IDS: - metadata_result = self._run_flatpak_command( - ["info", "--show-metadata", runtime], - capture_output=True, - text=True, - ) - if metadata_result.returncode != 0: - raise OSError( - metadata_result.stderr.strip() - or f"Could not inspect Flatpak runtime {runtime}" - ) - branch = self.runtime_branch_from_metadata(metadata_result.stdout) - else: - raise ValueError(f"Unsupported Flatpak runtime reference: {runtime}") - return runtime, branch - - def resolve_app_support(self, app_id: str): - try: - app_id = self._validate_app_id(app_id) - runtime, branch = self._resolve_runtime(app_id) - installed = self._installed_extension_branches() - ready = branch in installed - return self._success_response( - dict, - f"lsfg-vk support is ready for {app_id}" if ready - else f"lsfg-vk runtime extension {branch} is required for {app_id}", - flatpak_app_id=app_id, - runtime=runtime, - runtime_branch=branch, - support_status="ready" if ready else "needs-runtime", - extension_installed=ready, - installed_branches=sorted(installed), - ) - except ValueError as error: - return self._success_response( - dict, - str(error), - flatpak_app_id=app_id, - runtime=None, - runtime_branch=None, - support_status="unsupported", - extension_installed=False, - installed_branches=[], - error=str(error), - ) - except Exception as error: - return self._error_response( - dict, - str(error), - flatpak_app_id=app_id, - runtime=None, - runtime_branch=None, - support_status="error", - extension_installed=False, - installed_branches=[], - ) - - def install_extension(self, branch: str): - try: - branch = self._validate_runtime(branch) - if not self.check_flatpak_available(): - raise FileNotFoundError("Flatpak is not available on this system") - with self._lock: - if branch in self._installed_extension_branches(): - return self._extension_result(branch, True, False, "already installed") - bundle = self._bundled_extension_path(branch) - if not bundle.is_file(): - raise FileNotFoundError(f"Bundled Flatpak extension not found at {bundle}; reinstall the plugin") - result = self._run_flatpak_command( - ["install", "--user", "--noninteractive", "--or-update", str(bundle)], - capture_output=True, - text=True, - ) - if result.returncode != 0: - raise OSError(result.stderr.strip() or "Flatpak installation failed") - if branch not in self._installed_extension_branches("user"): - raise RuntimeError(f"Flatpak install completed but {self._extension_ref(branch)} was not visible afterwards") - owned = self._owned_branches() - owned.add(branch) - self._write_owned_branches(owned) - return self._extension_result(branch, True, False, "installed") - except Exception as error: - return self._error_response(dict, str(error), runtime_branch=branch, installed=False, enabled=False) - - def _remove_extension(self, branch: str) -> bool: - if branch not in self._installed_extension_branches("user"): - return False - result = self._run_flatpak_command( - ["uninstall", "--user", "--noninteractive", self._extension_ref(branch)], - capture_output=True, - text=True, - ) - if result.returncode != 0: - raise OSError(result.stderr.strip() or "Flatpak uninstall failed") - if branch in self._installed_extension_branches("user"): - raise RuntimeError(f"Flatpak uninstall completed but {self._extension_ref(branch)} is still installed") - return True - - def _extension_result(self, branch: str, installed: bool, removed: bool, verb: str): - return self._success_response( - dict, - f"lsfg-vk {branch} runtime extension {verb}", - runtime_branch=branch, - installed=installed, - enabled=installed, - removed=removed, - ) - - def uninstall_extension(self, branch: str): - try: - branch = self._validate_runtime(branch) - if not self.check_flatpak_available(): - raise FileNotFoundError("Flatpak is not available on this system") - with self._lock: - owned = self._owned_branches() - if branch not in owned: - installed = branch in self._installed_extension_branches() - return self._extension_result(branch, installed, False, "preserved (not plugin-owned)") - removed = self._remove_extension(branch) - owned.remove(branch) - self._write_owned_branches(owned) - installed = branch in self._installed_extension_branches() - return self._extension_result(branch, installed, removed, "uninstalled") - except Exception as error: - return self._error_response( - dict, - str(error), - runtime_branch=branch, - removed=False, - installed=False, - enabled=False, - ) - - def ensure_extension(self, branch: str): - try: - branch = self._validate_runtime(branch) - if branch in self._installed_extension_branches(): - return self._extension_result(branch, True, False, "is ready") - except Exception as error: - return self._error_response(dict, str(error), runtime_branch=branch, support_status="error") - return self.install_extension(branch) - - def ensure_app_support(self, app_id: str): - resolved = self.resolve_app_support(app_id) - if not resolved.get("success") or resolved.get("support_status") != "needs-runtime": - return resolved - result = self.ensure_extension(resolved["runtime_branch"]) - if not result.get("success"): - return self._error_response( - dict, - result.get("error") or "Could not install the required Flatpak runtime extension", - flatpak_app_id=app_id, - runtime=resolved.get("runtime"), - runtime_branch=resolved.get("runtime_branch"), - support_status="error", - extension_installed=False, - ) - return self.resolve_app_support(app_id) - - def set_extension_enabled(self, branch: str, enabled: bool): - if type(enabled) is not bool: - return self._error_response(dict, "enabled must be a boolean", runtime_branch=branch, installed=False, enabled=False) - return self.install_extension(branch) if enabled else self.uninstall_extension(branch) - - def remove_plugin_owned_extensions(self): - try: - with self._lock: - owned = self._owned_branches() - if not owned: - return self._success_response( - dict, - "No plugin-owned Flatpak extensions to remove", - removed_branches=[], - preserved_branches=[], - ownership_uncertain=False, - ) - if not self.check_flatpak_available(): - raise RuntimeError("Flatpak is not available; plugin-owned extension metadata was preserved") - removed, failures = [], [] - for branch in sorted(owned): - try: - self._remove_extension(branch) - removed.append(branch) - except Exception as error: - failures.append(f"{branch}: {error}") - remaining = owned - set(removed) - self._write_owned_branches(remaining) - if failures: - return self._error_response( - dict, - "; ".join(failures), - removed_branches=removed, - preserved_branches=sorted(remaining), - ownership_uncertain=False, - ) - return self._success_response( - dict, - "Plugin-owned Flatpak extensions removed", - removed_branches=removed, - preserved_branches=[], - ownership_uncertain=False, - ) - except Exception as error: - return self._error_response( - dict, - str(error), - removed_branches=[], - preserved_branches=[], - ownership_uncertain=True, - ) diff --git a/py_modules/lsfg_vk/plugin.py b/py_modules/lsfg_vk/plugin.py index 13df7a4..b010db6 100644 --- a/py_modules/lsfg_vk/plugin.py +++ b/py_modules/lsfg_vk/plugin.py @@ -1,10 +1,9 @@ import os -from typing import Any, Dict, Optional +from typing import Any, Dict import decky from .configuration import ConfigurationService -from .flatpak_service import FlatpakService from .installation import InstallationService from .runtime_service import RuntimeService from .steam_service import SteamService @@ -20,7 +19,6 @@ class Plugin: steam_service=self.steam_service, ) self.configuration_service = ConfigurationService(runtime_service=self.runtime_service) - self.flatpak_service = FlatpakService() self.wrapper_service = WrapperService() async def install_lsfg_vk(self): @@ -36,21 +34,7 @@ class Plugin: return self.configuration_service.get_game_configs() async def get_installed_games(self): - result = self.steam_service.get_installed_games() - if not result.get("success"): - return result - cache: Dict[str, Dict[str, Any]] = {} - for game in result.get("games", []): - transport = game.get("transport", {}) - if transport.get("kind") != "flatpak": - continue - app_id = transport.get("flatpakAppId") - if not app_id: - continue - if app_id not in cache: - cache[app_id] = self.flatpak_service.resolve_app_support(app_id) - game["flatpakSupport"] = cache[app_id] - return result + return self.steam_service.get_installed_games() async def update_game_config(self, appid: str, game_name: str, config: Dict[str, Any]): return self.configuration_service.update_game_config(appid, game_name, config) @@ -68,17 +52,9 @@ class Plugin: self, appid: str, state: Dict[str, Any], - shortcut_exe: Optional[str] = None, command_token_added: bool = False, - transport: Optional[Dict[str, Any]] = None, ): - return self.wrapper_service.set( - appid, - state, - shortcut_exe, - command_token_added, - transport, - ) + return self.wrapper_service.set(appid, state, command_token_added) async def remove_workaround_state(self, appid: str): return self.wrapper_service.remove(appid) @@ -112,7 +88,6 @@ class Plugin: ("config", "LSFG-VK configuration", self.configuration_service.config_file_path), ("workarounds", "Per-app workarounds", self.wrapper_service.sidecar_path), ("wrapper", "Generated launch wrapper", self.wrapper_service.wrapper_path), - ("flatpak_extensions", "Flatpak extension ownership", self.flatpak_service.ownership_path), ) contents = [] for file_id, label, path in files: @@ -147,18 +122,6 @@ class Plugin: async def get_lossless_scaling_branch_status(self): return self.steam_service.get_branch_status() - async def get_flatpak_support_status(self): - return self.flatpak_service.get_flatpak_support_status() - - async def ensure_flatpak_support(self, flatpak_app_id: str): - return self.flatpak_service.ensure_app_support(flatpak_app_id) - - async def repair_flatpak_support(self, flatpak_app_id: str): - return self.flatpak_service.ensure_app_support(flatpak_app_id) - - async def set_flatpak_extension_enabled(self, version: str, enabled: bool): - return self.flatpak_service.set_extension_enabled(version, enabled) - async def _main(self): repair = self.wrapper_service.repair() if not repair.get("success"): @@ -171,12 +134,6 @@ class Plugin: async def _uninstall(self): decky.logger.info("decky-lsfg-vk plugin being uninstalled") self.installation_service.cleanup_on_uninstall() - try: - result = self.flatpak_service.remove_plugin_owned_extensions() - if not result.get("success"): - decky.logger.warning(result.get("error")) - except Exception as error: - decky.logger.error(f"Error during Flatpak cleanup: {error}") decky.logger.info("decky-lsfg-vk plugin uninstall cleanup completed") async def _migration(self): diff --git a/py_modules/lsfg_vk/steam_service.py b/py_modules/lsfg_vk/steam_service.py index a952fcb..53748de 100644 --- a/py_modules/lsfg_vk/steam_service.py +++ b/py_modules/lsfg_vk/steam_service.py @@ -1,5 +1,4 @@ import re -import shlex from pathlib import Path from typing import Dict, Optional, Tuple @@ -7,51 +6,8 @@ from .base_service import BaseService from .constants import ( STEAM_LOSSLESS_SCALING_APP_ID, STEAM_LOSSLESS_SCALING_BRANCH, - WRAPPER_FILENAME, ) -_FLATPAK_APP_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9-]*(?:\.[A-Za-z0-9][A-Za-z0-9-]*)+$") -_WRAPPER_TOKEN = f"~/{WRAPPER_FILENAME}" - - -def _split_command(value: Optional[str]) -> Optional[list[str]]: - if not isinstance(value, str) or not value.strip(): - return [] - try: - return shlex.split(value, posix=True) - except ValueError: - return None - - -def _is_managed_wrapper(value: str) -> bool: - if value in {_WRAPPER_TOKEN, f"$HOME/{WRAPPER_FILENAME}"}: - return True - path = Path(value) - return path.is_absolute() and path.name == WRAPPER_FILENAME - - -def classify_shortcut_transport(executable: Optional[str], launch_options: Optional[str] = None) -> Dict[str, object]: - executable_tokens = _split_command(executable) - option_tokens = _split_command(launch_options) - if executable_tokens is None or option_tokens is None or not executable_tokens: - return {"kind": "host"} - direct_flatpak = executable_tokens[0] in {"flatpak", "/usr/bin/flatpak"} - managed_wrapper = len(executable_tokens) == 1 and _is_managed_wrapper(executable_tokens[0]) - if not direct_flatpak and not managed_wrapper: - return {"kind": "host"} - arguments = [*executable_tokens[1:], *option_tokens] - if not arguments or arguments[0] != "run": - return {"kind": "host"} - for argument in arguments[1:]: - if argument == "--" or argument.startswith("-"): - continue - return ( - {"kind": "flatpak", "flatpakAppId": argument} - if _FLATPAK_APP_ID.fullmatch(argument) - else {"kind": "host"} - ) - return {"kind": "host"} - def _first_string(values: Dict[str, object], *keys: str) -> Optional[str]: return next((values[key] for key in keys if isinstance(values.get(key), str)), None) @@ -74,7 +30,6 @@ class SteamService(BaseService): self.user_home / ".local/share/Steam", self.user_home / ".steam/steam", self.user_home / ".steam/root", - self.user_home / ".var/app/com.valvesoftware.Steam/.local/share/Steam", ): yield from self._unique_existing_root(candidate, seen) @@ -154,7 +109,6 @@ class SteamService(BaseService): "appid": str(appid & 0xFFFFFFFF), "name": name, "nonSteam": True, - "transport": classify_shortcut_transport(executable, arguments), } for key, value in (("executable", executable), ("arguments", arguments), ("startDir", start_dir)): if value is not None: @@ -309,7 +263,6 @@ class SteamService(BaseService): "appid": appid, "name": self._section_value(content, "AppState", "name") or f"App {appid}", "nonSteam": False, - "transport": {"kind": "host"}, } for game in self._shortcut_games(): games.setdefault(str(game["appid"]), game) diff --git a/py_modules/lsfg_vk/wrapper_service.py b/py_modules/lsfg_vk/wrapper_service.py index 980f7ae..69a3600 100644 --- a/py_modules/lsfg_vk/wrapper_service.py +++ b/py_modules/lsfg_vk/wrapper_service.py @@ -4,9 +4,7 @@ from __future__ import annotations import json import re -import shlex import threading -from pathlib import Path from typing import Any, Dict, Optional, Tuple from .base_service import BaseService @@ -85,52 +83,19 @@ class WrapperService(BaseService): raise ValueError(f"{field} must be a boolean") return state - @classmethod - def _validate_transport(cls, raw: Any) -> Dict[str, Any]: - if raw is None: - return {"kind": "host"} - 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} - 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") + unsupported = set(raw) - {"state", "command_token_added"} + if unsupported: + raise ValueError("Unsupported workaround entry fields: " + ", ".join(sorted(unsupported))) 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") - entry["shortcut_exe"] = shortcut_exe return entry @classmethod @@ -186,23 +151,8 @@ class WrapperService(BaseService): ) return True - @staticmethod - 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]: + def _state_lines(cls, state: Dict[str, Any]) -> list[str]: lines = [" unset " + " ".join(cls.MANAGED_ENV_KEYS)] if state["disableGamescopeWsi"]: lines.extend([" ENABLE_GAMESCOPE_WSI=0", " export ENABLE_GAMESCOPE_WSI"]) @@ -235,68 +185,8 @@ class WrapperService(BaseService): " fi", " export DXVK_CONFIG", ]) - lines.append(f" shortcut_exe={cls._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", @@ -320,69 +210,16 @@ class WrapperService(BaseService): ' *) appid="${STEAM_COMPAT_APP_ID}" ;;', " esac", "fi", - "shortcut_exe=", 'case "$appid" in', ] for appid in sorted(document["apps"], key=lambda value: int(value)): entry = document["apps"][appid] lines.append(f" {appid})") - lines.extend(self._state_lines(entry["state"], entry.get("shortcut_exe"))) + lines.extend(self._state_lines(entry["state"])) lines.append(" ;;") lines.extend([ "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", - ' exec "$shortcut_exe" "$@"', - "fi", 'exec "$@"', "", ]) @@ -424,9 +261,7 @@ class WrapperService(BaseService): "state": dict(entry["state"]) if entry else None, "wrapper_path": self.WRAPPER_TOKEN, "wrapper_owned": self._wrapper_marker() if document["apps"] else False, - "shortcut_exe": entry.get("shortcut_exe") if entry else None, "command_token_added": entry.get("command_token_added", False) if entry else False, - "transport": dict(entry.get("transport", {"kind": "host"})) if entry else None, } def get(self, appid: str) -> Dict[str, Any]: @@ -451,9 +286,7 @@ class WrapperService(BaseService): self, appid: str, state: Dict[str, Any], - shortcut_exe: Optional[str] = None, command_token_added: bool = False, - transport: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: try: normalized = self._valid_appid(appid) @@ -463,26 +296,10 @@ class WrapperService(BaseService): with self._lock: self._assert_wrapper_owned_or_absent() 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 - ) - ) 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"] document["apps"][normalized] = entry self._write_pair(document) return self._response(document, normalized) -- cgit v1.2.3