From 2120af63849385c236dc5f1658c995bcd69e800d Mon Sep 17 00:00:00 2001 From: R1kaB3rN <100738684+R1kaB3rN@users.noreply.github.com> Date: Wed, 1 Jan 2025 17:00:04 -0800 Subject: [PATCH] umu_proton: fix not finding delta builds when offline --- umu/umu_proton.py | 131 +++++++++++++++++++++------------------------- 1 file changed, 59 insertions(+), 72 deletions(-) diff --git a/umu/umu_proton.py b/umu/umu_proton.py index 8af880668..16eb14128 100644 --- a/umu/umu_proton.py +++ b/umu/umu_proton.py @@ -16,11 +16,7 @@ from urllib3.poolmanager import PoolManager from urllib3.response import BaseHTTPResponse -from umu.umu_bspatch import ( - Content, - ContentContainer, - CustomPatcher, -) +from umu.umu_bspatch import Content, ContentContainer, CustomPatcher from umu.umu_consts import STEAM_COMPAT, UMU_CACHE, UMU_COMPAT, UMU_LOCAL, HTTPMethod from umu.umu_log import log from umu.umu_util import ( @@ -79,23 +75,18 @@ def get_umu_proton(env: dict[str, str], session_pools: SessionPools) -> dict[str except HTTPError: log.debug("Network is unreachable") - with ( - TemporaryDirectory() as tmp, - TemporaryDirectory(dir=UMU_CACHE) as tmpcache, - ): + with TemporaryDirectory() as tmp, TemporaryDirectory(dir=UMU_CACHE) as tmpcache: tmpdirs: SessionCaches = (Path(tmp), Path(tmpcache)) - if _get_delta(env, UMU_COMPAT, tmpdirs, patch, session_pools) is env: + compatdirs = (UMU_COMPAT, STEAM_COMPAT) + if _get_delta(env, UMU_COMPAT, patch, assets, session_pools) is env: log.info("%s is up to date", os.environ["PROTONPATH"]) os.environ["PROTONPATH"] = str( UMU_COMPAT.joinpath(os.environ["PROTONPATH"]) ) return env - if ( - _get_latest(env, (UMU_COMPAT, STEAM_COMPAT), tmpdirs, assets, session_pools) - is env - ): + if _get_latest(env, compatdirs, tmpdirs, assets, session_pools) is env: return env - if _get_from_steamcompat(env, STEAM_COMPAT) is env: + if _get_from_compat(env, (UMU_COMPAT, STEAM_COMPAT)) is env: return env os.environ["PROTONPATH"] = "" @@ -303,8 +294,8 @@ def _fetch_proton( return env -def _get_from_steamcompat( - env: dict[str, str], steam_compat: Path +def _get_from_compat( + env: dict[str, str], compats: tuple[Path, Path] ) -> dict[str, str] | None: """Refer to Steam's compatibilitytools.d folder for any existing Protons. @@ -312,26 +303,28 @@ def _get_from_steamcompat( from a digest mismatch, request failure or unreachable network, the latest existing Proton build of that same version will be used """ - version: str = ( - "GE-Proton" if os.environ.get("PROTONPATH") == "GE-Proton" else "UMU-Proton" - ) + version: str = os.environ.get("PROTONPATH", "") + proton_versions: set[str] = {member.value for member in ProtonVersion} + + if version not in proton_versions: + return None try: - latest: Path = max( - ( - proton - for proton in steam_compat.glob("*") - if proton.name.startswith(version) - ), - key=lambda proton: [ - int(text) if text.isdigit() else text.lower() - for text in resplit(r"(\d+)", proton.name) - ], - ) - log.info("%s found in '%s'", latest.name, steam_compat) - log.info("Using %s", latest.name) - os.environ["PROTONPATH"] = str(latest) - env["PROTONPATH"] = os.environ["PROTONPATH"] + for compat in compats: + latest: Path = max( + filter( + lambda proton: proton.name.startswith(version), compat.glob("*") + ), + key=lambda proton: [ + int(text) if text.isdigit() else text.lower() + for text in resplit(r"(\d+)", proton.name) + ], + ) + log.info("%s found in '%s'", latest.name, compat) + log.info("Using %s", latest.name) + os.environ["PROTONPATH"] = str(latest) + env["PROTONPATH"] = os.environ["PROTONPATH"] + break except ValueError: return None @@ -379,34 +372,6 @@ def _get_latest( if os.environ.get("PROTONPATH") in proton_versions: version = os.environ["PROTONPATH"] - # Return if the latest Proton is already installed in private directory - compat_version: Path = umu_compat.joinpath(version) - if ( - version in latest_candidates - and compat_version.is_dir() - and compat_version.joinpath("compatibilitytool.vdf").is_file() - ): - vdf: Path = umu_compat.joinpath(version, "compatibilitytool.vdf") - try: - with vdf.open(encoding="utf-8") as file: - # We're up to date if the internal tool is the GH asset name - # without the suffix - for line in file: - if proton not in line: - continue - log.info("%s is up to date", version) - os.environ["PROTONPATH"] = str(umu_compat.joinpath(version)) - env["PROTONPATH"] = os.environ["PROTONPATH"] - return env - except UnicodeDecodeError: - # Case when the VDF file is the binary format/has non-utf-8 chars - # Return and fallback to Steam's compatibilitytools.d - log.warning( - "Failed opening file '%s', unable to determine latest build", - vdf, - ) - return None - # Return if the latest Proton is already installed if steam_compat.joinpath(proton).is_dir(): log.info("%s is up to date", version) @@ -519,11 +484,10 @@ def _install_proton( def _get_delta( env: dict[str, str], umu_compat: Path, - session_caches: SessionCaches, patch: bytes, + assets: tuple[tuple[str, str], tuple[str, str]], session_pools: SessionPools, ) -> dict[str, str] | None: - _, cache = session_caches thread_pool, _ = session_pools version: str = ( "GE-Latest" if os.environ.get("PROTONPATH") == "GE-Latest" else "UMU-Latest" @@ -532,12 +496,18 @@ def _get_delta( cbor: ContentContainer lockfile: str = f"{UMU_LOCAL}/compatibilitytools.d.lock" - if not proton.is_dir(): - log.debug("File '%s' does not exist, skipping update", proton) + if os.environ.get("PROTONPATH") not in { + ProtonVersion.GELatest.value, + ProtonVersion.UMULatest.value, + }: + log.debug("PROTONPATH not *-Latest, skipping") + return None + + if not assets: return None if not patch: - log.debug("Received empty byte string for patch, skipping update") + log.debug("Received empty byte string for patch, skipping") return None from cbor2 import CBORDecodeError, dumps, loads @@ -552,8 +522,26 @@ def _get_delta( log.debug("Acquiring lock '%s'", lockfile) with unix_flock(lockfile): + tarball, _ = assets[1] + build: str = tarball.removesuffix(".tar.gz") + vdf: Path = umu_compat.joinpath(version, "compatibilitytool.vdf") + log.debug("Acquired lock '%s'", lockfile) + # Check if we're up to date + try: + with vdf.open(encoding="utf-8") as file: + # We're up to date if the internal tool is the GH asset name w/o suffix + if any(filter(lambda line: build in line, file)): + log.info("%s is up to date", version) + os.environ["PROTONPATH"] = str(umu_compat.joinpath(version)) + env["PROTONPATH"] = os.environ["PROTONPATH"] + return env + except (UnicodeDecodeError, FileNotFoundError): + # Case when the VDF file DNE/or has non-utf-8 chars + log.error("Failed opening file '%s', unable to determine latest build", vdf) + return None + # Validate the integrity of the embedded public key. Use RustCrypto's SHA2 # implementation to keep the security boundary consistent if not valid_key(cbor["public_key"]): @@ -582,7 +570,7 @@ def _get_delta( src: str = content["source"] if src.startswith((ProtonVersion.GE.value, ProtonVersion.UMU.value)): - patchers.append(_apply_delta(proton, cache, content, thread_pool)) + patchers.append(_apply_delta(proton, content, thread_pool)) continue subdir: Path | None = next(umu_compat.joinpath(version).rglob(src), None) @@ -590,7 +578,7 @@ def _get_delta( log.error("Could not find subdirectory '%s', skipping", subdir) continue - patchers.append(_apply_delta(subdir, cache, content, thread_pool)) + patchers.append(_apply_delta(subdir, content, thread_pool)) renames.append((subdir, subdir.parent / content["target"])) # Wait for results and rename versioned subdirectories @@ -609,11 +597,10 @@ def _get_delta( def _apply_delta( path: Path, - cache: Path, content: Content, thread_pool: ThreadPoolExecutor, ) -> CustomPatcher | None: - patcher: CustomPatcher = CustomPatcher(content, path, cache, thread_pool) + patcher: CustomPatcher = CustomPatcher(content, path, thread_pool) # Verify the identity of the build. At this point the patch file is authenticated. # Note, this will skip the update if the user had tinkered with their build. We do