Reworked cache for news and update manifest.

This commit is contained in:
2026-08-28 22:48:19 +02:00
parent c49c56bc97
commit d20eea6877
2 changed files with 101 additions and 5 deletions

View File

@@ -153,6 +153,12 @@ def fetch_news(
_write_cache(cache_path, articles) _write_cache(cache_path, articles)
return articles return articles
except (OSError, NewsError, UnicodeDecodeError, json.JSONDecodeError): except (OSError, NewsError, UnicodeDecodeError, json.JSONDecodeError):
return read_cached_news(cache_path)
def read_cached_news(cache_path: Path) -> tuple[NewsArticle, ...]:
"""Read the last valid news feed without performing network I/O."""
return _read_cache(cache_path) return _read_cache(cache_path)
@@ -179,6 +185,12 @@ def fetch_latest_broadcast(
_write_broadcast_cache(cache_path, broadcast) _write_broadcast_cache(cache_path, broadcast)
return broadcast return broadcast
except (OSError, NewsError, UnicodeDecodeError, json.JSONDecodeError): except (OSError, NewsError, UnicodeDecodeError, json.JSONDecodeError):
return read_cached_broadcast(cache_path)
def read_cached_broadcast(cache_path: Path) -> Broadcast | None:
"""Read the last valid broadcast without performing network I/O."""
return _read_broadcast_cache(cache_path) return _read_broadcast_cache(cache_path)
@@ -288,7 +300,7 @@ def _read_cache(path: Path) -> tuple[NewsArticle, ...]:
if not isinstance(value, list): if not isinstance(value, list):
return () return ()
return tuple(NewsArticle(**item) for item in value if isinstance(item, dict)) return tuple(NewsArticle(**item) for item in value if isinstance(item, dict))
except (OSError, TypeError, json.JSONDecodeError): except (OSError, UnicodeError, TypeError, json.JSONDecodeError):
return () return ()
@@ -309,7 +321,7 @@ def _read_broadcast_cache(path: Path) -> Broadcast | None:
try: try:
value = json.loads(path.read_text(encoding="utf-8")) value = json.loads(path.read_text(encoding="utf-8"))
return Broadcast(**value) if isinstance(value, dict) else None return Broadcast(**value) if isinstance(value, dict) else None
except (OSError, TypeError, json.JSONDecodeError): except (OSError, UnicodeError, TypeError, json.JSONDecodeError):
return None return None

View File

@@ -45,6 +45,7 @@ UPDATE_ARCHIVE_PREFIX = "pristontale-eu-linux-update-v"
_VERSION_PATTERN = re.compile(r"(?:0|[1-9][0-9]*)\.(?:0|[1-9][0-9]*)\.(?:0|[1-9][0-9]*)\Z") _VERSION_PATTERN = re.compile(r"(?:0|[1-9][0-9]*)\.(?:0|[1-9][0-9]*)\.(?:0|[1-9][0-9]*)\Z")
_SHA256_PATTERN = re.compile(r"[0-9a-f]{64}\Z") _SHA256_PATTERN = re.compile(r"[0-9a-f]{64}\Z")
_ALLOWED_RELEASE_ROOTS = {"ptlinux", "ptdiag", "release.json"} _ALLOWED_RELEASE_ROOTS = {"ptlinux", "ptdiag", "release.json"}
UPDATE_CHECK_CACHE_SCHEMA = 1
class LauncherUpdateError(RuntimeError): class LauncherUpdateError(RuntimeError):
@@ -143,6 +144,89 @@ def check_for_launcher_update(
) )
_verify_manifest_signature(manifest_bytes, signature_bytes, key) _verify_manifest_signature(manifest_bytes, signature_bytes, key)
manifest = parse_update_manifest(manifest_bytes) manifest = parse_update_manifest(manifest_bytes)
return _evaluate_launcher_update(manifest, current_version, url)
def read_cached_launcher_update_check(
cache_path: Path,
*,
current_version: str = __version__,
manifest_url: str | None = None,
public_key_path: Path | None = None,
environment: Mapping[str, str] | None = None,
) -> LauncherUpdateCheck | None:
"""Read a display-only result from the last authenticated live check."""
url = manifest_url or configured_manifest_url(environment)
key = public_key_path or configured_public_key_path(environment)
try:
_parse_version(current_version, "current launcher version")
_require_https_url(url, "update manifest URL")
key_sha256 = _sha256_file(key.expanduser().resolve())
value = json.loads(cache_path.read_text(encoding="utf-8"))
except (
OSError,
UnicodeError,
json.JSONDecodeError,
LauncherUpdateError,
RuntimeError,
):
return None
if not isinstance(value, dict):
return None
if value.get("schema") != UPDATE_CHECK_CACHE_SCHEMA:
return None
if value.get("current_version") != current_version:
return None
if value.get("manifest_url") != url:
return None
if value.get("public_key_sha256") != key_sha256:
return None
manifest_value = value.get("manifest")
if not isinstance(manifest_value, dict):
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)
def cache_launcher_update_check(
cache_path: Path,
check: LauncherUpdateCheck,
*,
public_key_path: Path | None = None,
environment: Mapping[str, str] | None = None,
) -> bool:
"""Cache an authenticated result for startup display, never for installation."""
key = public_key_path or configured_public_key_path(environment)
try:
key_sha256 = _sha256_file(key.expanduser().resolve())
_write_json(
cache_path,
{
"schema": UPDATE_CHECK_CACHE_SCHEMA,
"checked_at": datetime.now().astimezone().isoformat(),
"current_version": check.current_version,
"manifest_url": check.manifest_url,
"public_key_sha256": key_sha256,
"manifest": asdict(check.manifest),
},
)
except (OSError, RuntimeError):
return False
return True
def _evaluate_launcher_update(
manifest: UpdateManifest,
current_version: str,
manifest_url: str,
) -> LauncherUpdateCheck:
current = _parse_version(current_version, "current launcher version") current = _parse_version(current_version, "current launcher version")
latest = _parse_version(manifest.version, "manifest version") latest = _parse_version(manifest.version, "manifest version")
minimum = _parse_version( minimum = _parse_version(
@@ -170,7 +254,7 @@ def check_for_launcher_update(
available=available, available=available,
supported=supported, supported=supported,
reason=reason, reason=reason,
manifest_url=url, manifest_url=manifest_url,
manifest=manifest, manifest=manifest,
) )