Added resolution settings and updater modal with rollback functionality for launcher versions.
This commit is contained in:
@@ -74,6 +74,7 @@ class LauncherUpdateCheck:
|
||||
reason: str
|
||||
manifest_url: str
|
||||
manifest: UpdateManifest
|
||||
checked_at: str | None = None
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
@@ -110,6 +111,26 @@ class LauncherRollbackResult:
|
||||
restart_required: bool
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class LauncherInstalledRelease:
|
||||
"""One validated release retained in the launcher's version store."""
|
||||
|
||||
version: str
|
||||
installed_at: str
|
||||
directory: str
|
||||
active: bool
|
||||
previous: bool
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class LauncherUpdateHistory:
|
||||
"""Validated launcher releases available for inspection or rollback."""
|
||||
|
||||
current_version: str | None
|
||||
previous_version: str | None
|
||||
releases: tuple[LauncherInstalledRelease, ...]
|
||||
|
||||
|
||||
def default_public_key_path() -> Path:
|
||||
return Path(__file__).with_name(PUBLIC_KEY_FILENAME)
|
||||
|
||||
@@ -157,7 +178,12 @@ def check_for_launcher_update(
|
||||
)
|
||||
_verify_manifest_signature(manifest_bytes, signature_bytes, key)
|
||||
manifest = parse_update_manifest(manifest_bytes)
|
||||
return _evaluate_launcher_update(manifest, current_version, url)
|
||||
return _evaluate_launcher_update(
|
||||
manifest,
|
||||
current_version,
|
||||
url,
|
||||
checked_at=datetime.now().astimezone().isoformat(timespec="seconds"),
|
||||
)
|
||||
|
||||
|
||||
def read_cached_launcher_update_check(
|
||||
@@ -198,13 +224,21 @@ def read_cached_launcher_update_check(
|
||||
manifest_value = value.get("manifest")
|
||||
if not isinstance(manifest_value, dict):
|
||||
return None
|
||||
checked_at = _cached_timestamp(value.get("checked_at"))
|
||||
if checked_at is None:
|
||||
return None
|
||||
try:
|
||||
manifest = parse_update_manifest(
|
||||
(json.dumps(manifest_value, sort_keys=True) + "\n").encode("utf-8")
|
||||
)
|
||||
except (LauncherUpdateError, TypeError, ValueError):
|
||||
return None
|
||||
return _evaluate_launcher_update(manifest, current_version, url)
|
||||
return _evaluate_launcher_update(
|
||||
manifest,
|
||||
current_version,
|
||||
url,
|
||||
checked_at=checked_at,
|
||||
)
|
||||
|
||||
|
||||
def cache_launcher_update_check(
|
||||
@@ -223,7 +257,8 @@ def cache_launcher_update_check(
|
||||
cache_path,
|
||||
{
|
||||
"schema": UPDATE_CHECK_CACHE_SCHEMA,
|
||||
"checked_at": datetime.now().astimezone().isoformat(),
|
||||
"checked_at": check.checked_at
|
||||
or datetime.now().astimezone().isoformat(timespec="seconds"),
|
||||
"current_version": check.current_version,
|
||||
"manifest_url": check.manifest_url,
|
||||
"public_key_sha256": key_sha256,
|
||||
@@ -239,6 +274,8 @@ def _evaluate_launcher_update(
|
||||
manifest: UpdateManifest,
|
||||
current_version: str,
|
||||
manifest_url: str,
|
||||
*,
|
||||
checked_at: str | None = None,
|
||||
) -> LauncherUpdateCheck:
|
||||
current = _parse_version(current_version, "current launcher version")
|
||||
latest = _parse_version(manifest.version, "manifest version")
|
||||
@@ -269,6 +306,7 @@ def _evaluate_launcher_update(
|
||||
reason=reason,
|
||||
manifest_url=manifest_url,
|
||||
manifest=manifest,
|
||||
checked_at=checked_at,
|
||||
)
|
||||
|
||||
|
||||
@@ -408,12 +446,22 @@ def rollback_launcher_update(paths: AppPaths) -> LauncherRollbackResult:
|
||||
raise LauncherUpdateError("the current launcher release link is missing")
|
||||
if previous is None:
|
||||
raise LauncherUpdateError("no previous launcher release is available")
|
||||
current_version = _validate_release_directory(current)
|
||||
previous_version = _validate_release_directory(previous)
|
||||
if current == previous:
|
||||
current_release = current
|
||||
previous_release = previous
|
||||
current_version = _validate_release_directory(current_release)
|
||||
previous_version = _validate_release_directory(previous_release)
|
||||
if current_release == previous_release:
|
||||
raise LauncherUpdateError("current and previous launcher releases are identical")
|
||||
_replace_release_link(paths.launcher_previous, current, paths.launcher_releases)
|
||||
_replace_release_link(paths.launcher_current, previous, paths.launcher_releases)
|
||||
_replace_release_link(
|
||||
paths.launcher_previous,
|
||||
current_release,
|
||||
paths.launcher_releases,
|
||||
)
|
||||
_replace_release_link(
|
||||
paths.launcher_current,
|
||||
previous_release,
|
||||
paths.launcher_releases,
|
||||
)
|
||||
_fsync_directory(paths.app_home)
|
||||
_write_json(
|
||||
paths.launcher_update_state,
|
||||
@@ -422,19 +470,64 @@ def rollback_launcher_update(paths: AppPaths) -> LauncherRollbackResult:
|
||||
"rolled_back_at": datetime.now().astimezone().isoformat(),
|
||||
"from_version": current_version,
|
||||
"to_version": previous_version,
|
||||
"release_directory": str(previous),
|
||||
"previous_release": str(current),
|
||||
"release_directory": str(previous_release),
|
||||
"previous_release": str(current_release),
|
||||
},
|
||||
)
|
||||
return LauncherRollbackResult(
|
||||
from_version=current_version,
|
||||
to_version=previous_version,
|
||||
current_release=str(previous),
|
||||
previous_release=str(current),
|
||||
current_release=str(previous_release),
|
||||
previous_release=str(current_release),
|
||||
restart_required=True,
|
||||
)
|
||||
|
||||
|
||||
def inspect_launcher_update_history(paths: AppPaths) -> LauncherUpdateHistory:
|
||||
"""Return validated retained releases without trusting directory names alone."""
|
||||
|
||||
current = _release_from_link(paths.launcher_current, paths.launcher_releases)
|
||||
previous = _release_from_link(paths.launcher_previous, paths.launcher_releases)
|
||||
try:
|
||||
candidates = tuple(paths.launcher_releases.iterdir())
|
||||
except FileNotFoundError:
|
||||
candidates = ()
|
||||
except OSError as exc:
|
||||
raise LauncherUpdateError(f"cannot inspect installed launcher releases: {exc}") from exc
|
||||
releases: list[LauncherInstalledRelease] = []
|
||||
for candidate in candidates:
|
||||
if _VERSION_PATTERN.fullmatch(candidate.name) is None:
|
||||
continue
|
||||
try:
|
||||
version = _validate_release_directory(candidate)
|
||||
installed_at = datetime.fromtimestamp(
|
||||
candidate.stat().st_mtime
|
||||
).astimezone().isoformat(timespec="seconds")
|
||||
except (LauncherUpdateError, OSError, ValueError, OverflowError):
|
||||
continue
|
||||
if version != candidate.name:
|
||||
continue
|
||||
resolved = candidate.resolve()
|
||||
releases.append(
|
||||
LauncherInstalledRelease(
|
||||
version=version,
|
||||
installed_at=installed_at,
|
||||
directory=str(resolved),
|
||||
active=resolved == current,
|
||||
previous=resolved == previous,
|
||||
)
|
||||
)
|
||||
releases.sort(
|
||||
key=lambda release: _parse_version(release.version, "launcher release version"),
|
||||
reverse=True,
|
||||
)
|
||||
return LauncherUpdateHistory(
|
||||
current_version=_release_version(releases, active=True),
|
||||
previous_version=_release_version(releases, previous=True),
|
||||
releases=tuple(releases),
|
||||
)
|
||||
|
||||
|
||||
def parse_update_manifest(data: bytes) -> UpdateManifest:
|
||||
"""Strictly parse one already-authenticated manifest."""
|
||||
|
||||
@@ -857,9 +950,15 @@ def _activate_release(paths: AppPaths, release: Path) -> Path | None:
|
||||
raise LauncherUpdateError(
|
||||
f"launcher previous path is not a symbolic link: {paths.launcher_previous}"
|
||||
)
|
||||
if current is not None and current != release:
|
||||
_validate_release_directory(current)
|
||||
_replace_release_link(paths.launcher_previous, current, paths.launcher_releases)
|
||||
if current is not None:
|
||||
current_release = current
|
||||
if current_release != release:
|
||||
_validate_release_directory(current_release)
|
||||
_replace_release_link(
|
||||
paths.launcher_previous,
|
||||
current_release,
|
||||
paths.launcher_releases,
|
||||
)
|
||||
_replace_release_link(paths.launcher_current, release, paths.launcher_releases)
|
||||
_fsync_directory(paths.app_home)
|
||||
return current
|
||||
@@ -980,6 +1079,30 @@ def _parse_published_at(value: str) -> datetime:
|
||||
return parsed
|
||||
|
||||
|
||||
def _cached_timestamp(value: object) -> str | None:
|
||||
if not isinstance(value, str) or not value:
|
||||
return None
|
||||
try:
|
||||
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
||||
except ValueError:
|
||||
return None
|
||||
if parsed.tzinfo is None or parsed.utcoffset() is None:
|
||||
return None
|
||||
return value
|
||||
|
||||
|
||||
def _release_version(
|
||||
releases: list[LauncherInstalledRelease],
|
||||
*,
|
||||
active: bool = False,
|
||||
previous: bool = False,
|
||||
) -> str | None:
|
||||
for release in releases:
|
||||
if (active and release.active) or (previous and release.previous):
|
||||
return release.version
|
||||
return None
|
||||
|
||||
|
||||
def _required_string(value: dict[str, object], key: str) -> str:
|
||||
item = value.get(key)
|
||||
if not isinstance(item, str) or not item:
|
||||
|
||||
Reference in New Issue
Block a user