From e64e87e9eb9e3c183ad7807dffa356f47d14e825 Mon Sep 17 00:00:00 2001 From: Kurt Himebauch <136133082+xXJSONDeruloXx@users.noreply.github.com> Date: Sat, 5 Sep 2026 15:33:30 -0400 Subject: fix: migrate flatpak integration to v2 --- py_modules/lsfg_vk/flatpak_service.py | 662 ++++++++++++++-------------------- 1 file changed, 262 insertions(+), 400 deletions(-) (limited to 'py_modules/lsfg_vk/flatpak_service.py') diff --git a/py_modules/lsfg_vk/flatpak_service.py b/py_modules/lsfg_vk/flatpak_service.py index c9be0ec..7aa3cda 100644 --- a/py_modules/lsfg_vk/flatpak_service.py +++ b/py_modules/lsfg_vk/flatpak_service.py @@ -1,459 +1,321 @@ -""" -Flatpak service for managing lsfg-vk Flatpak runtime extensions. -""" - -import subprocess import os +import shutil +import subprocess from pathlib import Path -from typing import Dict, Any, List, Optional +from typing import Any, Dict, List from .base_service import BaseService -from .constants import ( - FLATPAK_23_08_FILENAME, FLATPAK_24_08_FILENAME, FLATPAK_25_08_FILENAME, BIN_DIR, CONFIG_DIR -) +from .config_schema import ConfigurationManager +from .dll_detection import DllDetectionService from .types import BaseResponse -class FlatpakExtensionStatus(BaseResponse): - """Response for Flatpak extension status""" - def __init__(self, success: bool = False, message: str = "", error: str = "", - installed_23_08: bool = False, installed_24_08: bool = False, installed_25_08: bool = False): - super().__init__(success, message, error) - self.installed_23_08 = installed_23_08 - self.installed_24_08 = installed_24_08 - self.installed_25_08 = installed_25_08 - - -class FlatpakAppInfo(BaseResponse): - """Response for Flatpak app information""" - def __init__(self, success: bool = False, message: str = "", error: str = "", - apps: List[Dict[str, Any]] = None, total_apps: int = 0): - super().__init__(success, message, error) - self.apps = apps or [] - self.total_apps = total_apps - - -class FlatpakOverrideResponse(BaseResponse): - """Response for Flatpak override operations""" - def __init__(self, success: bool = False, message: str = "", error: str = "", - app_id: str = "", operation: str = ""): - super().__init__(success, message, error) - self.app_id = app_id - self.operation = operation - - class FlatpakService(BaseService): - """Service for handling Flatpak runtime extensions and app overrides""" + EXTENSION_ID = "org.freedesktop.Platform.VulkanLayer.lsfgvk" + SUPPORTED_RUNTIMES = ("24.08", "25.08") def __init__(self, logger=None): super().__init__(logger) - self.extension_id_23_08 = "org.freedesktop.Platform.VulkanLayer.lsfgvk/x86_64/23.08" - self.extension_id_24_08 = "org.freedesktop.Platform.VulkanLayer.lsfgvk/x86_64/24.08" - self.extension_id_25_08 = "org.freedesktop.Platform.VulkanLayer.lsfgvk/x86_64/25.08" self.flatpak_command = None - def _get_clean_env(self): - """Get a clean environment without PyInstaller's bundled libraries""" + def _get_clean_env(self) -> Dict[str, str]: env = os.environ.copy() - - if 'LD_LIBRARY_PATH' in env: - del env['LD_LIBRARY_PATH'] - - standard_paths = ['/usr/bin', '/usr/local/bin', '/bin'] - current_path = env.get('PATH', '') - - path_parts = current_path.split(':') if current_path else [] - for std_path in standard_paths: - if std_path not in path_parts: - path_parts.insert(0, std_path) - - env['PATH'] = ':'.join(path_parts) - + env.pop("LD_LIBRARY_PATH", None) + path_entries = [entry for entry in env.get("PATH", "").split(":") if entry] + for entry in ("/usr/bin", "/usr/local/bin", "/bin"): + if entry not in path_entries: + path_entries.insert(0, entry) + env["PATH"] = ":".join(path_entries) return env - def _run_flatpak_command(self, args: List[str], **kwargs): - """Run flatpak command with clean environment to avoid library conflicts""" - if self.flatpak_command is None: - raise FileNotFoundError("Flatpak command not available") - - env = self._get_clean_env() - - self.log.info(f"Running flatpak with PATH: {env.get('PATH')}") - self.log.info(f"LD_LIBRARY_PATH removed: {'LD_LIBRARY_PATH' not in env}") - - return subprocess.run([self.flatpak_command] + args, env=env, **kwargs) - def check_flatpak_available(self) -> bool: - """Check if flatpak command is available and store the working command""" - self.log.info(f"PATH: {os.environ.get('PATH', 'Not set')}") - self.log.info(f"HOME: {os.environ.get('HOME', 'Not set')}") - self.log.info(f"USER: {os.environ.get('USER', 'Not set')}") - - flatpak_paths = [ - "flatpak", - "/usr/bin/flatpak", - "/var/lib/flatpak/exports/bin/flatpak", - "/home/deck/.local/bin/flatpak" - ] - - for flatpak_path in flatpak_paths: - try: - result = subprocess.run([flatpak_path, "--version"], - capture_output=True, check=True, text=True, - env=self._get_clean_env()) - self.log.info(f"Flatpak found at {flatpak_path}: {result.stdout.strip()}") - self.flatpak_command = flatpak_path - return True - except (subprocess.CalledProcessError, FileNotFoundError): - self.log.debug(f"Flatpak not found at {flatpak_path}") - continue - - self.log.error("Flatpak command not found in any known locations") - self.flatpak_command = None - return False + env = self._get_clean_env() + self.flatpak_command = shutil.which("flatpak", path=env["PATH"]) + return self.flatpak_command is not None - def get_extension_status(self) -> FlatpakExtensionStatus: - """Check if lsfg-vk Flatpak extensions are installed""" + def _run_flatpak_command(self, args: List[str], **kwargs): + if self.flatpak_command is None and not self.check_flatpak_available(): + raise FileNotFoundError("Flatpak command not available") + return subprocess.run( + [self.flatpak_command, *args], + env=self._get_clean_env(), + **kwargs, + ) + + @classmethod + def _extension_ref(cls, version: str) -> str: + return f"{cls.EXTENSION_ID}/x86_64/{version}" + + @classmethod + def _validate_runtime(cls, version: str) -> None: + if version not in cls.SUPPORTED_RUNTIMES: + raise ValueError("Unsupported Flatpak runtime") + + def get_extension_status(self) -> Dict[str, Any]: try: if not self.check_flatpak_available(): - error_msg = "Flatpak is not available on this system" - if self.flatpak_command is None: - error_msg += ". Command not found in PATH or common install locations." - self.log.error(error_msg) - return self._error_response(FlatpakExtensionStatus, - error_msg, - installed_23_08=False, installed_24_08=False, installed_25_08=False) + raise FileNotFoundError("Flatpak is not available on this system") result = self._run_flatpak_command( - ["list", "--runtime"], - capture_output=True, text=True, check=True + ["list", "--user", "--runtime", "--columns=application,arch,branch"], + capture_output=True, + text=True, + check=True, + ) + installed = { + tuple(line.split("\t")[:3]) + for line in result.stdout.splitlines() + if line.strip() + } + return self._success_response( + BaseResponse, + "Flatpak runtime status retrieved", + installed_24_08=(self.EXTENSION_ID, "x86_64", "24.08") in installed, + installed_25_08=(self.EXTENSION_ID, "x86_64", "25.08") in installed, + ) + except Exception as error: + return self._error_response( + BaseResponse, + str(error), + installed_24_08=False, + installed_25_08=False, ) - installed_runtimes = result.stdout - - base_extension_name = "org.freedesktop.Platform.VulkanLayer.lsfgvk" - installed_23_08 = False - installed_24_08 = False - installed_25_08 = False - - for line in installed_runtimes.split('\n'): - if base_extension_name in line: - if "23.08" in line: - installed_23_08 = True - elif "24.08" in line: - installed_24_08 = True - elif "25.08" in line: - installed_25_08 = True - - status_msg = [] - if installed_23_08: - status_msg.append("23.08 runtime extension installed") - if installed_24_08: - status_msg.append("24.08 runtime extension installed") - if installed_25_08: - status_msg.append("25.08 runtime extension installed") - - if not status_msg: - status_msg.append("No lsfg-vk runtime extensions installed") - - return self._success_response(FlatpakExtensionStatus, - "; ".join(status_msg), - installed_23_08=installed_23_08, - installed_24_08=installed_24_08, - installed_25_08=installed_25_08) - - except subprocess.CalledProcessError as e: - error_msg = f"Error checking Flatpak extensions: {e.stderr if e.stderr else str(e)}" - self.log.error(error_msg) - return self._error_response(FlatpakExtensionStatus, error_msg, - installed_23_08=False, installed_24_08=False, installed_25_08=False) - - def install_extension(self, version: str) -> BaseResponse: - """Install a specific version of the lsfg-vk Flatpak extension""" + def install_extension(self, version: str) -> Dict[str, Any]: try: - if version not in ["23.08", "24.08", "25.08"]: - return self._error_response(BaseResponse, "Invalid version. Must be '23.08', '24.08', or '25.08'") - + self._validate_runtime(version) if not self.check_flatpak_available(): - return self._error_response(BaseResponse, "Flatpak is not available on this system") - - plugin_dir = Path(__file__).parent.parent.parent - if version == "23.08": - filename = FLATPAK_23_08_FILENAME - elif version == "24.08": - filename = FLATPAK_24_08_FILENAME - else: - filename = FLATPAK_25_08_FILENAME - flatpak_path = plugin_dir / BIN_DIR / filename - - if not flatpak_path.exists(): - return self._error_response(BaseResponse, f"Flatpak file not found: {flatpak_path}") - + raise FileNotFoundError("Flatpak is not available on this system") result = self._run_flatpak_command( - ["install", "--user", "--noninteractive", str(flatpak_path)], - capture_output=True, text=True + [ + "install", + "--user", + "--noninteractive", + "--or-update", + "flathub", + f"{self.EXTENSION_ID}//{version}", + ], + capture_output=True, + text=True, ) - if result.returncode != 0: - error_msg = f"Failed to install Flatpak extension: {result.stderr}" - self.log.error(error_msg) - return self._error_response(BaseResponse, error_msg) - - self.log.info(f"Successfully installed lsfg-vk Flatpak extension {version}") - return self._success_response(BaseResponse, f"lsfg-vk {version} runtime extension installed successfully") - - except Exception as e: - error_msg = f"Error installing Flatpak extension {version}: {str(e)}" - self.log.error(error_msg) - return self._error_response(BaseResponse, error_msg) + raise OSError(result.stderr.strip() or "Flatpak installation failed") + return self._success_response( + BaseResponse, + f"lsfg-vk {version} runtime extension installed", + ) + except Exception as error: + return self._error_response(BaseResponse, str(error)) - def uninstall_extension(self, version: str) -> BaseResponse: - """Uninstall a specific version of the lsfg-vk Flatpak extension""" + def uninstall_extension(self, version: str) -> Dict[str, Any]: try: - if version not in ["23.08", "24.08", "25.08"]: - return self._error_response(BaseResponse, "Invalid version. Must be '23.08', '24.08', or '25.08'") - + self._validate_runtime(version) if not self.check_flatpak_available(): - return self._error_response(BaseResponse, "Flatpak is not available on this system") - - if version == "23.08": - extension_id = self.extension_id_23_08 - elif version == "24.08": - extension_id = self.extension_id_24_08 - else: - extension_id = self.extension_id_25_08 - + raise FileNotFoundError("Flatpak is not available on this system") result = self._run_flatpak_command( - ["uninstall", "--user", "--noninteractive", extension_id], - capture_output=True, text=True + ["uninstall", "--user", "--noninteractive", self._extension_ref(version)], + capture_output=True, + text=True, ) - if result.returncode != 0: - error_msg = f"Failed to uninstall Flatpak extension: {result.stderr}" - self.log.error(error_msg) - return self._error_response(BaseResponse, error_msg) - - self.log.info(f"Successfully uninstalled lsfg-vk Flatpak extension {version}") - return self._success_response(BaseResponse, f"lsfg-vk {version} runtime extension uninstalled successfully") + raise OSError(result.stderr.strip() or "Flatpak uninstall failed") + return self._success_response( + BaseResponse, + f"lsfg-vk {version} runtime extension uninstalled", + ) + except Exception as error: + return self._error_response(BaseResponse, str(error)) - except Exception as e: - error_msg = f"Error uninstalling Flatpak extension {version}: {str(e)}" - self.log.error(error_msg) - return self._error_response(BaseResponse, error_msg) + def _dll_directory(self) -> Path: + if self.config_file_path.exists(): + try: + profile_data = ConfigurationManager.parse_toml_content_multi_profile( + self.config_file_path.read_text(encoding="utf-8") + ) + dll_path = profile_data["global_config"].get("dll") + if dll_path: + return Path(dll_path).parent + except Exception: + pass + + result = DllDetectionService(self.log).check_lossless_scaling_dll() + if result.get("detected") and result.get("path"): + return Path(result["path"]).parent + return self.user_home / ".local/share/Steam/steamapps/common" + + def _override_output(self, app_id: str) -> str: + result = self._run_flatpak_command( + ["override", "--user", "--show", app_id], + capture_output=True, + text=True, + ) + return result.stdout if result.returncode == 0 else "" + + def _override_paths(self) -> Dict[str, str]: + return { + "config_dir": str(self.config_dir), + "config_file": str(self.config_file_path), + "dll_dir": str(self._dll_directory()), + "legacy_dll": str( + self.user_home + / ".local/share/Steam/steamapps/common/Lossless Scaling/Lossless.dll" + ), + "legacy_script": str(self.lsfg_launch_script_path), + } - def get_flatpak_apps(self) -> FlatpakAppInfo: - """Get list of installed Flatpak apps and their lsfg-vk override status""" + def _check_app_override_status(self, app_id: str) -> Dict[str, bool]: + output = self._override_output(app_id) + paths = self._override_paths() + return { + "filesystem": ( + paths["config_dir"] in output + and paths["dll_dir"] in output + ), + "env": f"LSFGVK_CONFIG={paths['config_file']}" in output, + } + + def get_flatpak_apps(self) -> Dict[str, Any]: try: if not self.check_flatpak_available(): - error_msg = "Flatpak is not available on this system" - if self.flatpak_command is None: - error_msg += ". Command not found in PATH or common install locations." - return self._error_response(FlatpakAppInfo, - error_msg, - apps=[], total_apps=0) - + raise FileNotFoundError("Flatpak is not available on this system") result = self._run_flatpak_command( - ["list", "--app"], - capture_output=True, text=True, check=True + ["list", "--user", "--app", "--columns=name,application"], + capture_output=True, + text=True, + check=True, ) - apps = [] - for line in result.stdout.strip().split('\n'): - if not line.strip(): + for line in result.stdout.splitlines(): + parts = line.split("\t", 1) + if len(parts) != 2: continue + status = self._check_app_override_status(parts[1]) + apps.append( + { + "app_id": parts[1], + "app_name": parts[0], + "has_filesystem_override": status["filesystem"], + "has_env_override": status["env"], + } + ) + return self._success_response( + BaseResponse, + f"Found {len(apps)} Flatpak applications", + apps=apps, + total_apps=len(apps), + ) + except Exception as error: + return self._error_response( + BaseResponse, + str(error), + apps=[], + total_apps=0, + ) - parts = line.split('\t') - if len(parts) >= 2: - app_name = parts[0].strip() - app_id = parts[1].strip() - - # Check override status - override_status = self._check_app_override_status(app_id) - - apps.append({ - "app_id": app_id, - "app_name": app_name, - "has_filesystem_override": override_status["filesystem"], - "has_env_override": override_status["env"] - }) - - return self._success_response(FlatpakAppInfo, - f"Found {len(apps)} Flatpak applications", - apps=apps, total_apps=len(apps)) - - except subprocess.CalledProcessError as e: - error_msg = f"Error getting Flatpak apps: {e.stderr if e.stderr else str(e)}" - self.log.error(error_msg) - return self._error_response(FlatpakAppInfo, error_msg, apps=[], total_apps=0) - - def _check_app_override_status(self, app_id: str) -> Dict[str, bool]: - """Check if an app has lsfg-vk overrides set""" + def set_app_override(self, app_id: str) -> Dict[str, Any]: try: + if not self.check_flatpak_available(): + raise FileNotFoundError("Flatpak is not available on this system") + paths = self._override_paths() result = self._run_flatpak_command( - ["override", "--user", "--show", app_id], - capture_output=True, text=True + [ + "override", + "--user", + f"--filesystem={paths['config_dir']}:rw", + f"--filesystem={paths['dll_dir']}:ro", + f"--env=LSFGVK_CONFIG={paths['config_file']}", + f"--nofilesystem={paths['legacy_dll']}", + f"--nofilesystem={paths['legacy_script']}", + "--unset-env=LSFG_CONFIG", + app_id, + ], + capture_output=True, + text=True, ) - if result.returncode != 0: - return {"filesystem": False, "env": False} - - output = result.stdout - home_path = os.path.expanduser("~") - config_path = f"{home_path}/.config/lsfg-vk" - dll_path = f"{home_path}/.local/share/Steam/steamapps/common/Lossless Scaling/Lossless.dll" - lsfg_path = f"{home_path}/lsfg" - - filesystem_section = "" - in_context = False - - for line in output.split('\n'): - line = line.strip() - if line == "[Context]": - in_context = True - elif line.startswith("[") and line != "[Context]": - in_context = False - elif in_context and line.startswith("filesystems="): - filesystem_section = line - break - - has_config_fs = config_path in filesystem_section - has_dll_fs = dll_path in filesystem_section - has_lsfg_fs = lsfg_path in filesystem_section - - filesystem_override = has_config_fs and has_dll_fs and has_lsfg_fs - - env_override = False - in_environment = False - - for line in output.split('\n'): - line = line.strip() - if line == "[Environment]": - in_environment = True - elif line.startswith("[") and line != "[Environment]": - in_environment = False - elif in_environment and line.startswith(f"LSFG_CONFIG={config_path}/conf.toml"): - env_override = True - break - - self.log.debug(f"Override status for {app_id}: filesystem={filesystem_override} ({has_config_fs}/{has_dll_fs}/{has_lsfg_fs}), env={env_override}") - - return {"filesystem": filesystem_override, "env": env_override} - - except Exception as e: - self.log.error(f"Error checking override status for {app_id}: {e}") - return {"filesystem": False, "env": False} + raise OSError(result.stderr.strip() or "Failed to set Flatpak overrides") + return self._success_response( + BaseResponse, + f"lsfg-vk overrides set for {app_id}", + app_id=app_id, + operation="set", + ) + except Exception as error: + return self._error_response( + BaseResponse, + str(error), + app_id=app_id, + operation="set", + ) - def set_app_override(self, app_id: str) -> FlatpakOverrideResponse: - """Set lsfg-vk overrides for a Flatpak app""" + def remove_app_override(self, app_id: str) -> Dict[str, Any]: try: if not self.check_flatpak_available(): - return self._error_response(FlatpakOverrideResponse, - "Flatpak is not available on this system", - app_id=app_id, operation="set") - - home_path = os.path.expanduser("~") - config_path = f"{home_path}/.config/lsfg-vk" - dll_path = f"{home_path}/.local/share/Steam/steamapps/common/Lossless Scaling/Lossless.dll" - lsfg_path = f"{home_path}/lsfg" - - filesystem_overrides = [ - f"--filesystem={dll_path}", - f"--filesystem={config_path}:rw", - f"--filesystem={lsfg_path}:rw" - ] - - for override in filesystem_overrides: - result = self._run_flatpak_command( - ["override", "--user", override, app_id], - capture_output=True, text=True - ) - if result.returncode != 0: - error_msg = f"Failed to set filesystem override {override}: {result.stderr}" - return self._error_response(FlatpakOverrideResponse, error_msg, - app_id=app_id, operation="set") - + raise FileNotFoundError("Flatpak is not available on this system") + paths = self._override_paths() result = self._run_flatpak_command( - ["override", "--user", f"--env=LSFG_CONFIG={config_path}/conf.toml", app_id], - capture_output=True, text=True + [ + "override", + "--user", + f"--nofilesystem={paths['config_dir']}", + f"--nofilesystem={paths['dll_dir']}", + f"--nofilesystem={paths['legacy_dll']}", + f"--nofilesystem={paths['legacy_script']}", + "--unset-env=LSFGVK_CONFIG", + "--unset-env=LSFG_CONFIG", + app_id, + ], + capture_output=True, + text=True, ) - if result.returncode != 0: - error_msg = f"Failed to set environment override: {result.stderr}" - return self._error_response(FlatpakOverrideResponse, error_msg, - app_id=app_id, operation="set") - - self.log.info(f"Successfully set lsfg-vk overrides for {app_id}") - return self._success_response(FlatpakOverrideResponse, - f"lsfg-vk overrides set for {app_id}", - app_id=app_id, operation="set") - - except Exception as e: - error_msg = f"Error setting overrides for {app_id}: {str(e)}" - self.log.error(error_msg) - return self._error_response(FlatpakOverrideResponse, error_msg, - app_id=app_id, operation="set") - - def remove_app_override(self, app_id: str) -> FlatpakOverrideResponse: - """Remove lsfg-vk overrides for a Flatpak app""" - try: - if not self.check_flatpak_available(): - return self._error_response(FlatpakOverrideResponse, - "Flatpak is not available on this system", - app_id=app_id, operation="remove") - - home_path = os.path.expanduser("~") - config_path = f"{home_path}/.config/lsfg-vk" - dll_path = f"{home_path}/.local/share/Steam/steamapps/common/Lossless Scaling/Lossless.dll" - lsfg_path = f"{home_path}/lsfg" - - reset_result = self._run_flatpak_command( - ["override", "--user", "--reset", app_id], - capture_output=True, text=True + raise OSError(result.stderr.strip() or "Failed to remove Flatpak overrides") + return self._success_response( + BaseResponse, + f"lsfg-vk overrides removed for {app_id}", + app_id=app_id, + operation="remove", + ) + except Exception as error: + return self._error_response( + BaseResponse, + str(error), + app_id=app_id, + operation="remove", ) - - if reset_result.returncode == 0: - self.log.info(f"Successfully reset all overrides for {app_id}") - return self._success_response(FlatpakOverrideResponse, - f"All overrides reset for {app_id}", - app_id=app_id, operation="remove") - - self.log.debug(f"Reset failed, trying individual removal: {reset_result.stderr}") - - filesystem_overrides = [ - f"--nofilesystem={dll_path}", - f"--nofilesystem={config_path}", - f"--nofilesystem={lsfg_path}" - ] - - removal_errors = [] - - # Remove filesystem overrides - for override in filesystem_overrides: - result = self._run_flatpak_command( - ["override", "--user", override, app_id], - capture_output=True, text=True - ) - if result.returncode != 0: - removal_errors.append(f"{override}: {result.stderr}") + def migrate_v2(self) -> None: + if not self.check_flatpak_available(): + return + + status = self.get_extension_status() + for version, key in ( + ("24.08", "installed_24_08"), + ("25.08", "installed_25_08"), + ): + if not status.get(key): + continue result = self._run_flatpak_command( - ["override", "--user", "--unset-env=LSFG_CONFIG", app_id], - capture_output=True, text=True + ["update", "--user", "--noninteractive", self._extension_ref(version)], + capture_output=True, + text=True, ) - if result.returncode != 0: - removal_errors.append(f"unset-env: {result.stderr}") - - if removal_errors: - self.log.warning(f"Some override removals had issues for {app_id}: {'; '.join(removal_errors)}") - - self.log.info(f"Completed override removal for {app_id}") - return self._success_response(FlatpakOverrideResponse, - f"lsfg-vk overrides removed for {app_id}", - app_id=app_id, operation="remove") - - except Exception as e: - error_msg = f"Error removing overrides for {app_id}: {str(e)}" - self.log.error(error_msg) - return self._error_response(FlatpakOverrideResponse, error_msg, - app_id=app_id, operation="remove") \ No newline at end of file + self.log.warning(result.stderr.strip()) + + apps_result = self._run_flatpak_command( + ["list", "--user", "--app", "--columns=application"], + capture_output=True, + text=True, + ) + if apps_result.returncode != 0: + return + + for app_id in apps_result.stdout.splitlines(): + app_id = app_id.strip() + if not app_id: + continue + if "LSFG_CONFIG=" in self._override_output(app_id): + result = self.set_app_override(app_id) + if not result.get("success"): + self.log.warning(result.get("error")) -- cgit v1.2.3 From 688c7c6e0e49deaedb3edc658dd9342453cfe4c1 Mon Sep 17 00:00:00 2001 From: Kurt Himebauch <136133082+xXJSONDeruloXx@users.noreply.github.com> Date: Sat, 5 Sep 2026 17:51:25 -0400 Subject: fix: rebase flatpak runtimes onto flathub --- py_modules/lsfg_vk/flatpak_service.py | 10 +++------- 1 file changed, 3 insertions(+), 7 deletions(-) (limited to 'py_modules/lsfg_vk/flatpak_service.py') diff --git a/py_modules/lsfg_vk/flatpak_service.py b/py_modules/lsfg_vk/flatpak_service.py index 7aa3cda..627bbd8 100644 --- a/py_modules/lsfg_vk/flatpak_service.py +++ b/py_modules/lsfg_vk/flatpak_service.py @@ -295,13 +295,9 @@ class FlatpakService(BaseService): ): if not status.get(key): continue - result = self._run_flatpak_command( - ["update", "--user", "--noninteractive", self._extension_ref(version)], - capture_output=True, - text=True, - ) - if result.returncode != 0: - self.log.warning(result.stderr.strip()) + result = self.install_extension(version) + if not result.get("success"): + self.log.warning(result.get("error")) apps_result = self._run_flatpak_command( ["list", "--user", "--app", "--columns=application"], -- cgit v1.2.3 From e8e469f99078858dc953663cba6f3428e80b1c5d Mon Sep 17 00:00:00 2001 From: xXJSONDeruloXx Date: Sat, 5 Sep 2026 23:44:37 -0400 Subject: refactor: delegate runtime checks to lsfg-vk --- py_modules/lsfg_vk/flatpak_service.py | 80 ++++++++++++++++++++++------------- 1 file changed, 51 insertions(+), 29 deletions(-) (limited to 'py_modules/lsfg_vk/flatpak_service.py') diff --git a/py_modules/lsfg_vk/flatpak_service.py b/py_modules/lsfg_vk/flatpak_service.py index 627bbd8..302efc7 100644 --- a/py_modules/lsfg_vk/flatpak_service.py +++ b/py_modules/lsfg_vk/flatpak_service.py @@ -6,13 +6,18 @@ from typing import Any, Dict, List from .base_service import BaseService from .config_schema import ConfigurationManager -from .dll_detection import DllDetectionService +from .constants import ( + BIN_DIR, + FLATPAK_23_08_FILENAME, + FLATPAK_24_08_FILENAME, + FLATPAK_25_08_FILENAME, +) from .types import BaseResponse class FlatpakService(BaseService): EXTENSION_ID = "org.freedesktop.Platform.VulkanLayer.lsfgvk" - SUPPORTED_RUNTIMES = ("24.08", "25.08") + SUPPORTED_RUNTIMES = ("23.08", "24.08", "25.08") def __init__(self, logger=None): super().__init__(logger) @@ -51,6 +56,18 @@ class FlatpakService(BaseService): if version not in cls.SUPPORTED_RUNTIMES: raise ValueError("Unsupported Flatpak runtime") + @classmethod + def _bundle_filename(cls, version: str) -> str: + return { + "23.08": FLATPAK_23_08_FILENAME, + "24.08": FLATPAK_24_08_FILENAME, + "25.08": FLATPAK_25_08_FILENAME, + }[version] + + def _bundled_extension_path(self, version: str) -> Path: + self._validate_runtime(version) + return Path(__file__).resolve().parent.parent.parent / BIN_DIR / self._bundle_filename(version) + def get_extension_status(self) -> Dict[str, Any]: try: if not self.check_flatpak_available(): @@ -70,6 +87,7 @@ class FlatpakService(BaseService): return self._success_response( BaseResponse, "Flatpak runtime status retrieved", + installed_23_08=(self.EXTENSION_ID, "x86_64", "23.08") in installed, installed_24_08=(self.EXTENSION_ID, "x86_64", "24.08") in installed, installed_25_08=(self.EXTENSION_ID, "x86_64", "25.08") in installed, ) @@ -77,6 +95,7 @@ class FlatpakService(BaseService): return self._error_response( BaseResponse, str(error), + installed_23_08=False, installed_24_08=False, installed_25_08=False, ) @@ -86,14 +105,18 @@ class FlatpakService(BaseService): self._validate_runtime(version) if not self.check_flatpak_available(): raise FileNotFoundError("Flatpak is not available on this system") + bundle_path = self._bundled_extension_path(version) + if not bundle_path.is_file(): + raise FileNotFoundError( + f"Bundled Flatpak extension not found at {bundle_path}; reinstall the plugin" + ) result = self._run_flatpak_command( [ "install", "--user", "--noninteractive", "--or-update", - "flathub", - f"{self.EXTENSION_ID}//{version}", + str(bundle_path), ], capture_output=True, text=True, @@ -102,7 +125,7 @@ class FlatpakService(BaseService): raise OSError(result.stderr.strip() or "Flatpak installation failed") return self._success_response( BaseResponse, - f"lsfg-vk {version} runtime extension installed", + f"lsfg-vk {version} runtime extension installed from the bundled asset", ) except Exception as error: return self._error_response(BaseResponse, str(error)) @@ -126,6 +149,14 @@ class FlatpakService(BaseService): except Exception as error: return self._error_response(BaseResponse, str(error)) + def _override_output(self, app_id: str) -> str: + result = self._run_flatpak_command( + ["override", "--user", "--show", app_id], + capture_output=True, + text=True, + ) + return result.stdout if result.returncode == 0 else "" + def _dll_directory(self) -> Path: if self.config_file_path.exists(): try: @@ -138,24 +169,14 @@ class FlatpakService(BaseService): except Exception: pass - result = DllDetectionService(self.log).check_lossless_scaling_dll() - if result.get("detected") and result.get("path"): - return Path(result["path"]).parent - return self.user_home / ".local/share/Steam/steamapps/common" - - def _override_output(self, app_id: str) -> str: - result = self._run_flatpak_command( - ["override", "--user", "--show", app_id], - capture_output=True, - text=True, - ) - return result.stdout if result.returncode == 0 else "" + return self.user_home / ".local/share/Steam/steamapps/common/Lossless Scaling" def _override_paths(self) -> Dict[str, str]: return { "config_dir": str(self.config_dir), "config_file": str(self.config_file_path), "dll_dir": str(self._dll_directory()), + "legacy_home": str(self.user_home), "legacy_dll": str( self.user_home / ".local/share/Steam/steamapps/common/Lossless Scaling/Lossless.dll" @@ -224,6 +245,9 @@ class FlatpakService(BaseService): f"--filesystem={paths['config_dir']}:rw", f"--filesystem={paths['dll_dir']}:ro", f"--env=LSFGVK_CONFIG={paths['config_file']}", + # Remove permissions/env from the pre-v2 plugin when an + # existing app is explicitly migrated or reconfigured. + f"--nofilesystem={paths['legacy_home']}", f"--nofilesystem={paths['legacy_dll']}", f"--nofilesystem={paths['legacy_script']}", "--unset-env=LSFG_CONFIG", @@ -259,6 +283,7 @@ class FlatpakService(BaseService): "--user", f"--nofilesystem={paths['config_dir']}", f"--nofilesystem={paths['dll_dir']}", + f"--nofilesystem={paths['legacy_home']}", f"--nofilesystem={paths['legacy_dll']}", f"--nofilesystem={paths['legacy_script']}", "--unset-env=LSFGVK_CONFIG", @@ -288,17 +313,6 @@ class FlatpakService(BaseService): if not self.check_flatpak_available(): return - status = self.get_extension_status() - for version, key in ( - ("24.08", "installed_24_08"), - ("25.08", "installed_25_08"), - ): - if not status.get(key): - continue - result = self.install_extension(version) - if not result.get("success"): - self.log.warning(result.get("error")) - apps_result = self._run_flatpak_command( ["list", "--user", "--app", "--columns=application"], capture_output=True, @@ -311,7 +325,15 @@ class FlatpakService(BaseService): app_id = app_id.strip() if not app_id: continue - if "LSFG_CONFIG=" in self._override_output(app_id): + output = self._override_output(app_id) + paths = self._override_paths() + legacy_markers = ( + "LSFG_CONFIG=", + paths["legacy_home"], + paths["legacy_dll"], + paths["legacy_script"], + ) + if any(marker in output for marker in legacy_markers): result = self.set_app_override(app_id) if not result.get("success"): self.log.warning(result.get("error")) -- cgit v1.2.3 From c9be32287ad5b72fcd86b10fa726d09dbd97d7fd Mon Sep 17 00:00:00 2001 From: xXJSONDeruloXx Date: Sun, 6 Sep 2026 17:09:54 -0400 Subject: refactor: organize plugin into native tabs --- py_modules/lsfg_vk/flatpak_service.py | 21 ++++++++++++++++++--- 1 file changed, 18 insertions(+), 3 deletions(-) (limited to 'py_modules/lsfg_vk/flatpak_service.py') diff --git a/py_modules/lsfg_vk/flatpak_service.py b/py_modules/lsfg_vk/flatpak_service.py index 302efc7..6f8a596 100644 --- a/py_modules/lsfg_vk/flatpak_service.py +++ b/py_modules/lsfg_vk/flatpak_service.py @@ -1,4 +1,5 @@ import os +import pwd import shutil import subprocess from pathlib import Path @@ -26,6 +27,7 @@ class FlatpakService(BaseService): def _get_clean_env(self) -> Dict[str, str]: env = os.environ.copy() env.pop("LD_LIBRARY_PATH", None) + env["HOME"] = str(self.user_home) path_entries = [entry for entry in env.get("PATH", "").split(":") if entry] for entry in ("/usr/bin", "/usr/local/bin", "/bin"): if entry not in path_entries: @@ -33,6 +35,12 @@ class FlatpakService(BaseService): env["PATH"] = ":".join(path_entries) return env + def _flatpak_user(self) -> pwd.struct_passwd: + try: + return 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 + def check_flatpak_available(self) -> bool: env = self._get_clean_env() self.flatpak_command = shutil.which("flatpak", path=env["PATH"]) @@ -41,8 +49,15 @@ class FlatpakService(BaseService): def _run_flatpak_command(self, args: List[str], **kwargs): if self.flatpak_command is None and not self.check_flatpak_available(): raise FileNotFoundError("Flatpak command not available") + command = [self.flatpak_command, *args] + target_user = self._flatpak_user() + if os.geteuid() != target_user.pw_uid: + runuser = shutil.which("runuser", path=self._get_clean_env()["PATH"]) + if runuser is None: + raise FileNotFoundError("runuser command not available") + command = [runuser, "--user", target_user.pw_name, "--", *command] return subprocess.run( - [self.flatpak_command, *args], + command, env=self._get_clean_env(), **kwargs, ) @@ -181,7 +196,7 @@ class FlatpakService(BaseService): self.user_home / ".local/share/Steam/steamapps/common/Lossless Scaling/Lossless.dll" ), - "legacy_script": str(self.lsfg_launch_script_path), + "legacy_script": str(self.legacy_script_path), } def _check_app_override_status(self, app_id: str) -> Dict[str, bool]: @@ -200,7 +215,7 @@ class FlatpakService(BaseService): if not self.check_flatpak_available(): raise FileNotFoundError("Flatpak is not available on this system") result = self._run_flatpak_command( - ["list", "--user", "--app", "--columns=name,application"], + ["list", "--app", "--columns=name,application"], capture_output=True, text=True, check=True, -- cgit v1.2.3 From 54acc36f16e1336772354f979a618cce65061f82 Mon Sep 17 00:00:00 2001 From: xXJSONDeruloXx Date: Mon, 7 Sep 2026 20:57:26 -0400 Subject: refactor: make runtime installation explicit --- py_modules/lsfg_vk/flatpak_service.py | 29 ----------------------------- 1 file changed, 29 deletions(-) (limited to 'py_modules/lsfg_vk/flatpak_service.py') diff --git a/py_modules/lsfg_vk/flatpak_service.py b/py_modules/lsfg_vk/flatpak_service.py index 6f8a596..6aebf11 100644 --- a/py_modules/lsfg_vk/flatpak_service.py +++ b/py_modules/lsfg_vk/flatpak_service.py @@ -323,32 +323,3 @@ class FlatpakService(BaseService): app_id=app_id, operation="remove", ) - - def migrate_v2(self) -> None: - if not self.check_flatpak_available(): - return - - apps_result = self._run_flatpak_command( - ["list", "--user", "--app", "--columns=application"], - capture_output=True, - text=True, - ) - if apps_result.returncode != 0: - return - - for app_id in apps_result.stdout.splitlines(): - app_id = app_id.strip() - if not app_id: - continue - output = self._override_output(app_id) - paths = self._override_paths() - legacy_markers = ( - "LSFG_CONFIG=", - paths["legacy_home"], - paths["legacy_dll"], - paths["legacy_script"], - ) - if any(marker in output for marker in legacy_markers): - result = self.set_app_override(app_id) - if not result.get("success"): - self.log.warning(result.get("error")) -- cgit v1.2.3 From 790132668c4421c68c32bdc8fc9792b0d6028f97 Mon Sep 17 00:00:00 2001 From: xXJSONDeruloXx Date: Tue, 8 Sep 2026 22:18:13 -0400 Subject: I really dont want to but here you go little guy --- py_modules/lsfg_vk/flatpak_service.py | 134 ++++++++++++++++++++++++++-------- 1 file changed, 105 insertions(+), 29 deletions(-) (limited to 'py_modules/lsfg_vk/flatpak_service.py') diff --git a/py_modules/lsfg_vk/flatpak_service.py b/py_modules/lsfg_vk/flatpak_service.py index 6aebf11..0e89d3c 100644 --- a/py_modules/lsfg_vk/flatpak_service.py +++ b/py_modules/lsfg_vk/flatpak_service.py @@ -1,5 +1,6 @@ import os import pwd +import re import shutil import subprocess from pathlib import Path @@ -19,6 +20,10 @@ from .types import BaseResponse class FlatpakService(BaseService): EXTENSION_ID = "org.freedesktop.Platform.VulkanLayer.lsfgvk" SUPPORTED_RUNTIMES = ("23.08", "24.08", "25.08") + COMPATIBILITY_ENV = ( + ("ENABLE_GAMESCOPE_WSI", "0"), + ("DXVK_HDR", "0"), + ) def __init__(self, logger=None): super().__init__(logger) @@ -170,7 +175,9 @@ class FlatpakService(BaseService): capture_output=True, text=True, ) - return result.stdout if result.returncode == 0 else "" + if result.returncode != 0: + raise OSError(result.stderr.strip() or "Failed to read Flatpak overrides") + return result.stdout def _dll_directory(self) -> Path: if self.config_file_path.exists(): @@ -199,15 +206,99 @@ class FlatpakService(BaseService): "legacy_script": str(self.legacy_script_path), } + def _override_file_path(self, app_id: str) -> Path: + if not app_id or Path(app_id).name != app_id or app_id in {".", ".."}: + raise ValueError("Invalid Flatpak application ID") + return self.user_home / ".local/share/flatpak/overrides" / app_id + + @staticmethod + def _filesystem_entry_path(entry: str) -> str: + value = entry.strip() + if value.startswith("!"): + value = value[1:] + for suffix in (":ro", ":rw", ":create"): + if value.endswith(suffix): + return value[: -len(suffix)] + return value + + @staticmethod + def _override_value(content: str, key: str) -> str: + match = re.search(rf"(?m)^[ \t]*{re.escape(key)}[ \t]*=([^\r\n]*)", content) + return match.group(1).strip() if match else "" + + def _clean_override_file(self, app_id: str, paths: Dict[str, str]) -> bool: + path = self._override_file_path(app_id) + if not path.is_file(): + return False + + owner = path.stat().st_uid, path.stat().st_gid + original = path.read_text(encoding="utf-8") + managed_paths = { + paths[name] + for name in ("config_dir", "dll_dir", "legacy_home", "legacy_dll", "legacy_script") + } + managed_env = { + "LSFGVK_CONFIG", + "LSFG_CONFIG", + *(name for name, _ in self.COMPATIBILITY_ENV), + } + def clean_list(match): + key, value, newline = match.groups() + is_filesystem = key.split("=", 1)[0].strip() == "filesystems" + keep = [item for item in value.split(";") if item and ( + self._filesystem_entry_path(item) not in managed_paths + if is_filesystem else item.strip() not in managed_env + )] + return f"{key}{';'.join(keep)}{newline}" if keep else "" + + updated = re.sub( + r"(?m)^([ \t]*(?:filesystems|unset-environment)[ \t]*=)([^\r\n]*)(\r?\n|$)", + clean_list, + original, + ) + env_pattern = "|".join(re.escape(name) for name in managed_env) + updated = re.sub( + rf"(?m)^[ \t]*(?:{env_pattern})[ \t]*=[^\r\n]*(?:\r?\n|$)", + "", + updated, + ) + if updated != original: + self._write_file(path, updated) + if os.geteuid() == 0: + os.chown(path, *owner) + return updated != original + def _check_app_override_status(self, app_id: str) -> Dict[str, bool]: output = self._override_output(app_id) paths = self._override_paths() + filesystem_entries = self._override_value(output, "filesystems").split(";") + positive_filesystems = { + self._filesystem_entry_path(entry) + for entry in filesystem_entries + if not entry.strip().startswith("!") + } + blocked_filesystems = { + self._filesystem_entry_path(entry) + for entry in filesystem_entries + if entry.strip().startswith("!") + } + unset_environment = set( + item.strip() + for item in self._override_value(output, "unset-environment").split(";") + if item.strip() + ) return { - "filesystem": ( - paths["config_dir"] in output - and paths["dll_dir"] in output + "filesystem": all( + path in positive_filesystems and path not in blocked_filesystems + for path in (paths["config_dir"], paths["dll_dir"]) + ), + "env": all( + self._override_value(output, name) == value and name not in unset_environment + for name, value in ( + ("LSFGVK_CONFIG", paths["config_file"]), + *self.COMPATIBILITY_ENV, + ) ), - "env": f"LSFGVK_CONFIG={paths['config_file']}" in output, } def get_flatpak_apps(self) -> Dict[str, Any]: @@ -253,6 +344,7 @@ class FlatpakService(BaseService): if not self.check_flatpak_available(): raise FileNotFoundError("Flatpak is not available on this system") paths = self._override_paths() + self._clean_override_file(app_id, paths) result = self._run_flatpak_command( [ "override", @@ -260,12 +352,7 @@ class FlatpakService(BaseService): f"--filesystem={paths['config_dir']}:rw", f"--filesystem={paths['dll_dir']}:ro", f"--env=LSFGVK_CONFIG={paths['config_file']}", - # Remove permissions/env from the pre-v2 plugin when an - # existing app is explicitly migrated or reconfigured. - f"--nofilesystem={paths['legacy_home']}", - f"--nofilesystem={paths['legacy_dll']}", - f"--nofilesystem={paths['legacy_script']}", - "--unset-env=LSFG_CONFIG", + *(f"--env={name}={value}" for name, value in self.COMPATIBILITY_ENV), app_id, ], capture_output=True, @@ -273,6 +360,9 @@ class FlatpakService(BaseService): ) if result.returncode != 0: raise OSError(result.stderr.strip() or "Failed to set Flatpak overrides") + status = self._check_app_override_status(app_id) + if not status["filesystem"] or not status["env"]: + raise RuntimeError("Flatpak overrides could not be verified after setting") return self._success_response( BaseResponse, f"lsfg-vk overrides set for {app_id}", @@ -292,24 +382,10 @@ class FlatpakService(BaseService): if not self.check_flatpak_available(): raise FileNotFoundError("Flatpak is not available on this system") paths = self._override_paths() - result = self._run_flatpak_command( - [ - "override", - "--user", - f"--nofilesystem={paths['config_dir']}", - f"--nofilesystem={paths['dll_dir']}", - f"--nofilesystem={paths['legacy_home']}", - f"--nofilesystem={paths['legacy_dll']}", - f"--nofilesystem={paths['legacy_script']}", - "--unset-env=LSFGVK_CONFIG", - "--unset-env=LSFG_CONFIG", - app_id, - ], - capture_output=True, - text=True, - ) - if result.returncode != 0: - raise OSError(result.stderr.strip() or "Failed to remove Flatpak overrides") + self._clean_override_file(app_id, paths) + status = self._check_app_override_status(app_id) + if status["filesystem"] or status["env"]: + raise RuntimeError("Flatpak overrides could not be verified after removal") return self._success_response( BaseResponse, f"lsfg-vk overrides removed for {app_id}", -- cgit v1.2.3 From cc1e6f47dd9838b066822162a607d2859c043aff Mon Sep 17 00:00:00 2001 From: xXJSONDeruloXx Date: Wed, 9 Sep 2026 19:16:30 -0400 Subject: refactor: unify flatpak targets with steam profiles --- py_modules/lsfg_vk/flatpak_service.py | 724 ++++++++++++++++++++-------------- 1 file changed, 437 insertions(+), 287 deletions(-) (limited to 'py_modules/lsfg_vk/flatpak_service.py') diff --git a/py_modules/lsfg_vk/flatpak_service.py b/py_modules/lsfg_vk/flatpak_service.py index 0e89d3c..071b29c 100644 --- a/py_modules/lsfg_vk/flatpak_service.py +++ b/py_modules/lsfg_vk/flatpak_service.py @@ -1,33 +1,52 @@ +"""Flatpak runtime-extension infrastructure for unified game 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 Any, Dict, List +from typing import Any, Dict, List, Optional, Set, Tuple from .base_service import BaseService -from .config_schema import ConfigurationManager from .constants import ( BIN_DIR, FLATPAK_23_08_FILENAME, FLATPAK_24_08_FILENAME, FLATPAK_25_08_FILENAME, ) -from .types import BaseResponse class FlatpakService(BaseService): + """Resolve and provision only the runtime support a target actually needs. + + Flatpak application permissions are deliberately not persisted here. The + generated per-AppID wrapper supplies the narrow launch-time permissions and + environment instead, while this service owns only the shared Vulkan layer + runtime extensions installed from the plugin bundle. + """ + EXTENSION_ID = "org.freedesktop.Platform.VulkanLayer.lsfgvk" SUPPORTED_RUNTIMES = ("23.08", "24.08", "25.08") - COMPATIBILITY_ENV = ( - ("ENABLE_GAMESCOPE_WSI", "0"), - ("DXVK_HDR", "0"), + 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-]*)+$" ) + BRANCH_PATTERN = re.compile(r"^[0-9]+\.[0-9]+$") def __init__(self, logger=None): super().__init__(logger) - self.flatpak_command = None + self.flatpak_command: Optional[str] = None + self._lock = threading.RLock() + + @property + def ownership_path(self) -> Path: + return self.config_dir / self.OWNERSHIP_FILENAME def _get_clean_env(self) -> Dict[str, str]: env = os.environ.copy() @@ -61,20 +80,39 @@ class FlatpakService(BaseService): if runuser is None: raise FileNotFoundError("runuser command not available") command = [runuser, "--user", target_user.pw_name, "--", *command] - return subprocess.run( - command, - env=self._get_clean_env(), - **kwargs, - ) + return subprocess.run(command, env=self._get_clean_env(), **kwargs) @classmethod - def _extension_ref(cls, version: str) -> str: - return f"{cls.EXTENSION_ID}/x86_64/{version}" + 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, version: str) -> None: + def _validate_runtime(cls, version: str) -> str: if version not in cls.SUPPORTED_RUNTIMES: - raise ValueError("Unsupported Flatpak runtime") + raise ValueError( + f"Unsupported Flatpak runtime branch {version}; " + f"supported branches are {', '.join(cls.SUPPORTED_RUNTIMES)}" + ) + return version + + @classmethod + def _extension_ref(cls, version: str) -> str: + return f"{cls.EXTENSION_ID}/x86_64/{cls._validate_runtime(version)}" + + @classmethod + def runtime_branch_from_ref(cls, runtime_ref: str) -> str: + """Return the supported Freedesktop branch from a runtime ref.""" + if not isinstance(runtime_ref, str): + raise ValueError("Flatpak did not return a runtime reference") + parts = runtime_ref.strip().split("/") + if len(parts) != 3 or parts[0] != "org.freedesktop.Platform": + raise ValueError(f"Unsupported Flatpak runtime reference: {runtime_ref}") + branch = parts[2] + if not cls.BRANCH_PATTERN.fullmatch(branch): + raise ValueError(f"Unrecognized Flatpak runtime branch: {branch}") + return cls._validate_runtime(branch) @classmethod def _bundle_filename(cls, version: str) -> str: @@ -82,320 +120,432 @@ class FlatpakService(BaseService): "23.08": FLATPAK_23_08_FILENAME, "24.08": FLATPAK_24_08_FILENAME, "25.08": FLATPAK_25_08_FILENAME, - }[version] + }[cls._validate_runtime(version)] def _bundled_extension_path(self, version: str) -> Path: - self._validate_runtime(version) - return Path(__file__).resolve().parent.parent.parent / BIN_DIR / self._bundle_filename(version) + return ( + Path(__file__).resolve().parent.parent.parent + / BIN_DIR + / self._bundle_filename(version) + ) - def get_extension_status(self) -> Dict[str, Any]: - try: - if not self.check_flatpak_available(): - raise FileNotFoundError("Flatpak is not available on this system") + def _installed_extension_branches(self) -> Set[str]: + result = self._run_flatpak_command( + ["list", "--runtime", "--columns=application,arch,branch"], + capture_output=True, + text=True, + check=True, + ) + installed: Set[str] = set() + for line in result.stdout.splitlines(): + if not line.strip(): + continue + fields = line.split("\t") + if len(fields) < 3: + fields = line.split() + if len(fields) < 3: + continue + application, arch, branch = (field.strip() for field in fields[:3]) + if application == self.EXTENSION_ID and arch == "x86_64": + installed.add(branch) + return installed - result = self._run_flatpak_command( - ["list", "--user", "--runtime", "--columns=application,arch,branch"], - capture_output=True, - text=True, - check=True, - ) - installed = { - tuple(line.split("\t")[:3]) - for line in result.stdout.splitlines() - if line.strip() + def _read_owned_branches(self) -> Tuple[Set[str], bool]: + """Read ownership without guessing when metadata is damaged.""" + path = self.ownership_path + if path.is_symlink(): + self.log.warning(f"Flatpak ownership metadata is not a regular file: {path}") + return set(), True + if not path.exists(): + return set(), False + if not path.is_file(): + self.log.warning(f"Flatpak ownership metadata is not a regular file: {path}") + return set(), True + try: + raw = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(raw, dict) or raw.get("version") != self.OWNERSHIP_VERSION: + raise ValueError("unsupported ownership metadata version") + branches = raw.get("plugin_owned_branches") + if not isinstance(branches, list): + raise ValueError("plugin_owned_branches is not a list") + normalized = { + self._validate_runtime(branch) + for branch in branches + if isinstance(branch, str) } - return self._success_response( - BaseResponse, - "Flatpak runtime status retrieved", - installed_23_08=(self.EXTENSION_ID, "x86_64", "23.08") in installed, - installed_24_08=(self.EXTENSION_ID, "x86_64", "24.08") in installed, - installed_25_08=(self.EXTENSION_ID, "x86_64", "25.08") in installed, - ) - except Exception as error: - return self._error_response( - BaseResponse, - str(error), - installed_23_08=False, - installed_24_08=False, - installed_25_08=False, - ) + if len(normalized) != len(branches): + raise ValueError("ownership metadata contains invalid branches") + return normalized, False + except (OSError, json.JSONDecodeError, TypeError, ValueError) as error: + self.log.warning(f"Could not trust Flatpak ownership metadata: {error}") + return set(), True - def install_extension(self, version: str) -> Dict[str, Any]: + def _write_owned_branches(self, branches: Set[str]) -> None: + if not branches: + if self.ownership_path.exists() or self.ownership_path.is_symlink(): + self.ownership_path.unlink() + return + document = { + "version": self.OWNERSHIP_VERSION, + "plugin_owned_branches": sorted(branches), + } + self._write_file(self.ownership_path, json.dumps(document, indent=2) + "\n") + + def get_extension_status(self) -> Dict[str, Any]: + """Return global extension inventory for Setup diagnostics.""" try: - self._validate_runtime(version) if not self.check_flatpak_available(): - raise FileNotFoundError("Flatpak is not available on this system") - bundle_path = self._bundled_extension_path(version) - if not bundle_path.is_file(): - raise FileNotFoundError( - f"Bundled Flatpak extension not found at {bundle_path}; reinstall the plugin" + return self._success_response( + dict, + "Flatpak is not available", + available=False, + extension_id=self.EXTENSION_ID, + supported_branches=list(self.SUPPORTED_RUNTIMES), + installed_branches=[], + owned_branches=[], + ownership_uncertain=False, ) - result = self._run_flatpak_command( - [ - "install", - "--user", - "--noninteractive", - "--or-update", - str(bundle_path), - ], - capture_output=True, - text=True, - ) - if result.returncode != 0: - raise OSError(result.stderr.strip() or "Flatpak installation failed") + installed = self._installed_extension_branches() + owned, uncertain = self._read_owned_branches() return self._success_response( - BaseResponse, - f"lsfg-vk {version} runtime extension installed from the bundled asset", + dict, + "Flatpak runtime extension status retrieved", + available=True, + extension_id=self.EXTENSION_ID, + supported_branches=list(self.SUPPORTED_RUNTIMES), + installed_branches=sorted(installed), + owned_branches=sorted(owned), + ownership_uncertain=uncertain, ) except Exception as error: - return self._error_response(BaseResponse, str(error)) - - def uninstall_extension(self, version: str) -> Dict[str, Any]: - try: - self._validate_runtime(version) - if not self.check_flatpak_available(): - raise FileNotFoundError("Flatpak is not available on this system") - result = self._run_flatpak_command( - ["uninstall", "--user", "--noninteractive", self._extension_ref(version)], - capture_output=True, - text=True, - ) - if result.returncode != 0: - raise OSError(result.stderr.strip() or "Flatpak uninstall failed") - return self._success_response( - BaseResponse, - f"lsfg-vk {version} runtime extension uninstalled", + return self._error_response( + dict, + str(error), + available=self.check_flatpak_available(), + extension_id=self.EXTENSION_ID, + supported_branches=list(self.SUPPORTED_RUNTIMES), + installed_branches=[], + owned_branches=[], + ownership_uncertain=False, ) - except Exception as error: - return self._error_response(BaseResponse, str(error)) - def _override_output(self, app_id: str) -> str: + def get_flatpak_support_status(self) -> Dict[str, Any]: + return self.get_extension_status() + + def _resolve_runtime(self, app_id: str) -> Dict[str, Any]: + 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( - ["override", "--user", "--show", app_id], + ["info", "--show-runtime", app_id], capture_output=True, text=True, ) if result.returncode != 0: - raise OSError(result.stderr.strip() or "Failed to read Flatpak overrides") - return result.stdout - - def _dll_directory(self) -> Path: - if self.config_file_path.exists(): - try: - profile_data = ConfigurationManager.parse_toml_content_multi_profile( - self.config_file_path.read_text(encoding="utf-8") - ) - dll_path = profile_data["global_config"].get("dll") - if dll_path: - return Path(dll_path).parent - except Exception: - pass - - return self.user_home / ".local/share/Steam/steamapps/common/Lossless Scaling" + raise OSError(result.stderr.strip() or f"Could not inspect Flatpak app {app_id}") + runtime_ref = result.stdout.strip().splitlines()[0] if result.stdout.strip() else "" + branch = self.runtime_branch_from_ref(runtime_ref) + return {"runtime": runtime_ref, "runtime_branch": branch} - def _override_paths(self) -> Dict[str, str]: - return { - "config_dir": str(self.config_dir), - "config_file": str(self.config_file_path), - "dll_dir": str(self._dll_directory()), - "legacy_home": str(self.user_home), - "legacy_dll": str( - self.user_home - / ".local/share/Steam/steamapps/common/Lossless Scaling/Lossless.dll" - ), - "legacy_script": str(self.legacy_script_path), - } - - def _override_file_path(self, app_id: str) -> Path: - if not app_id or Path(app_id).name != app_id or app_id in {".", ".."}: - raise ValueError("Invalid Flatpak application ID") - return self.user_home / ".local/share/flatpak/overrides" / app_id - - @staticmethod - def _filesystem_entry_path(entry: str) -> str: - value = entry.strip() - if value.startswith("!"): - value = value[1:] - for suffix in (":ro", ":rw", ":create"): - if value.endswith(suffix): - return value[: -len(suffix)] - return value - - @staticmethod - def _override_value(content: str, key: str) -> str: - match = re.search(rf"(?m)^[ \t]*{re.escape(key)}[ \t]*=([^\r\n]*)", content) - return match.group(1).strip() if match else "" - - def _clean_override_file(self, app_id: str, paths: Dict[str, str]) -> bool: - path = self._override_file_path(app_id) - if not path.is_file(): - return False - - owner = path.stat().st_uid, path.stat().st_gid - original = path.read_text(encoding="utf-8") - managed_paths = { - paths[name] - for name in ("config_dir", "dll_dir", "legacy_home", "legacy_dll", "legacy_script") - } - managed_env = { - "LSFGVK_CONFIG", - "LSFG_CONFIG", - *(name for name, _ in self.COMPATIBILITY_ENV), - } - def clean_list(match): - key, value, newline = match.groups() - is_filesystem = key.split("=", 1)[0].strip() == "filesystems" - keep = [item for item in value.split(";") if item and ( - self._filesystem_entry_path(item) not in managed_paths - if is_filesystem else item.strip() not in managed_env - )] - return f"{key}{';'.join(keep)}{newline}" if keep else "" - - updated = re.sub( - r"(?m)^([ \t]*(?:filesystems|unset-environment)[ \t]*=)([^\r\n]*)(\r?\n|$)", - clean_list, - original, - ) - env_pattern = "|".join(re.escape(name) for name in managed_env) - updated = re.sub( - rf"(?m)^[ \t]*(?:{env_pattern})[ \t]*=[^\r\n]*(?:\r?\n|$)", - "", - updated, - ) - if updated != original: - self._write_file(path, updated) - if os.geteuid() == 0: - os.chown(path, *owner) - return updated != original - - def _check_app_override_status(self, app_id: str) -> Dict[str, bool]: - output = self._override_output(app_id) - paths = self._override_paths() - filesystem_entries = self._override_value(output, "filesystems").split(";") - positive_filesystems = { - self._filesystem_entry_path(entry) - for entry in filesystem_entries - if not entry.strip().startswith("!") - } - blocked_filesystems = { - self._filesystem_entry_path(entry) - for entry in filesystem_entries - if entry.strip().startswith("!") - } - unset_environment = set( - item.strip() - for item in self._override_value(output, "unset-environment").split(";") - if item.strip() - ) - return { - "filesystem": all( - path in positive_filesystems and path not in blocked_filesystems - for path in (paths["config_dir"], paths["dll_dir"]) - ), - "env": all( - self._override_value(output, name) == value and name not in unset_environment - for name, value in ( - ("LSFGVK_CONFIG", paths["config_file"]), - *self.COMPATIBILITY_ENV, - ) - ), - } - - def get_flatpak_apps(self) -> Dict[str, Any]: + def resolve_app_support(self, app_id: str) -> Dict[str, Any]: + """Resolve the exact runtime branch required by one Flatpak app.""" try: - if not self.check_flatpak_available(): - raise FileNotFoundError("Flatpak is not available on this system") - result = self._run_flatpak_command( - ["list", "--app", "--columns=name,application"], - capture_output=True, - text=True, - check=True, + app_id = self._validate_app_id(app_id) + resolved = self._resolve_runtime(app_id) + installed = self._installed_extension_branches() + branch = resolved["runtime_branch"] + 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=resolved["runtime"], + runtime_branch=branch, + support_status="ready" if ready else "needs-runtime", + extension_installed=ready, + installed_branches=sorted(installed), ) - apps = [] - for line in result.stdout.splitlines(): - parts = line.split("\t", 1) - if len(parts) != 2: - continue - status = self._check_app_override_status(parts[1]) - apps.append( - { - "app_id": parts[1], - "app_name": parts[0], - "has_filesystem_override": status["filesystem"], - "has_env_override": status["env"], - } - ) + except ValueError as error: return self._success_response( - BaseResponse, - f"Found {len(apps)} Flatpak applications", - apps=apps, - total_apps=len(apps), + 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( - BaseResponse, + dict, str(error), - apps=[], - total_apps=0, + flatpak_app_id=app_id, + runtime=None, + runtime_branch=None, + support_status="error", + extension_installed=False, + installed_branches=[], ) - def set_app_override(self, app_id: str) -> Dict[str, Any]: + def install_extension(self, version: str) -> Dict[str, Any]: + """Install one missing branch and record ownership only after readback.""" try: + version = self._validate_runtime(version) if not self.check_flatpak_available(): raise FileNotFoundError("Flatpak is not available on this system") - paths = self._override_paths() - self._clean_override_file(app_id, paths) - result = self._run_flatpak_command( - [ - "override", - "--user", - f"--filesystem={paths['config_dir']}:rw", - f"--filesystem={paths['dll_dir']}:ro", - f"--env=LSFGVK_CONFIG={paths['config_file']}", - *(f"--env={name}={value}" for name, value in self.COMPATIBILITY_ENV), - app_id, - ], - capture_output=True, - text=True, + bundle_path = self._bundled_extension_path(version) + if not bundle_path.is_file(): + raise FileNotFoundError( + f"Bundled Flatpak extension not found at {bundle_path}; reinstall the plugin" + ) + with self._lock: + owned, uncertain = self._read_owned_branches() + if uncertain: + raise RuntimeError( + "Flatpak ownership metadata is uncertain; refusing to install " + "until it is repaired" + ) + installed_before = self._installed_extension_branches() + if version in installed_before: + return self._success_response( + dict, + f"lsfg-vk {version} runtime extension is already installed", + runtime_branch=version, + installed=True, + owned_by_plugin=version in owned, + ) + result = self._run_flatpak_command( + [ + "install", + "--user", + "--noninteractive", + "--or-update", + str(bundle_path), + ], + capture_output=True, + text=True, + ) + if result.returncode != 0: + raise OSError(result.stderr.strip() or "Flatpak installation failed") + installed_after = self._installed_extension_branches() + if version not in installed_after: + raise RuntimeError( + f"Flatpak install completed but {self._extension_ref(version)} " + "was not visible afterwards" + ) + owned.add(version) + self._write_owned_branches(owned) + return self._success_response( + dict, + f"lsfg-vk {version} runtime extension installed", + runtime_branch=version, + installed=True, + owned_by_plugin=True, + ) + except Exception as error: + return self._error_response( + dict, + str(error), + runtime_branch=version, + installed=False, + owned_by_plugin=False, ) - if result.returncode != 0: - raise OSError(result.stderr.strip() or "Failed to set Flatpak overrides") - status = self._check_app_override_status(app_id) - if not status["filesystem"] or not status["env"]: - raise RuntimeError("Flatpak overrides could not be verified after setting") + + def ensure_extension(self, version: str) -> Dict[str, Any]: + status = self.get_extension_status() + if not status.get("success"): + return status + if not status.get("available"): + return self._error_response( + dict, + "Flatpak is not available on this system", + runtime_branch=version, + support_status="error", + ) + try: + version = self._validate_runtime(version) + except ValueError as error: + return self._error_response(dict, str(error), runtime_branch=version, support_status="unsupported") + if version in status.get("installed_branches", []): return self._success_response( - BaseResponse, - f"lsfg-vk overrides set for {app_id}", - app_id=app_id, - operation="set", + dict, + f"lsfg-vk {version} runtime extension is ready", + runtime_branch=version, + installed=True, + owned_by_plugin=version in status.get("owned_branches", []), ) - except Exception as error: + return self.install_extension(version) + + def ensure_app_support(self, app_id: str) -> Dict[str, Any]: + """Provision only the branch returned by flatpak info for this app.""" + resolved = self.resolve_app_support(app_id) + if not resolved.get("success") or resolved.get("support_status") != "needs-runtime": + return resolved + branch = resolved.get("runtime_branch") + result = self.ensure_extension(branch) + if not result.get("success"): return self._error_response( - BaseResponse, - str(error), - app_id=app_id, - operation="set", + dict, + result.get("error") or "Could not install the required Flatpak runtime extension", + flatpak_app_id=app_id, + runtime=resolved.get("runtime"), + runtime_branch=branch, + support_status="error", + extension_installed=False, ) + final = self.resolve_app_support(app_id) + if final.get("success") and final.get("support_status") == "ready": + return final + return self._error_response( + dict, + final.get("error") or "Required Flatpak runtime extension could not be verified", + flatpak_app_id=app_id, + runtime=resolved.get("runtime"), + runtime_branch=branch, + support_status="error", + extension_installed=False, + ) - def remove_app_override(self, app_id: str) -> Dict[str, Any]: + def uninstall_extension(self, version: str) -> Dict[str, Any]: + """Uninstall only when explicitly requested for a plugin-owned branch.""" try: + version = self._validate_runtime(version) if not self.check_flatpak_available(): raise FileNotFoundError("Flatpak is not available on this system") - paths = self._override_paths() - self._clean_override_file(app_id, paths) - status = self._check_app_override_status(app_id) - if status["filesystem"] or status["env"]: - raise RuntimeError("Flatpak overrides could not be verified after removal") - return self._success_response( - BaseResponse, - f"lsfg-vk overrides removed for {app_id}", - app_id=app_id, - operation="remove", + with self._lock: + owned, uncertain = self._read_owned_branches() + if uncertain: + raise RuntimeError( + "Flatpak ownership metadata is uncertain; refusing to uninstall" + ) + if version not in owned: + return self._success_response( + dict, + f"Preserved Flatpak extension {version}; it is not plugin-owned", + runtime_branch=version, + removed=False, + preserved=True, + ) + installed = self._installed_extension_branches() + if version in installed: + result = self._run_flatpak_command( + [ + "uninstall", + "--user", + "--noninteractive", + self._extension_ref(version), + ], + capture_output=True, + text=True, + ) + if result.returncode != 0: + raise OSError(result.stderr.strip() or "Flatpak uninstall failed") + if version in self._installed_extension_branches(): + raise RuntimeError( + f"Flatpak uninstall completed but {self._extension_ref(version)} " + "is still installed" + ) + owned.remove(version) + self._write_owned_branches(owned) + return self._success_response( + dict, + f"Plugin-owned lsfg-vk {version} runtime extension removed", + runtime_branch=version, + removed=True, + preserved=False, + ) + except Exception as error: + return self._error_response( + dict, + str(error), + runtime_branch=version, + removed=False, + preserved=False, ) + + def remove_plugin_owned_extensions(self) -> Dict[str, Any]: + """Uninstall only branches recorded as installed by this plugin.""" + try: + with self._lock: + owned, uncertain = self._read_owned_branches() + if uncertain: + return self._error_response( + dict, + "Flatpak ownership metadata is uncertain; no extensions were removed", + removed_branches=[], + preserved_branches=[], + ownership_uncertain=True, + ) + 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(): + return self._error_response( + dict, + "Flatpak is not available; plugin-owned extension metadata was preserved", + removed_branches=[], + preserved_branches=sorted(owned), + ownership_uncertain=False, + ) + removed: List[str] = [] + failures: List[str] = [] + for branch in sorted(owned): + try: + installed = self._installed_extension_branches() + if branch in installed: + 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(): + raise RuntimeError( + f"Flatpak uninstall completed but {self._extension_ref(branch)} " + "is still installed" + ) + 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( - BaseResponse, + dict, str(error), - app_id=app_id, - operation="remove", + removed_branches=[], + preserved_branches=[], + ownership_uncertain=False, ) -- cgit v1.2.3 From fb4d053213bdbda271a54b517a11a89c4780f80a Mon Sep 17 00:00:00 2001 From: xXJSONDeruloXx Date: Wed, 9 Sep 2026 20:45:20 -0400 Subject: fix: restore profile and Flatpak controls --- py_modules/lsfg_vk/flatpak_service.py | 56 ++++++++++++++++++++++++++++++----- 1 file changed, 48 insertions(+), 8 deletions(-) (limited to 'py_modules/lsfg_vk/flatpak_service.py') diff --git a/py_modules/lsfg_vk/flatpak_service.py b/py_modules/lsfg_vk/flatpak_service.py index 071b29c..c0a9c0c 100644 --- a/py_modules/lsfg_vk/flatpak_service.py +++ b/py_modules/lsfg_vk/flatpak_service.py @@ -299,11 +299,6 @@ class FlatpakService(BaseService): version = self._validate_runtime(version) if not self.check_flatpak_available(): raise FileNotFoundError("Flatpak is not available on this system") - bundle_path = self._bundled_extension_path(version) - if not bundle_path.is_file(): - raise FileNotFoundError( - f"Bundled Flatpak extension not found at {bundle_path}; reinstall the plugin" - ) with self._lock: owned, uncertain = self._read_owned_branches() if uncertain: @@ -318,7 +313,14 @@ class FlatpakService(BaseService): f"lsfg-vk {version} runtime extension is already installed", runtime_branch=version, installed=True, + enabled=True, owned_by_plugin=version in owned, + preserved=version not in owned, + ) + bundle_path = self._bundled_extension_path(version) + if not bundle_path.is_file(): + raise FileNotFoundError( + f"Bundled Flatpak extension not found at {bundle_path}; reinstall the plugin" ) result = self._run_flatpak_command( [ @@ -346,7 +348,9 @@ class FlatpakService(BaseService): f"lsfg-vk {version} runtime extension installed", runtime_branch=version, installed=True, + enabled=True, owned_by_plugin=True, + preserved=False, ) except Exception as error: return self._error_response( @@ -354,7 +358,9 @@ class FlatpakService(BaseService): str(error), runtime_branch=version, installed=False, + enabled=False, owned_by_plugin=False, + preserved=False, ) def ensure_extension(self, version: str) -> Dict[str, Any]: @@ -424,15 +430,29 @@ class FlatpakService(BaseService): raise RuntimeError( "Flatpak ownership metadata is uncertain; refusing to uninstall" ) + installed = self._installed_extension_branches() if version not in owned: + if version in installed: + return self._success_response( + dict, + f"Preserved Flatpak extension {version}; it is not plugin-owned", + runtime_branch=version, + removed=False, + installed=True, + enabled=True, + owned_by_plugin=False, + preserved=True, + ) return self._success_response( dict, - f"Preserved Flatpak extension {version}; it is not plugin-owned", + f"Flatpak extension {version} is already not installed", runtime_branch=version, removed=False, - preserved=True, + installed=False, + enabled=False, + owned_by_plugin=False, + preserved=False, ) - installed = self._installed_extension_branches() if version in installed: result = self._run_flatpak_command( [ @@ -458,6 +478,9 @@ class FlatpakService(BaseService): f"Plugin-owned lsfg-vk {version} runtime extension removed", runtime_branch=version, removed=True, + installed=False, + enabled=False, + owned_by_plugin=False, preserved=False, ) except Exception as error: @@ -466,8 +489,25 @@ class FlatpakService(BaseService): str(error), runtime_branch=version, removed=False, + installed=False, + enabled=False, + owned_by_plugin=False, + preserved=False, + ) + + def set_extension_enabled(self, version: str, enabled: bool) -> Dict[str, Any]: + """Set one runtime branch to the requested state, safely and idempotently.""" + if type(enabled) is not bool: + return self._error_response( + dict, + "enabled must be a boolean", + runtime_branch=version, + installed=False, + enabled=False, + owned_by_plugin=False, preserved=False, ) + return self.install_extension(version) if enabled else self.uninstall_extension(version) def remove_plugin_owned_extensions(self) -> Dict[str, Any]: """Uninstall only branches recorded as installed by this plugin.""" -- cgit v1.2.3 From bc75bb93d5aa9aa176262a138f47d3a1d53afbcb Mon Sep 17 00:00:00 2001 From: xXJSONDeruloXx Date: Thu, 10 Sep 2026 00:25:58 -0400 Subject: fix: simplify Flatpak runtime toggles --- py_modules/lsfg_vk/flatpak_service.py | 76 +++++++---------------------------- 1 file changed, 14 insertions(+), 62 deletions(-) (limited to 'py_modules/lsfg_vk/flatpak_service.py') diff --git a/py_modules/lsfg_vk/flatpak_service.py b/py_modules/lsfg_vk/flatpak_service.py index c0a9c0c..11f1a82 100644 --- a/py_modules/lsfg_vk/flatpak_service.py +++ b/py_modules/lsfg_vk/flatpak_service.py @@ -202,11 +202,8 @@ class FlatpakService(BaseService): extension_id=self.EXTENSION_ID, supported_branches=list(self.SUPPORTED_RUNTIMES), installed_branches=[], - owned_branches=[], - ownership_uncertain=False, ) installed = self._installed_extension_branches() - owned, uncertain = self._read_owned_branches() return self._success_response( dict, "Flatpak runtime extension status retrieved", @@ -214,8 +211,6 @@ class FlatpakService(BaseService): extension_id=self.EXTENSION_ID, supported_branches=list(self.SUPPORTED_RUNTIMES), installed_branches=sorted(installed), - owned_branches=sorted(owned), - ownership_uncertain=uncertain, ) except Exception as error: return self._error_response( @@ -225,8 +220,6 @@ class FlatpakService(BaseService): extension_id=self.EXTENSION_ID, supported_branches=list(self.SUPPORTED_RUNTIMES), installed_branches=[], - owned_branches=[], - ownership_uncertain=False, ) def get_flatpak_support_status(self) -> Dict[str, Any]: @@ -294,18 +287,12 @@ class FlatpakService(BaseService): ) def install_extension(self, version: str) -> Dict[str, Any]: - """Install one missing branch and record ownership only after readback.""" + """Install one branch, treating an already-installed branch as success.""" try: version = self._validate_runtime(version) if not self.check_flatpak_available(): raise FileNotFoundError("Flatpak is not available on this system") with self._lock: - owned, uncertain = self._read_owned_branches() - if uncertain: - raise RuntimeError( - "Flatpak ownership metadata is uncertain; refusing to install " - "until it is repaired" - ) installed_before = self._installed_extension_branches() if version in installed_before: return self._success_response( @@ -314,8 +301,6 @@ class FlatpakService(BaseService): runtime_branch=version, installed=True, enabled=True, - owned_by_plugin=version in owned, - preserved=version not in owned, ) bundle_path = self._bundled_extension_path(version) if not bundle_path.is_file(): @@ -341,16 +326,16 @@ class FlatpakService(BaseService): f"Flatpak install completed but {self._extension_ref(version)} " "was not visible afterwards" ) - owned.add(version) - self._write_owned_branches(owned) + owned, uncertain = self._read_owned_branches() + if not uncertain: + owned.add(version) + self._write_owned_branches(owned) return self._success_response( dict, f"lsfg-vk {version} runtime extension installed", runtime_branch=version, installed=True, enabled=True, - owned_by_plugin=True, - preserved=False, ) except Exception as error: return self._error_response( @@ -359,8 +344,6 @@ class FlatpakService(BaseService): runtime_branch=version, installed=False, enabled=False, - owned_by_plugin=False, - preserved=False, ) def ensure_extension(self, version: str) -> Dict[str, Any]: @@ -384,7 +367,6 @@ class FlatpakService(BaseService): f"lsfg-vk {version} runtime extension is ready", runtime_branch=version, installed=True, - owned_by_plugin=version in status.get("owned_branches", []), ) return self.install_extension(version) @@ -419,41 +401,15 @@ class FlatpakService(BaseService): ) def uninstall_extension(self, version: str) -> Dict[str, Any]: - """Uninstall only when explicitly requested for a plugin-owned branch.""" + """Uninstall one branch, treating an already-absent branch as success.""" try: version = self._validate_runtime(version) if not self.check_flatpak_available(): raise FileNotFoundError("Flatpak is not available on this system") with self._lock: - owned, uncertain = self._read_owned_branches() - if uncertain: - raise RuntimeError( - "Flatpak ownership metadata is uncertain; refusing to uninstall" - ) installed = self._installed_extension_branches() - if version not in owned: - if version in installed: - return self._success_response( - dict, - f"Preserved Flatpak extension {version}; it is not plugin-owned", - runtime_branch=version, - removed=False, - installed=True, - enabled=True, - owned_by_plugin=False, - preserved=True, - ) - return self._success_response( - dict, - f"Flatpak extension {version} is already not installed", - runtime_branch=version, - removed=False, - installed=False, - enabled=False, - owned_by_plugin=False, - preserved=False, - ) - if version in installed: + was_installed = version in installed + if was_installed: result = self._run_flatpak_command( [ "uninstall", @@ -471,17 +427,17 @@ class FlatpakService(BaseService): f"Flatpak uninstall completed but {self._extension_ref(version)} " "is still installed" ) - owned.remove(version) - self._write_owned_branches(owned) + owned, uncertain = self._read_owned_branches() + if not uncertain and version in owned: + owned.remove(version) + self._write_owned_branches(owned) return self._success_response( dict, - f"Plugin-owned lsfg-vk {version} runtime extension removed", + f"lsfg-vk {version} runtime extension uninstalled", runtime_branch=version, - removed=True, + removed=was_installed, installed=False, enabled=False, - owned_by_plugin=False, - preserved=False, ) except Exception as error: return self._error_response( @@ -491,8 +447,6 @@ class FlatpakService(BaseService): removed=False, installed=False, enabled=False, - owned_by_plugin=False, - preserved=False, ) def set_extension_enabled(self, version: str, enabled: bool) -> Dict[str, Any]: @@ -504,8 +458,6 @@ class FlatpakService(BaseService): runtime_branch=version, installed=False, enabled=False, - owned_by_plugin=False, - preserved=False, ) return self.install_extension(version) if enabled else self.uninstall_extension(version) -- cgit v1.2.3 From 450d3e5e6612d467a00bb937538fded640c66ecb Mon Sep 17 00:00:00 2001 From: Kurt Himebauch <136133082+xXJSONDeruloXx@users.noreply.github.com> Date: Thu, 10 Sep 2026 07:24:05 -0400 Subject: refactor: simplify migration implementation --- py_modules/lsfg_vk/flatpak_service.py | 460 +++++++++++----------------------- 1 file changed, 150 insertions(+), 310 deletions(-) (limited to 'py_modules/lsfg_vk/flatpak_service.py') diff --git a/py_modules/lsfg_vk/flatpak_service.py b/py_modules/lsfg_vk/flatpak_service.py index 11f1a82..d2241ab 100644 --- a/py_modules/lsfg_vk/flatpak_service.py +++ b/py_modules/lsfg_vk/flatpak_service.py @@ -1,4 +1,4 @@ -"""Flatpak runtime-extension infrastructure for unified game targets.""" +"""Flatpak runtime support for classified Steam targets.""" from __future__ import annotations @@ -10,7 +10,7 @@ import shutil import subprocess import threading from pathlib import Path -from typing import Any, Dict, List, Optional, Set, Tuple +from typing import Dict, Optional, Set from .base_service import BaseService from .constants import ( @@ -22,14 +22,6 @@ from .constants import ( class FlatpakService(BaseService): - """Resolve and provision only the runtime support a target actually needs. - - Flatpak application permissions are deliberately not persisted here. The - generated per-AppID wrapper supplies the narrow launch-time permissions and - environment instead, while this service owns only the shared Vulkan layer - runtime extensions installed from the plugin bundle. - """ - EXTENSION_ID = "org.freedesktop.Platform.VulkanLayer.lsfgvk" SUPPORTED_RUNTIMES = ("23.08", "24.08", "25.08") OWNERSHIP_FILENAME = "flatpak_extensions.json" @@ -37,7 +29,6 @@ class FlatpakService(BaseService): APP_ID_PATTERN = re.compile( r"^[A-Za-z0-9][A-Za-z0-9-]*(?:\.[A-Za-z0-9][A-Za-z0-9-]*)+$" ) - BRANCH_PATTERN = re.compile(r"^[0-9]+\.[0-9]+$") def __init__(self, logger=None): super().__init__(logger) @@ -48,39 +39,37 @@ class FlatpakService(BaseService): def ownership_path(self) -> Path: return self.config_dir / self.OWNERSHIP_FILENAME - def _get_clean_env(self) -> Dict[str, str]: + def _clean_env(self) -> Dict[str, str]: env = os.environ.copy() env.pop("LD_LIBRARY_PATH", None) env["HOME"] = str(self.user_home) - path_entries = [entry for entry in env.get("PATH", "").split(":") if entry] + 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_entries: - path_entries.insert(0, entry) - env["PATH"] = ":".join(path_entries) + if entry not in path: + path.insert(0, entry) + env["PATH"] = ":".join(path) return env - def _flatpak_user(self) -> pwd.struct_passwd: - try: - return 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 - def check_flatpak_available(self) -> bool: - env = self._get_clean_env() + 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: List[str], **kwargs): + 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] - target_user = self._flatpak_user() - if os.geteuid() != target_user.pw_uid: - runuser = shutil.which("runuser", path=self._get_clean_env()["PATH"]) + 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", target_user.pw_name, "--", *command] - return subprocess.run(command, env=self._get_clean_env(), **kwargs) + command = [runuser, "--user", user.pw_name, "--", *command] + return subprocess.run(command, env=env, **kwargs) @classmethod def _validate_app_id(cls, app_id: str) -> str: @@ -89,45 +78,32 @@ class FlatpakService(BaseService): return app_id @classmethod - def _validate_runtime(cls, version: str) -> str: - if version not in cls.SUPPORTED_RUNTIMES: + def _validate_runtime(cls, branch: str) -> str: + if branch not in cls.SUPPORTED_RUNTIMES: raise ValueError( - f"Unsupported Flatpak runtime branch {version}; " - f"supported branches are {', '.join(cls.SUPPORTED_RUNTIMES)}" + f"Unsupported Flatpak runtime branch {branch}; supported branches are " + + ", ".join(cls.SUPPORTED_RUNTIMES) ) - return version - - @classmethod - def _extension_ref(cls, version: str) -> str: - return f"{cls.EXTENSION_ID}/x86_64/{cls._validate_runtime(version)}" + return branch @classmethod def runtime_branch_from_ref(cls, runtime_ref: str) -> str: - """Return the supported Freedesktop branch from a runtime ref.""" - if not isinstance(runtime_ref, str): - raise ValueError("Flatpak did not return a runtime reference") - parts = runtime_ref.strip().split("/") + 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}") - branch = parts[2] - if not cls.BRANCH_PATTERN.fullmatch(branch): - raise ValueError(f"Unrecognized Flatpak runtime branch: {branch}") - return cls._validate_runtime(branch) + return cls._validate_runtime(parts[2]) @classmethod - def _bundle_filename(cls, version: str) -> str: - return { + 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, - }[cls._validate_runtime(version)] - - def _bundled_extension_path(self, version: str) -> Path: - return ( - Path(__file__).resolve().parent.parent.parent - / BIN_DIR - / self._bundle_filename(version) - ) + }[self._validate_runtime(branch)] + return Path(__file__).resolve().parent.parent.parent / BIN_DIR / filename def _installed_extension_branches(self) -> Set[str]: result = self._run_flatpak_command( @@ -136,78 +112,54 @@ class FlatpakService(BaseService): text=True, check=True, ) - installed: Set[str] = set() + installed = set() for line in result.stdout.splitlines(): - if not line.strip(): - continue - fields = line.split("\t") - if len(fields) < 3: - fields = line.split() - if len(fields) < 3: - continue - application, arch, branch = (field.strip() for field in fields[:3]) - if application == self.EXTENSION_ID and arch == "x86_64": - installed.add(branch) + 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 _read_owned_branches(self) -> Tuple[Set[str], bool]: - """Read ownership without guessing when metadata is damaged.""" + def _owned_branches(self) -> Set[str]: path = self.ownership_path - if path.is_symlink(): - self.log.warning(f"Flatpak ownership metadata is not a regular file: {path}") - return set(), True - if not path.exists(): - return set(), False - if not path.is_file(): - self.log.warning(f"Flatpak ownership metadata is not a regular file: {path}") - return set(), True + 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: - raw = json.loads(path.read_text(encoding="utf-8")) - if not isinstance(raw, dict) or raw.get("version") != self.OWNERSHIP_VERSION: - raise ValueError("unsupported ownership metadata version") - branches = raw.get("plugin_owned_branches") - if not isinstance(branches, list): - raise ValueError("plugin_owned_branches is not a list") - normalized = { - self._validate_runtime(branch) - for branch in branches - if isinstance(branch, str) - } - if len(normalized) != len(branches): - raise ValueError("ownership metadata contains invalid branches") - return normalized, False + 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: - self.log.warning(f"Could not trust Flatpak ownership metadata: {error}") - return set(), True + raise RuntimeError(f"Could not trust Flatpak ownership metadata: {error}") from error def _write_owned_branches(self, branches: Set[str]) -> None: if not branches: - if self.ownership_path.exists() or self.ownership_path.is_symlink(): - self.ownership_path.unlink() + self.ownership_path.unlink(missing_ok=True) return - document = { - "version": self.OWNERSHIP_VERSION, - "plugin_owned_branches": sorted(branches), - } - self._write_file(self.ownership_path, json.dumps(document, indent=2) + "\n") + 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) -> Dict[str, Any]: - """Return global extension inventory for Setup diagnostics.""" + def get_extension_status(self): try: - if not self.check_flatpak_available(): - return self._success_response( - dict, - "Flatpak is not available", - available=False, - extension_id=self.EXTENSION_ID, - supported_branches=list(self.SUPPORTED_RUNTIMES), - installed_branches=[], - ) - installed = self._installed_extension_branches() + available = self.check_flatpak_available() + installed = self._installed_extension_branches() if available else set() return self._success_response( dict, - "Flatpak runtime extension status retrieved", - available=True, + "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), @@ -216,16 +168,15 @@ class FlatpakService(BaseService): return self._error_response( dict, str(error), - available=self.check_flatpak_available(), + available=False, extension_id=self.EXTENSION_ID, supported_branches=list(self.SUPPORTED_RUNTIMES), installed_branches=[], ) - def get_flatpak_support_status(self) -> Dict[str, Any]: - return self.get_extension_status() + get_flatpak_support_status = get_extension_status - def _resolve_runtime(self, app_id: str) -> Dict[str, Any]: + 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") @@ -236,27 +187,21 @@ class FlatpakService(BaseService): ) if result.returncode != 0: raise OSError(result.stderr.strip() or f"Could not inspect Flatpak app {app_id}") - runtime_ref = result.stdout.strip().splitlines()[0] if result.stdout.strip() else "" - branch = self.runtime_branch_from_ref(runtime_ref) - return {"runtime": runtime_ref, "runtime_branch": branch} + runtime = result.stdout.strip().splitlines()[0] if result.stdout.strip() else "" + return runtime, self.runtime_branch_from_ref(runtime) - def resolve_app_support(self, app_id: str) -> Dict[str, Any]: - """Resolve the exact runtime branch required by one Flatpak app.""" + def resolve_app_support(self, app_id: str): try: app_id = self._validate_app_id(app_id) - resolved = self._resolve_runtime(app_id) + runtime, branch = self._resolve_runtime(app_id) installed = self._installed_extension_branches() - branch = resolved["runtime_branch"] 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}" - ), + 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=resolved["runtime"], + runtime=runtime, runtime_branch=branch, support_status="ready" if ready else "needs-runtime", extension_installed=ready, @@ -286,194 +231,114 @@ class FlatpakService(BaseService): installed_branches=[], ) - def install_extension(self, version: str) -> Dict[str, Any]: - """Install one branch, treating an already-installed branch as success.""" + def install_extension(self, branch: str): try: - version = self._validate_runtime(version) + branch = self._validate_runtime(branch) if not self.check_flatpak_available(): raise FileNotFoundError("Flatpak is not available on this system") with self._lock: - installed_before = self._installed_extension_branches() - if version in installed_before: - return self._success_response( - dict, - f"lsfg-vk {version} runtime extension is already installed", - runtime_branch=version, - installed=True, - enabled=True, - ) - bundle_path = self._bundled_extension_path(version) - if not bundle_path.is_file(): - raise FileNotFoundError( - f"Bundled Flatpak extension not found at {bundle_path}; reinstall the plugin" - ) + 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_path), - ], + ["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") - installed_after = self._installed_extension_branches() - if version not in installed_after: - raise RuntimeError( - f"Flatpak install completed but {self._extension_ref(version)} " - "was not visible afterwards" - ) - owned, uncertain = self._read_owned_branches() - if not uncertain: - owned.add(version) + if branch not in self._installed_extension_branches(): + 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(): + 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(): + 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: + removed = self._remove_extension(branch) + owned = self._owned_branches() + if branch in owned: + owned.remove(branch) self._write_owned_branches(owned) - return self._success_response( - dict, - f"lsfg-vk {version} runtime extension installed", - runtime_branch=version, - installed=True, - enabled=True, - ) + return self._extension_result(branch, False, removed, "uninstalled") except Exception as error: return self._error_response( dict, str(error), - runtime_branch=version, + runtime_branch=branch, + removed=False, installed=False, enabled=False, ) - def ensure_extension(self, version: str) -> Dict[str, Any]: - status = self.get_extension_status() - if not status.get("success"): - return status - if not status.get("available"): - return self._error_response( - dict, - "Flatpak is not available on this system", - runtime_branch=version, - support_status="error", - ) + def ensure_extension(self, branch: str): try: - version = self._validate_runtime(version) - except ValueError as error: - return self._error_response(dict, str(error), runtime_branch=version, support_status="unsupported") - if version in status.get("installed_branches", []): - return self._success_response( - dict, - f"lsfg-vk {version} runtime extension is ready", - runtime_branch=version, - installed=True, - ) - return self.install_extension(version) + 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) -> Dict[str, Any]: - """Provision only the branch returned by flatpak info for this app.""" + 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 - branch = resolved.get("runtime_branch") - result = self.ensure_extension(branch) + 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=branch, + runtime_branch=resolved.get("runtime_branch"), support_status="error", extension_installed=False, ) - final = self.resolve_app_support(app_id) - if final.get("success") and final.get("support_status") == "ready": - return final - return self._error_response( - dict, - final.get("error") or "Required Flatpak runtime extension could not be verified", - flatpak_app_id=app_id, - runtime=resolved.get("runtime"), - runtime_branch=branch, - support_status="error", - extension_installed=False, - ) - - def uninstall_extension(self, version: str) -> Dict[str, Any]: - """Uninstall one branch, treating an already-absent branch as success.""" - try: - version = self._validate_runtime(version) - if not self.check_flatpak_available(): - raise FileNotFoundError("Flatpak is not available on this system") - with self._lock: - installed = self._installed_extension_branches() - was_installed = version in installed - if was_installed: - result = self._run_flatpak_command( - [ - "uninstall", - "--user", - "--noninteractive", - self._extension_ref(version), - ], - capture_output=True, - text=True, - ) - if result.returncode != 0: - raise OSError(result.stderr.strip() or "Flatpak uninstall failed") - if version in self._installed_extension_branches(): - raise RuntimeError( - f"Flatpak uninstall completed but {self._extension_ref(version)} " - "is still installed" - ) - owned, uncertain = self._read_owned_branches() - if not uncertain and version in owned: - owned.remove(version) - self._write_owned_branches(owned) - return self._success_response( - dict, - f"lsfg-vk {version} runtime extension uninstalled", - runtime_branch=version, - removed=was_installed, - installed=False, - enabled=False, - ) - except Exception as error: - return self._error_response( - dict, - str(error), - runtime_branch=version, - removed=False, - installed=False, - enabled=False, - ) + return self.resolve_app_support(app_id) - def set_extension_enabled(self, version: str, enabled: bool) -> Dict[str, Any]: - """Set one runtime branch to the requested state, safely and idempotently.""" + 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=version, - installed=False, - enabled=False, - ) - return self.install_extension(version) if enabled else self.uninstall_extension(version) + 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) -> Dict[str, Any]: - """Uninstall only branches recorded as installed by this plugin.""" + def remove_plugin_owned_extensions(self): try: with self._lock: - owned, uncertain = self._read_owned_branches() - if uncertain: - return self._error_response( - dict, - "Flatpak ownership metadata is uncertain; no extensions were removed", - removed_branches=[], - preserved_branches=[], - ownership_uncertain=True, - ) + owned = self._owned_branches() if not owned: return self._success_response( dict, @@ -483,36 +348,11 @@ class FlatpakService(BaseService): ownership_uncertain=False, ) if not self.check_flatpak_available(): - return self._error_response( - dict, - "Flatpak is not available; plugin-owned extension metadata was preserved", - removed_branches=[], - preserved_branches=sorted(owned), - ownership_uncertain=False, - ) - removed: List[str] = [] - failures: List[str] = [] + raise RuntimeError("Flatpak is not available; plugin-owned extension metadata was preserved") + removed, failures = [], [] for branch in sorted(owned): try: - installed = self._installed_extension_branches() - if branch in installed: - 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(): - raise RuntimeError( - f"Flatpak uninstall completed but {self._extension_ref(branch)} " - "is still installed" - ) + self._remove_extension(branch) removed.append(branch) except Exception as error: failures.append(f"{branch}: {error}") @@ -539,5 +379,5 @@ class FlatpakService(BaseService): str(error), removed_branches=[], preserved_branches=[], - ownership_uncertain=False, + ownership_uncertain=True, ) -- cgit v1.2.3 From b197e25b45d53c6c7175a45dd0b14642f2aab198 Mon Sep 17 00:00:00 2001 From: xXJSONDeruloXx Date: Thu, 10 Sep 2026 11:51:22 -0400 Subject: fixes for appimage and flatpak --- py_modules/lsfg_vk/flatpak_service.py | 88 +++++++++++++++++++++++++++-------- 1 file changed, 68 insertions(+), 20 deletions(-) (limited to 'py_modules/lsfg_vk/flatpak_service.py') diff --git a/py_modules/lsfg_vk/flatpak_service.py b/py_modules/lsfg_vk/flatpak_service.py index d2241ab..f486b74 100644 --- a/py_modules/lsfg_vk/flatpak_service.py +++ b/py_modules/lsfg_vk/flatpak_service.py @@ -24,6 +24,8 @@ from .constants import ( 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( @@ -93,6 +95,26 @@ class FlatpakService(BaseService): 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)}" @@ -105,18 +127,22 @@ class FlatpakService(BaseService): }[self._validate_runtime(branch)] return Path(__file__).resolve().parent.parent.parent / BIN_DIR / filename - def _installed_extension_branches(self) -> Set[str]: - result = self._run_flatpak_command( - ["list", "--runtime", "--columns=application,arch,branch"], - capture_output=True, - text=True, - check=True, - ) + 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 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]) + 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]: @@ -188,7 +214,26 @@ class FlatpakService(BaseService): 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 "" - return runtime, self.runtime_branch_from_ref(runtime) + 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: @@ -249,7 +294,7 @@ class FlatpakService(BaseService): ) if result.returncode != 0: raise OSError(result.stderr.strip() or "Flatpak installation failed") - if branch not in self._installed_extension_branches(): + 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) @@ -259,7 +304,7 @@ class FlatpakService(BaseService): 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(): + if branch not in self._installed_extension_branches("user"): return False result = self._run_flatpak_command( ["uninstall", "--user", "--noninteractive", self._extension_ref(branch)], @@ -268,7 +313,7 @@ class FlatpakService(BaseService): ) if result.returncode != 0: raise OSError(result.stderr.strip() or "Flatpak uninstall failed") - if branch in self._installed_extension_branches(): + if branch in self._installed_extension_branches("user"): raise RuntimeError(f"Flatpak uninstall completed but {self._extension_ref(branch)} is still installed") return True @@ -288,12 +333,15 @@ class FlatpakService(BaseService): if not self.check_flatpak_available(): raise FileNotFoundError("Flatpak is not available on this system") with self._lock: - removed = self._remove_extension(branch) owned = self._owned_branches() - if branch in owned: - owned.remove(branch) - self._write_owned_branches(owned) - return self._extension_result(branch, False, removed, "uninstalled") + 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, -- cgit v1.2.3 From 6991f81c9522ff61059f4cc8c84527f20b860032 Mon Sep 17 00:00:00 2001 From: Kurt Himebauch <136133082+xXJSONDeruloXx@users.noreply.github.com> Date: Thu, 10 Sep 2026 15:38:19 -0400 Subject: refactor: make flatpak support explicit per app --- py_modules/lsfg_vk/flatpak_service.py | 535 +++++++++++++++++++++++----------- 1 file changed, 365 insertions(+), 170 deletions(-) (limited to 'py_modules/lsfg_vk/flatpak_service.py') diff --git a/py_modules/lsfg_vk/flatpak_service.py b/py_modules/lsfg_vk/flatpak_service.py index f486b74..62f6a50 100644 --- a/py_modules/lsfg_vk/flatpak_service.py +++ b/py_modules/lsfg_vk/flatpak_service.py @@ -1,7 +1,6 @@ -"""Flatpak runtime support for classified Steam targets.""" - from __future__ import annotations +import hashlib import json import os import pwd @@ -26,8 +25,8 @@ class FlatpakService(BaseService): 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 + OWNERSHIP_FILENAME = "flatpak_state.json" + OWNERSHIP_VERSION = 2 APP_ID_PATTERN = re.compile( r"^[A-Za-z0-9][A-Za-z0-9-]*(?:\.[A-Za-z0-9][A-Za-z0-9-]*)+$" ) @@ -41,6 +40,10 @@ class FlatpakService(BaseService): def ownership_path(self) -> Path: return self.config_dir / self.OWNERSHIP_FILENAME + @property + def backup_dir(self) -> Path: + return self.config_dir / "flatpak-overrides" + def _clean_env(self) -> Dict[str, str]: env = os.environ.copy() env.pop("LD_LIBRARY_PATH", None) @@ -88,13 +91,6 @@ class FlatpakService(BaseService): ) 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 @@ -129,8 +125,6 @@ class FlatpakService(BaseService): 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( @@ -145,64 +139,78 @@ class FlatpakService(BaseService): 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(): + def _empty_state(self) -> Dict[str, object]: + return { + "version": self.OWNERSHIP_VERSION, + "plugin_owned_branches": [], + "prepared_apps": {}, + } + + def _read_state(self) -> Dict[str, object]: + if not self.ownership_path.exists(): + return self._empty_state() + if self.ownership_path.is_symlink() or not self.ownership_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: + data = json.loads(self.ownership_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as error: + raise RuntimeError(f"Could not read Flatpak ownership metadata: {error}") from error + if data.get("version") != self.OWNERSHIP_VERSION: + raise RuntimeError("Unsupported Flatpak ownership metadata version") + branches = data.get("plugin_owned_branches") + apps = data.get("prepared_apps") + if not isinstance(branches, list) or not isinstance(apps, dict): + raise RuntimeError("Invalid Flatpak ownership metadata") + for branch in branches: + self._validate_runtime(branch) + for app_id, entry in apps.items(): + self._validate_app_id(app_id) + if not isinstance(entry, dict): + raise RuntimeError("Invalid Flatpak app ownership metadata") + if type(entry.get("override_existed")) is not bool: + raise RuntimeError("Invalid Flatpak app ownership metadata") + if not isinstance(entry.get("managed_sha256"), str): + raise RuntimeError("Invalid Flatpak app ownership metadata") + return data + + def _write_state(self, state: Dict[str, object]) -> None: + branches = state.get("plugin_owned_branches", []) + apps = state.get("prepared_apps", {}) + if not branches and not apps: self.ownership_path.unlink(missing_ok=True) + if self.backup_dir.exists() and not any(self.backup_dir.iterdir()): + self.backup_dir.rmdir() return self._write_file( self.ownership_path, - json.dumps( - { - "version": self.OWNERSHIP_VERSION, - "plugin_owned_branches": sorted(branches), - }, - indent=2, - ) + "\n", + json.dumps(state, indent=2, sort_keys=True) + "\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=[], - ) + def _owned_branches(self, state: Optional[Dict[str, object]] = None) -> Set[str]: + current = state if state is not None else self._read_state() + return {self._validate_runtime(branch) for branch in current["plugin_owned_branches"]} - get_flatpak_support_status = get_extension_status + def _override_path(self, app_id: str) -> Path: + return self.user_home / ".local/share/flatpak/overrides" / self._validate_app_id(app_id) + + def _backup_path(self, app_id: str) -> Path: + return self.backup_dir / f"{self._validate_app_id(app_id)}.ini" + + @staticmethod + def _sha256(content: bytes) -> str: + return hashlib.sha256(content).hexdigest() + + def _snapshot_override(self, app_id: str) -> tuple[bool, bytes]: + path = self._override_path(app_id) + if path.is_symlink(): + raise RuntimeError("Flatpak override path is a symlink") + if not path.exists(): + return False, b"" + if not path.is_file(): + raise RuntimeError("Flatpak override path is not a regular file") + return True, path.read_bytes() - def _resolve_runtime(self, app_id: str): + def _resolve_runtime(self, app_id: str) -> tuple[str, str]: self._validate_app_id(app_id) if not self.check_flatpak_available(): raise FileNotFoundError("Flatpak is not available on this system") @@ -218,72 +226,126 @@ class FlatpakService(BaseService): 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: + return runtime, self._validate_runtime(parts[2]) + if parts[0] not in self.DERIVED_RUNTIME_IDS: raise ValueError(f"Unsupported Flatpak runtime reference: {runtime}") - return runtime, branch + 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}") + return runtime, self.runtime_branch_from_metadata(metadata_result.stdout) + + 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 resolve_app_support(self, app_id: str): + def _filesystem_present(self, entries: str, host_path: Path) -> bool: + accepted = {str(host_path)} 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 + accepted.add(f"~/{host_path.relative_to(self.user_home).as_posix()}") + except ValueError: + pass + enabled = False + for raw in entries.split(";"): + value = raw.strip() + if not value: + continue + denied = value.startswith("!") + path = value[1:] if denied else value + path = path.split(":", 1)[0] + if path in accepted: + if denied: + return False + enabled = True + return enabled + + def _app_override_status(self, app_id: str) -> Dict[str, object]: + result = self._run_flatpak_command( + ["override", "--user", "--show", app_id], + capture_output=True, + text=True, + ) + output = result.stdout if result.returncode == 0 else "" + section = None + filesystems = "" + unset_environment = set() + environment = {} + for raw_line in output.splitlines(): + line = raw_line.strip() + if line.startswith("[") and line.endswith("]"): + section = line[1:-1] + continue + key, separator, value = line.partition("=") + if not separator: + continue + if section == "Context" and key == "filesystems": + filesystems = value + elif section == "Context" and key == "unset-environment": + unset_environment.update(item for item in value.split(";") if item) + elif section == "Environment": + environment[key] = value + config_ready = self._filesystem_present(filesystems, self.config_dir) + dll_ready = self._filesystem_present(filesystems, self._dll_directory()) + env_ready = ( + environment.get("LSFGVK_CONFIG") == str(self.config_file_path) + and environment.get("LSFGVK_FLATPAK") == "1" + and "DISABLE_LSFGVK" in unset_environment + and "DISABLE_LSFG" in unset_environment + ) + return { + "filesystem_ready": config_ready and dll_ready, + "environment_ready": env_ready, + "prepared": config_ready and dll_ready and env_ready, + } + + 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, - 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, + "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 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, + available=False, + extension_id=self.EXTENSION_ID, + supported_branches=list(self.SUPPORTED_RUNTIMES), installed_branches=[], ) + get_flatpak_support_status = get_extension_status + 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") + installed = self._installed_extension_branches() + if branch in installed: + return self._extension_result(branch, True, False, "is ready") bundle = self._bundled_extension_path(branch) if not bundle.is_file(): raise FileNotFoundError(f"Bundled Flatpak extension not found at {bundle}; reinstall the plugin") @@ -296,9 +358,11 @@ class FlatpakService(BaseService): 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() + state = self._read_state() + owned = self._owned_branches(state) owned.add(branch) - self._write_owned_branches(owned) + state["plugin_owned_branches"] = sorted(owned) + self._write_state(state) 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) @@ -313,8 +377,6 @@ class FlatpakService(BaseService): ) 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): @@ -333,24 +395,19 @@ class FlatpakService(BaseService): if not self.check_flatpak_available(): raise FileNotFoundError("Flatpak is not available on this system") with self._lock: - owned = self._owned_branches() + state = self._read_state() + owned = self._owned_branches(state) 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) + state["plugin_owned_branches"] = sorted(owned) + self._write_state(state) 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, - ) + return self._error_response(dict, str(error), runtime_branch=branch, removed=False, installed=False, enabled=False) def ensure_extension(self, branch: str): try: @@ -358,74 +415,212 @@ class FlatpakService(BaseService): 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._error_response(dict, str(error), runtime_branch=branch, installed=False, enabled=False) 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): + def get_flatpak_apps(self): + try: + if not self.check_flatpak_available(): + raise FileNotFoundError("Flatpak is not available on this system") + installed_extensions = self._installed_extension_branches() + state = self._read_state() + owned_apps = state["prepared_apps"] + result = self._run_flatpak_command( + ["list", "--app", "--columns=name,application"], + capture_output=True, + text=True, + check=True, + ) + apps = [] + for line in result.stdout.splitlines(): + fields = line.split("\t") + if len(fields) < 2: + continue + name, app_id = fields[0].strip(), fields[1].strip() + if not app_id: + continue + item = { + "app_id": app_id, + "app_name": name or app_id, + "runtime": None, + "runtime_branch": None, + "runtime_ready": False, + "prepared": False, + "owned": app_id in owned_apps, + "error": None, + } + try: + runtime, branch = self._resolve_runtime(app_id) + status = self._app_override_status(app_id) + item.update({ + "runtime": runtime, + "runtime_branch": branch, + "runtime_ready": branch in installed_extensions, + "prepared": status["prepared"], + }) + except Exception as error: + item["error"] = str(error) + apps.append(item) + apps.sort(key=lambda item: str(item["app_name"]).lower()) + return self._success_response(dict, f"Found {len(apps)} Flatpak applications", apps=apps) + except Exception as error: + return self._error_response(dict, str(error), apps=[]) + + def prepare_app(self, app_id: str): + try: + app_id = self._validate_app_id(app_id) + with self._lock: + runtime, branch = self._resolve_runtime(app_id) + extension = self.ensure_extension(branch) + if not extension.get("success") or not extension.get("installed"): + raise RuntimeError(extension.get("error") or f"Could not install Flatpak runtime {branch}") + state = self._read_state() + apps = state["prepared_apps"] + status = self._app_override_status(app_id) + if status["prepared"] and app_id not in apps: + return self._success_response( + dict, + "Flatpak application is already prepared outside this plugin", + app_id=app_id, + runtime=runtime, + runtime_branch=branch, + prepared=True, + owned=False, + ) + if app_id not in apps: + existed, original = self._snapshot_override(app_id) + backup = self._backup_path(app_id) + if existed: + self._write_file(backup, original.decode("utf-8")) + else: + backup.unlink(missing_ok=True) + apps[app_id] = { + "override_existed": existed, + "managed_sha256": "", + } + result = self._run_flatpak_command( + [ + "override", + "--user", + f"--filesystem={self.config_dir}:ro", + f"--filesystem={self._dll_directory()}:ro", + f"--env=LSFGVK_CONFIG={self.config_file_path}", + "--env=LSFGVK_FLATPAK=1", + "--unset-env=DISABLE_LSFGVK", + "--unset-env=DISABLE_LSFG", + app_id, + ], + capture_output=True, + text=True, + ) + if result.returncode != 0: + raise OSError(result.stderr.strip() or f"Could not prepare Flatpak app {app_id}") + status = self._app_override_status(app_id) + if not status["prepared"]: + raise RuntimeError(f"Flatpak preparation did not become visible for {app_id}") + existed, managed = self._snapshot_override(app_id) + if not existed: + raise RuntimeError(f"Flatpak override for {app_id} was not created") + apps[app_id]["managed_sha256"] = self._sha256(managed) + self._write_state(state) + return self._success_response( + dict, + "Flatpak application prepared for lsfg-vk", + app_id=app_id, + runtime=runtime, + runtime_branch=branch, + prepared=True, + owned=True, + ) + except Exception as error: + return self._error_response(dict, str(error), app_id=app_id, prepared=False, owned=False) + + def remove_app_override(self, app_id: str): try: + app_id = self._validate_app_id(app_id) with self._lock: - owned = self._owned_branches() - if not owned: + state = self._read_state() + apps = state["prepared_apps"] + entry = apps.get(app_id) + if entry is None: return self._success_response( dict, - "No plugin-owned Flatpak extensions to remove", + "Flatpak application is not plugin-owned; existing overrides were preserved", + app_id=app_id, + prepared=self._app_override_status(app_id)["prepared"], + owned=False, + ) + existed, current = self._snapshot_override(app_id) + current_hash = self._sha256(current) if existed else self._sha256(b"") + if current_hash != entry["managed_sha256"]: + raise RuntimeError( + "Flatpak override changed after preparation; refusing to overwrite unrelated settings" + ) + override_path = self._override_path(app_id) + backup_path = self._backup_path(app_id) + if entry["override_existed"]: + if not backup_path.is_file() or backup_path.is_symlink(): + raise RuntimeError("Flatpak override backup is unavailable") + self._write_file(override_path, backup_path.read_text(encoding="utf-8")) + else: + override_path.unlink(missing_ok=True) + backup_path.unlink(missing_ok=True) + apps.pop(app_id, None) + self._write_state(state) + return self._success_response( + dict, + "Plugin-owned Flatpak preparation removed", + app_id=app_id, + prepared=False, + owned=False, + ) + except Exception as error: + return self._error_response(dict, str(error), app_id=app_id, prepared=False, owned=True) + + def remove_plugin_owned_environment(self): + try: + with self._lock: + state = self._read_state() + failures = [] + removed_apps = [] + for app_id in list(state["prepared_apps"]): + result = self.remove_app_override(app_id) + if result.get("success"): + removed_apps.append(app_id) + else: + failures.append(f"{app_id}: {result.get('error')}") + if failures: + return self._error_response( + dict, + "; ".join(failures), + removed_apps=removed_apps, 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) + state = self._read_state() + removed_branches = [] + for branch in sorted(self._owned_branches(state)): + result = self.uninstall_extension(branch) + if result.get("success"): + removed_branches.append(branch) + else: + failures.append(f"{branch}: {result.get('error')}") if failures: return self._error_response( dict, "; ".join(failures), - removed_branches=removed, - preserved_branches=sorted(remaining), - ownership_uncertain=False, + removed_apps=removed_apps, + removed_branches=removed_branches, ) return self._success_response( dict, - "Plugin-owned Flatpak extensions removed", - removed_branches=removed, - preserved_branches=[], - ownership_uncertain=False, + "Plugin-owned Flatpak state removed", + removed_apps=removed_apps, + removed_branches=removed_branches, ) except Exception as error: - return self._error_response( - dict, - str(error), - removed_branches=[], - preserved_branches=[], - ownership_uncertain=True, - ) + return self._error_response(dict, str(error), removed_apps=[], removed_branches=[]) -- cgit v1.2.3 From 904e2e6131071c3b132d3148947b613c2830b1bb Mon Sep 17 00:00:00 2001 From: xXJSONDeruloXx Date: Thu, 10 Sep 2026 20:49:05 -0400 Subject: fix; correct flatpak extension sources --- py_modules/lsfg_vk/flatpak_service.py | 60 +++++++++++++++++++---------------- 1 file changed, 33 insertions(+), 27 deletions(-) (limited to 'py_modules/lsfg_vk/flatpak_service.py') diff --git a/py_modules/lsfg_vk/flatpak_service.py b/py_modules/lsfg_vk/flatpak_service.py index 62f6a50..4f04382 100644 --- a/py_modules/lsfg_vk/flatpak_service.py +++ b/py_modules/lsfg_vk/flatpak_service.py @@ -12,17 +12,12 @@ 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") + FLATHUB_REMOTE = "flathub" + SUPPORTED_RUNTIMES = ("24.08", "25.08") DERIVED_RUNTIME_IDS = {"org.gnome.Platform", "org.kde.Platform"} RUNTIME_METADATA_SECTION = "Extension org.freedesktop.Platform.GL" OWNERSHIP_FILENAME = "flatpak_state.json" @@ -34,6 +29,7 @@ class FlatpakService(BaseService): def __init__(self, logger=None): super().__init__(logger) self.flatpak_command: Optional[str] = None + self._verified_branches: Set[str] = set() self._lock = threading.RLock() @property @@ -115,14 +111,6 @@ class FlatpakService(BaseService): 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,) installed = set() @@ -139,6 +127,14 @@ class FlatpakService(BaseService): installed.add(fields[2]) return installed + def _user_extension_origin(self, branch: str) -> str: + result = self._run_flatpak_command( + ["info", "--user", "--show-origin", self._extension_ref(branch)], + capture_output=True, + text=True, + ) + return result.stdout.strip() if result.returncode == 0 else "" + def _empty_state(self) -> Dict[str, object]: return { "version": self.OWNERSHIP_VERSION, @@ -343,14 +339,29 @@ class FlatpakService(BaseService): if not self.check_flatpak_available(): raise FileNotFoundError("Flatpak is not available on this system") with self._lock: - installed = self._installed_extension_branches() - if branch in installed: + if branch in self._verified_branches: return self._extension_result(branch, True, False, "is ready") - bundle = self._bundled_extension_path(branch) - if not bundle.is_file(): - raise FileNotFoundError(f"Bundled Flatpak extension not found at {bundle}; reinstall the plugin") + user_installed = self._installed_extension_branches("user") + system_installed = self._installed_extension_branches("system") + if branch in system_installed and branch not in user_installed: + return self._extension_result(branch, True, False, "is ready") + if branch in user_installed and self._user_extension_origin(branch) != self.FLATHUB_REMOTE: + 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 "Could not replace the existing Flatpak extension") result = self._run_flatpak_command( - ["install", "--user", "--noninteractive", "--or-update", str(bundle)], + [ + "install", + "--user", + "--noninteractive", + "--or-update", + self.FLATHUB_REMOTE, + f"{self.EXTENSION_ID}//{branch}", + ], capture_output=True, text=True, ) @@ -363,6 +374,7 @@ class FlatpakService(BaseService): owned.add(branch) state["plugin_owned_branches"] = sorted(owned) self._write_state(state) + self._verified_branches.add(branch) 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) @@ -410,12 +422,6 @@ class FlatpakService(BaseService): 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, installed=False, enabled=False) return self.install_extension(branch) def set_extension_enabled(self, branch: str, enabled: bool): -- cgit v1.2.3 From d2bfdafa92f3d31cc5392bf8a5a1f1df5c2358ce Mon Sep 17 00:00:00 2001 From: xXJSONDeruloXx Date: Fri, 11 Sep 2026 09:05:14 -0400 Subject: flatpak correctness, ui alignment, tests --- py_modules/lsfg_vk/flatpak_service.py | 30 ++++++++++++++++++++++++++++++ 1 file changed, 30 insertions(+) (limited to 'py_modules/lsfg_vk/flatpak_service.py') diff --git a/py_modules/lsfg_vk/flatpak_service.py b/py_modules/lsfg_vk/flatpak_service.py index 4f04382..f2ed90c 100644 --- a/py_modules/lsfg_vk/flatpak_service.py +++ b/py_modules/lsfg_vk/flatpak_service.py @@ -44,6 +44,13 @@ class FlatpakService(BaseService): env = os.environ.copy() env.pop("LD_LIBRARY_PATH", None) env["HOME"] = str(self.user_home) + try: + user_id = self.user_home.stat().st_uid + except OSError: + user_id = None + if user_id is not None: + env["XDG_RUNTIME_DIR"] = f"/run/user/{user_id}" + env["DBUS_SESSION_BUS_ADDRESS"] = f"unix:path=/run/user/{user_id}/bus" 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: @@ -196,6 +203,29 @@ class FlatpakService(BaseService): def _sha256(content: bytes) -> str: return hashlib.sha256(content).hexdigest() + @staticmethod + def _parse_process_start_time(stat_content: str) -> Optional[int]: + closing_command = stat_content.rfind(")") + if closing_command < 0: + return None + fields = stat_content[closing_command + 2:].split() + if len(fields) <= 19: + return None + try: + return int(fields[19]) + except (TypeError, ValueError): + return None + + @classmethod + def _process_start_time(cls, pid: str) -> Optional[int]: + if not isinstance(pid, str) or re.fullmatch(r"[0-9]+", pid) is None: + return None + try: + stat_content = (Path("/proc") / pid / "stat").read_text(encoding="utf-8") + except OSError: + return None + return cls._parse_process_start_time(stat_content) + def _snapshot_override(self, app_id: str) -> tuple[bool, bytes]: path = self._override_path(app_id) if path.is_symlink(): -- cgit v1.2.3