Skip to content

Commit

Permalink
umu_proton: fix not finding delta builds when offline
Browse files Browse the repository at this point in the history
  • Loading branch information
@R1kaB3rN
R1kaB3rN committed Jan 2, 2025
1 parent 794e2bf commit 2120af6
Showing 1 changed file with 59 additions and 72 deletions.
131 changes: 59 additions & 72 deletions umu/umu_proton.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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 (
Expand Down Expand Up @@ -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"] = ""
Expand Down Expand Up @@ -303,35 +294,37 @@ 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.
When an error occurs in the process of using the latest Proton build either
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

Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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"
Expand All @@ -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
Expand All @@ -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"]):
Expand Down Expand Up @@ -582,15 +570,15 @@ 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)
if not subdir:
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
Expand All @@ -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
Expand Down

0 comments on commit 2120af6

Please sign in to comment.