diff options
Diffstat (limited to 'py_modules')
| -rw-r--r-- | py_modules/lsfg_vk/flatpak_service.py | 535 |
1 files changed, 365 insertions, 170 deletions
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) @@ -89,13 +92,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 versions = [] @@ -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=[]) |