diff --git a/ptlinux/news.py b/ptlinux/news.py index 3ad24fe..ffcc6f6 100644 --- a/ptlinux/news.py +++ b/ptlinux/news.py @@ -153,7 +153,13 @@ def fetch_news( _write_cache(cache_path, articles) return articles except (OSError, NewsError, UnicodeDecodeError, json.JSONDecodeError): - return _read_cache(cache_path) + 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) def fetch_latest_broadcast( @@ -179,7 +185,13 @@ def fetch_latest_broadcast( _write_broadcast_cache(cache_path, broadcast) return broadcast except (OSError, NewsError, UnicodeDecodeError, json.JSONDecodeError): - return _read_broadcast_cache(cache_path) + 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) def _parse_news(payload: bytes) -> tuple[NewsArticle, ...]: @@ -288,7 +300,7 @@ def _read_cache(path: Path) -> tuple[NewsArticle, ...]: if not isinstance(value, list): return () return tuple(NewsArticle(**item) for item in value if isinstance(item, dict)) - except (OSError, TypeError, json.JSONDecodeError): + except (OSError, UnicodeError, TypeError, json.JSONDecodeError): return () @@ -309,7 +321,7 @@ def _read_broadcast_cache(path: Path) -> Broadcast | None: try: value = json.loads(path.read_text(encoding="utf-8")) return Broadcast(**value) if isinstance(value, dict) else None - except (OSError, TypeError, json.JSONDecodeError): + except (OSError, UnicodeError, TypeError, json.JSONDecodeError): return None diff --git a/ptlinux/self_update.py b/ptlinux/self_update.py index 8c3d345..7bb3d7e 100644 --- a/ptlinux/self_update.py +++ b/ptlinux/self_update.py @@ -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") _SHA256_PATTERN = re.compile(r"[0-9a-f]{64}\Z") _ALLOWED_RELEASE_ROOTS = {"ptlinux", "ptdiag", "release.json"} +UPDATE_CHECK_CACHE_SCHEMA = 1 class LauncherUpdateError(RuntimeError): @@ -143,6 +144,89 @@ 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) + + +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") latest = _parse_version(manifest.version, "manifest version") minimum = _parse_version( @@ -170,7 +254,7 @@ def check_for_launcher_update( available=available, supported=supported, reason=reason, - manifest_url=url, + manifest_url=manifest_url, manifest=manifest, )