Coverage for src/ph/updater.py: 99.3%
212 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-28 08:17 +0000
1"""Discover and safely stage self-contained Pocket Harbor release updates."""
3import json
4import logging
5import os
6import re
7import shutil
8import ssl
9import stat
10import zipfile
11from collections.abc import Callable
12from dataclasses import dataclass
13from importlib.metadata import PackageNotFoundError, version
14from pathlib import Path, PurePosixPath
15from urllib.error import HTTPError, URLError
16from urllib.request import Request, urlopen
18from ph.store import USER_AGENT
19from ph.targets import DARKOS, LinuxTarget
21LOGGER = logging.getLogger(__name__)
23DEFAULT_UPDATE_API_URL = "https://api.github.com/repos/zhavir/PoketHarbor/releases/latest"
24PENDING_UPDATE_DIRECTORY = DARKOS.update_staging_directory
25READY_MARKER = ".ready"
26MAX_RELEASE_METADATA_BYTES = 2 * 1024 * 1024
27MAX_UPDATE_ARCHIVE_BYTES = 2 * 1024 * 1024 * 1024
28MAX_UPDATE_FILES = 25_000
29_VERSION_PATTERN = re.compile(r"^v?(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)$")
31type ProgressCallback = Callable[[str, int, int | None], None]
32type CancelCallback = Callable[[], bool]
35class UpdateError(RuntimeError):
36 """An application update could not be checked or staged."""
39class UpdateCancelled(UpdateError):
40 """The user cancelled an application update."""
43@dataclass(frozen=True, slots=True)
44class ReleaseUpdate:
45 """The exact device bundle attached to a newer GitHub release."""
47 version: str
48 tag: str
49 asset_name: str
50 asset_url: str
51 asset_size: int | None
54def installed_version() -> str:
55 """Return the packaged application version used for update comparisons."""
57 try:
58 return version("pocket-harbor")
59 except PackageNotFoundError:
60 return "development"
63def find_update(
64 current_version: str,
65 api_url: str = DEFAULT_UPDATE_API_URL,
66 timeout_seconds: float = 30.0,
67 target: LinuxTarget = DARKOS,
68) -> ReleaseUpdate | None:
69 """Return a newer stable device release, if GitHub publishes one."""
71 current = _parse_version(current_version, "installed application version")
72 LOGGER.info("Checking for an application update from version=%s", current_version)
73 request = Request(
74 api_url,
75 headers={
76 "Accept": "application/vnd.github+json",
77 "User-Agent": USER_AGENT,
78 "X-GitHub-Api-Version": "2022-11-28",
79 },
80 )
81 try:
82 with urlopen(
83 request, timeout=timeout_seconds, context=ssl.create_default_context()
84 ) as response:
85 content_length = response.headers.get("Content-Length")
86 if (
87 content_length
88 and content_length.isdigit()
89 and int(content_length) > MAX_RELEASE_METADATA_BYTES
90 ):
91 raise UpdateError("GitHub returned unexpectedly large release metadata.")
92 payload_bytes = response.read(MAX_RELEASE_METADATA_BYTES + 1)
93 except HTTPError as error:
94 LOGGER.error("GitHub release check returned status=%d", error.code)
95 raise UpdateError(f"GitHub release check returned HTTP {error.code}.") from error
96 except (URLError, TimeoutError, OSError) as error:
97 reason = getattr(error, "reason", error)
98 LOGGER.warning("Could not check GitHub for updates: %s", reason)
99 raise UpdateError(f"Could not check GitHub for updates: {reason}") from error
100 if len(payload_bytes) > MAX_RELEASE_METADATA_BYTES:
101 raise UpdateError("GitHub returned unexpectedly large release metadata.")
102 try:
103 payload = json.loads(payload_bytes)
104 except (UnicodeError, json.JSONDecodeError) as error:
105 raise UpdateError("GitHub returned invalid release metadata.") from error
106 if not isinstance(payload, dict):
107 raise UpdateError("GitHub returned invalid release metadata.")
108 tag = payload.get("tag_name")
109 if not isinstance(tag, str):
110 raise UpdateError("The latest GitHub release has no semantic version tag.")
111 remote = _parse_version(tag, "latest GitHub release tag")
112 if remote <= current:
113 LOGGER.info("No newer application release is available")
114 return None
115 version = ".".join(str(part) for part in remote)
116 asset_name = target.release_asset_name(version)
117 assets = payload.get("assets")
118 if not isinstance(assets, list):
119 raise UpdateError("The latest GitHub release has no downloadable assets.")
120 for asset in assets:
121 if not isinstance(asset, dict) or asset.get("name") != asset_name:
122 continue
123 asset_url = asset.get("browser_download_url")
124 asset_size = asset.get("size")
125 if not isinstance(asset_url, str) or not asset_url:
126 break
127 if not isinstance(asset_size, int) or isinstance(asset_size, bool) or asset_size < 0:
128 asset_size = None
129 if asset_size is not None and asset_size > MAX_UPDATE_ARCHIVE_BYTES:
130 raise UpdateError("The Pocket Harbor update bundle is unexpectedly large.")
131 LOGGER.info("Application update available version=%s size=%s", version, asset_size)
132 return ReleaseUpdate(version, tag, asset_name, asset_url, asset_size)
133 raise UpdateError(f"The latest release does not contain {asset_name}.")
136def stage_update(
137 release: ReleaseUpdate,
138 install_directory: Path,
139 timeout_seconds: float = 30.0,
140 progress: ProgressCallback | None = None,
141 cancelled: CancelCallback | None = None,
142 target: LinuxTarget = DARKOS,
143) -> Path:
144 """Download and validate a bundle for the launcher to install after exit."""
146 install_directory = install_directory.expanduser().resolve()
147 LOGGER.info("Staging application update version=%s", release.version)
148 executable = install_directory / target.executable_name
149 if install_directory.name != target.application_directory or not executable.is_file():
150 raise UpdateError(
151 f"Automatic updates require the self-contained {target.display_name} package."
152 )
153 tools_directory = install_directory.parent
154 pending_name = target.update_staging_directory
155 pending = tools_directory / pending_name
156 incomplete = tools_directory / f"{pending_name}.incomplete"
157 archive = tools_directory / f"{pending_name}.zip.part"
158 _remove_staging_path(incomplete)
159 archive.unlink(missing_ok=True)
160 try:
161 _raise_if_cancelled(cancelled)
162 _download_release(release, archive, timeout_seconds, progress, cancelled)
163 _raise_if_cancelled(cancelled)
164 incomplete.mkdir(parents=False)
165 _extract_bundle(archive, incomplete, cancelled, target)
166 _validate_staged_bundle(incomplete, release.version, target)
167 _remove_staging_path(pending)
168 os.replace(incomplete, pending)
169 except UpdateCancelled, UpdateError:
170 _remove_staging_path(incomplete)
171 LOGGER.info("Application update staging stopped")
172 raise
173 except OSError as error:
174 _remove_staging_path(incomplete)
175 LOGGER.error("Could not stage application update: %s", error)
176 raise UpdateError(f"Could not stage the application update: {error}") from error
177 except BaseException:
178 _remove_staging_path(incomplete)
179 raise
180 finally:
181 archive.unlink(missing_ok=True)
182 LOGGER.info("Application update staged successfully version=%s", release.version)
183 return pending
186def _parse_version(value: str, description: str) -> tuple[int, int, int]:
187 match = _VERSION_PATTERN.fullmatch(value.strip())
188 if match is None:
189 raise UpdateError(f"The {description} {value!r} is not a stable semantic version.")
190 return (int(match.group(1)), int(match.group(2)), int(match.group(3)))
193def _download_release(
194 release: ReleaseUpdate,
195 destination: Path,
196 timeout_seconds: float,
197 progress: ProgressCallback | None,
198 cancelled: CancelCallback | None,
199) -> None:
200 request = Request(release.asset_url, headers={"User-Agent": USER_AGENT})
201 try:
202 with urlopen(
203 request, timeout=timeout_seconds, context=ssl.create_default_context()
204 ) as response:
205 length_header = response.headers.get("Content-Length")
206 total = int(length_header) if length_header and length_header.isdigit() else None
207 if total is None:
208 total = release.asset_size
209 if total is not None and total > MAX_UPDATE_ARCHIVE_BYTES:
210 raise UpdateError("The Pocket Harbor update bundle is unexpectedly large.")
211 downloaded = 0
212 try:
213 with destination.open("wb") as output:
214 while True:
215 _raise_if_cancelled(cancelled)
216 chunk = response.read(1024 * 256)
217 if not chunk:
218 break
219 downloaded += len(chunk)
220 if downloaded > MAX_UPDATE_ARCHIVE_BYTES:
221 raise UpdateError(
222 "The Pocket Harbor update bundle is unexpectedly large."
223 )
224 output.write(chunk)
225 if progress is not None:
226 progress(release.asset_name, downloaded, total)
227 _raise_if_cancelled(cancelled)
228 except BaseException:
229 destination.unlink(missing_ok=True)
230 raise
231 except HTTPError as error:
232 LOGGER.error("Update download returned status=%d", error.code)
233 raise UpdateError(f"Update download returned HTTP {error.code}.") from error
234 except (URLError, TimeoutError, OSError) as error:
235 reason = getattr(error, "reason", error)
236 LOGGER.warning("Could not download application update: %s", reason)
237 raise UpdateError(f"Could not download the update: {reason}") from error
240def _extract_bundle(
241 archive: Path,
242 destination: Path,
243 cancelled: CancelCallback | None,
244 target: LinuxTarget = DARKOS,
245) -> None:
246 try:
247 with zipfile.ZipFile(archive) as bundle:
248 files = [item for item in bundle.infolist() if not item.is_dir()]
249 if len(files) > MAX_UPDATE_FILES:
250 raise UpdateError("The update bundle contains too many files.")
251 total_size = sum(item.file_size for item in files)
252 if total_size > MAX_UPDATE_ARCHIVE_BYTES:
253 raise UpdateError("The expanded update bundle is unexpectedly large.")
254 for item in files:
255 _raise_if_cancelled(cancelled)
256 relative = _safe_bundle_path(item, target)
257 output_path = destination.joinpath(*relative.parts)
258 output_path.parent.mkdir(parents=True, exist_ok=True)
259 with bundle.open(item) as source, output_path.open("wb") as output:
260 while chunk := source.read(1024 * 256):
261 _raise_if_cancelled(cancelled)
262 output.write(chunk)
263 except (zipfile.BadZipFile, zipfile.LargeZipFile, OSError) as error:
264 raise UpdateError(f"The downloaded update bundle is invalid: {error}") from error
267def _safe_bundle_path(
268 item: zipfile.ZipInfo,
269 target: LinuxTarget = DARKOS,
270) -> PurePosixPath:
271 path = PurePosixPath(item.filename)
272 mode = item.external_attr >> 16
273 if (
274 path.is_absolute()
275 or not path.parts
276 or path.parts[0] != target.tools_directory
277 or len(path.parts) == 1
278 or any(part in ("", ".", "..") for part in path.parts)
279 or stat.S_ISLNK(mode)
280 ):
281 raise UpdateError(f"Unsafe path in update bundle: {item.filename!r}.")
282 return PurePosixPath(*path.parts[1:])
285def _validate_staged_bundle(
286 destination: Path,
287 version: str,
288 target: LinuxTarget = DARKOS,
289) -> None:
290 launcher = destination / target.launcher_name
291 executable = destination / target.application_directory / target.executable_name
292 if not launcher.is_file() or launcher.is_symlink():
293 raise UpdateError("The update bundle is missing the Pocket Harbor Tools launcher.")
294 if not executable.is_file() or executable.is_symlink():
295 raise UpdateError("The update bundle is missing the ARM64 application executable.")
296 try:
297 with executable.open("rb") as stream:
298 elf_header = stream.read(20)
299 except OSError as error:
300 raise UpdateError(f"Could not inspect the staged executable: {error}") from error
301 if (
302 len(elf_header) < 20
303 or elf_header[:4] != b"\x7fELF"
304 or elf_header[4] != target.elf_class
305 or elf_header[5] != 1
306 or int.from_bytes(elf_header[18:20], "little") != target.elf_machine
307 ):
308 raise UpdateError(
309 f"The update bundle executable is not Linux {target.architecture.upper()}."
310 )
311 launcher.chmod(launcher.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
312 executable.chmod(executable.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
313 (destination / READY_MARKER).write_text(version + "\n", encoding="ascii")
316def _remove_staging_path(path: Path) -> None:
317 if path.is_dir() and not path.is_symlink():
318 shutil.rmtree(path)
319 else:
320 path.unlink(missing_ok=True)
323def _raise_if_cancelled(cancelled: CancelCallback | None) -> None:
324 if cancelled is not None and cancelled():
325 raise UpdateCancelled("Application update cancelled.")