diff options
Diffstat (limited to 'py_modules/lsfg_vk/wrapper_service.py')
| -rw-r--r-- | py_modules/lsfg_vk/wrapper_service.py | 139 |
1 files changed, 123 insertions, 16 deletions
diff --git a/py_modules/lsfg_vk/wrapper_service.py b/py_modules/lsfg_vk/wrapper_service.py index a3cba6e..dae565b 100644 --- a/py_modules/lsfg_vk/wrapper_service.py +++ b/py_modules/lsfg_vk/wrapper_service.py @@ -16,8 +16,10 @@ from .constants import WRAPPER_FILENAME class WrapperService(BaseService): """Persist workaround state and compile it into a safe POSIX wrapper.""" - FORMAT_VERSION = 1 - MARKER = "# lsfg-vk-wrapper-format: 1" + LEGACY_FORMAT_VERSION = 1 + FORMAT_VERSION = 2 + LEGACY_MARKER = "# lsfg-vk-wrapper-format: 1" + MARKER = "# lsfg-vk-wrapper-format: 2" WRAPPER_TOKEN = "~/.lsfg" STATE_FIELDS = ( "dxvkFrameRate", @@ -82,12 +84,38 @@ class WrapperService(BaseService): 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") 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") @@ -105,7 +133,10 @@ class WrapperService(BaseService): @classmethod def _validate_document(cls, raw: Any) -> Dict[str, Any]: - if not isinstance(raw, dict) or raw.get("version") != cls.FORMAT_VERSION: + if not isinstance(raw, dict) or raw.get("version") not in ( + cls.LEGACY_FORMAT_VERSION, + cls.FORMAT_VERSION, + ): raise ValueError("Unsupported lsfg-vk workaround state version") apps = raw.get("apps") if not isinstance(apps, dict): @@ -142,7 +173,7 @@ class WrapperService(BaseService): prefix = "\n".join(self.wrapper_path.read_text(encoding="utf-8").splitlines()[:8]) except OSError as error: raise RuntimeError(f"Could not read lsfg wrapper: {error}") from error - return self.MARKER in prefix + return self.MARKER in prefix or self.LEGACY_MARKER in prefix def _assert_wrapper_owned_or_absent(self) -> bool: if not self.wrapper_path.exists() and not self.wrapper_path.is_symlink(): @@ -157,6 +188,17 @@ 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)] @@ -194,9 +236,31 @@ class WrapperService(BaseService): lines.append(f" shortcut_exe={cls._shell(shortcut_exe or '')}") return lines - @classmethod - def _flatpak_args(cls, state: Dict[str, Any]) -> list[str]: + 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_GAMESCOPE_WSI"', '"--unset-env=ENABLE_GAMESCOPE_WSI"' if not state["disableGamescopeWsi"] else @@ -230,11 +294,10 @@ class WrapperService(BaseService): " fi", ] - @classmethod - def _render_wrapper(cls, document: Dict[str, Any]) -> str: + def _render_wrapper(self, document: Dict[str, Any]) -> str: lines = [ "#!/bin/sh", - cls.MARKER, + self.MARKER, "# Generated by Decky LSFG-VK; edits will be rejected on the next update.", "", "appid=", @@ -260,29 +323,61 @@ class WrapperService(BaseService): for appid in sorted(document["apps"], key=lambda value: int(value)): entry = document["apps"][appid] lines.append(f" {appid})") - lines.extend(cls._state_lines(entry["state"], entry.get("shortcut_exe"))) + lines.extend(self._state_lines(entry["state"], entry.get("shortcut_exe"))) lines.append(" ;;") lines.extend([ "esac", "", 'if [ -n "$shortcut_exe" ]; then', - ' if [ "${1-}" = "run" ]; then', - ' flatpak_command="$1"', - " shift", ]) # 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] - if not entry.get("shortcut_exe", "").endswith("/flatpak"): + 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(cls._flatpak_args(entry["state"])) + 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", - " fi", ' exec "$shortcut_exe" "$@"', "fi", 'exec "$@"', @@ -328,6 +423,7 @@ class WrapperService(BaseService): "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]: @@ -354,6 +450,7 @@ class WrapperService(BaseService): 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) @@ -364,9 +461,19 @@ class WrapperService(BaseService): 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 shortcut_exe is not None: entry = self._validate_entry({**entry, "shortcut_exe": shortcut_exe}) |