Coverage for src/ezcompiler/services/tuf_service.py: 73.31%
234 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-16 16:37 +0000
1# ///////////////////////////////////////////////////////////////
2# TUF_SERVICE - Local TUF tree management (status, withdrawal)
3# Project: ezcompiler
4# ///////////////////////////////////////////////////////////////
6"""
7TUF service - Reads and manages the local signed TUF tree.
9Reads metadata as plain JSON (no tufup, no keys), keeps the list of
10withdrawn versions in ``<repo>/withdrawn.json`` and enforces the rule that
11a new release must be higher than every withdrawn version: clients that
12installed a withdrawn version only leave it for a higher one.
13"""
15from __future__ import annotations
17# ///////////////////////////////////////////////////////////////
18# IMPORTS
19# ///////////////////////////////////////////////////////////////
20# Standard library imports
21import json
22import logging
23import os
24import re
25import shutil
26from datetime import UTC, datetime
27from pathlib import Path
28from typing import TYPE_CHECKING, Any, Literal, overload
30# Third-party imports
31from packaging.version import InvalidVersion, Version
33# Local imports
34from .._types import ReleaserPort
35from ..adapters import ReleaserFactory
36from ..shared import TufStatus, TufVersion
37from ..shared._constants import TUF_PUBLIC_DIRS
38from ..shared.exceptions import ReleaseError
40if TYPE_CHECKING: 40 ↛ 41line 40 didn't jump to line 41 because the condition on line 40 was never true
41 from ..shared import CompilerConfig
43# ///////////////////////////////////////////////////////////////
44# CONSTANTS
45# ///////////////////////////////////////////////////////////////
47WITHDRAWN_FILE = "withdrawn.json"
48# Timestamped backups of the signed tree, taken before any withdrawal.
49BACKUP_DIR = ".backup"
50TUF_ROLES = ("root", "targets", "snapshot", "timestamp")
52logger = logging.getLogger(__name__)
54# ///////////////////////////////////////////////////////////////
55# CLASSES
56# ///////////////////////////////////////////////////////////////
59class TufService:
60 """Local TUF tree operations shared by the CLI and the release path."""
62 # ////////////////////////////////////////////////
63 # PATHS
64 # ////////////////////////////////////////////////
66 @staticmethod
67 def repo_dir(config: CompilerConfig) -> Path:
68 """Local TUF repository directory."""
69 return config.tuf_repo_dir or (config.output_folder / "repo")
71 @staticmethod
72 def keys_dir(config: CompilerConfig) -> Path:
73 """Private signing keys directory."""
74 return config.tuf_keys_dir or (TufService.repo_dir(config) / "keystore")
76 # ////////////////////////////////////////////////
77 # WITHDRAWN VERSIONS
78 # ////////////////////////////////////////////////
80 @staticmethod
81 def withdrawn_versions(repo_dir: Path) -> list[str]:
82 """Versions removed from the tree, in withdrawal order.
84 Raises:
85 ReleaseError: If ``withdrawn.json`` exists but is malformed — a
86 silently ignored file would disable the version guard.
87 """
88 path = repo_dir / WITHDRAWN_FILE
89 if not path.is_file():
90 return []
91 try:
92 doc = json.loads(path.read_text(encoding="utf-8"))
93 entries = doc["withdrawn"]
94 if not isinstance(entries, list):
95 raise TypeError("'withdrawn' is not a list")
96 versions = [entry["version"] for entry in entries]
97 if not all(isinstance(v, str) for v in versions): 97 ↛ 98line 97 didn't jump to line 98 because the condition on line 97 was never true
98 raise TypeError("non-string version")
99 except (OSError, ValueError, KeyError, TypeError) as e:
100 raise ReleaseError(f"{path} is unreadable: {e}") from e
101 return versions
103 @staticmethod
104 def record_withdrawn(
105 repo_dir: Path, version: str, *, now: datetime | None = None
106 ) -> None:
107 """Append ``version`` to ``withdrawn.json``.
109 The file is written atomically (temporary file in the same directory,
110 then ``os.replace``): an interrupted write never truncates it.
112 Raises:
113 ReleaseError: If the existing file is malformed, or if the write
114 fails — the message names the version to add by hand.
115 """
116 path = repo_dir / WITHDRAWN_FILE
117 entries: list[dict[str, Any]] = [
118 {"version": v, "withdrawn_at": at}
119 for v, at in TufService._withdrawn_entries(path)
120 ]
121 stamp = (now or datetime.now(UTC)).strftime("%Y-%m-%dT%H:%M:%SZ")
122 entries.append({"version": version, "withdrawn_at": stamp})
123 tmp = path.with_name(f"{WITHDRAWN_FILE}.tmp")
124 try:
125 tmp.write_text(
126 json.dumps({"withdrawn": entries}, indent=2) + "\n", encoding="utf-8"
127 )
128 os.replace(tmp, path)
129 except OSError as e:
130 tmp.unlink(missing_ok=True)
131 raise ReleaseError(
132 f"Cannot record the withdrawal of {version} in {path}: "
133 f"{e}. Add {version} to that file by hand before any "
134 "new release."
135 ) from e
137 @staticmethod
138 def _withdrawn_entries(path: Path) -> list[tuple[str, str]]:
139 """Existing (version, withdrawn_at) pairs, validated."""
140 versions = TufService.withdrawn_versions(path.parent)
141 if not versions:
142 return []
143 doc = json.loads(path.read_text(encoding="utf-8"))
144 return [
145 (entry["version"], str(entry.get("withdrawn_at", "")))
146 for entry in doc["withdrawn"]
147 ]
149 @staticmethod
150 def ensure_releasable(repo_dir: Path, version: str) -> None:
151 """Refuse a version that is not above every withdrawn version.
153 Raises:
154 ReleaseError: If ``version`` <= a withdrawn version (PEP 440), or
155 equals a non-PEP 440 withdrawn string.
156 """
157 for withdrawn in TufService.withdrawn_versions(repo_dir):
158 if TufService._not_above(version, withdrawn):
159 raise ReleaseError(
160 f"Version {withdrawn} was withdrawn from the TUF tree: "
161 "clients that installed it will only accept a higher "
162 f"version. Use a version > {withdrawn} "
163 f"(requested: {version})."
164 )
166 @staticmethod
167 def _not_above(version: str, withdrawn: str) -> bool:
168 try:
169 return Version(version) <= Version(withdrawn)
170 except InvalidVersion:
171 return version == withdrawn
173 # ////////////////////////////////////////////////
174 # REMOVAL
175 # ////////////////////////////////////////////////
177 @staticmethod
178 def remove_latest(config: CompilerConfig, *, release_type: str = "tufup") -> str:
179 """Remove the latest version from the local tree and record it.
181 The signed tree (``metadata/``, ``targets/`` and ``withdrawn.json``) is
182 copied to a timestamped backup directory first, and the withdrawal is
183 recorded **before** the irreversible mutation: a releaser failure
184 (missing key, read-only keystore, expired role) can therefore never
185 leave a half-mutated tree with the withdrawal unrecorded. On failure the
186 three are restored from the backup, whose path is named in the error.
188 Returns:
189 str: The removed version.
191 Raises:
192 ReleaseError: If ``withdrawn.json`` is malformed or cannot be
193 written, if the tree holds no version, or if the backup cannot
194 be taken (nothing is removed in any of those cases).
195 ReleaseError / SigningKeyError: From the releaser; the tree and the
196 register are rolled back to their pre-removal state.
197 """
198 repo_dir = TufService.repo_dir(config)
199 keys_dir = TufService.keys_dir(config)
200 # Validate withdrawn.json BEFORE the irreversible withdrawal: an
201 # unreadable file would drop the version without recording it.
202 TufService.withdrawn_versions(repo_dir)
204 predicted = TufService.latest_tree_version(repo_dir, config.project_name)
205 if predicted is None:
206 raise ReleaseError(f"No version to withdraw in {repo_dir}.")
208 backup = TufService._backup_tree(repo_dir)
209 releaser: ReleaserPort = ReleaserFactory.create_releaser(
210 release_type,
211 {"keys_dir": keys_dir, "expiration_days": config.tuf_expiration_days},
212 )
213 TufService.record_withdrawn(repo_dir, predicted)
214 try:
215 removed = releaser.remove_latest(
216 app_name=config.project_name, repo_dir=repo_dir, keys_dir=keys_dir
217 )
218 except Exception as exc:
219 restored = TufService._restore_tree(repo_dir, backup)
220 detail = (
221 f"tree and register restored from {backup}"
222 if restored
223 else f"RESTORE FAILED: restore by hand from {backup}"
224 )
225 raise ReleaseError(
226 f"Withdrawal of {predicted} failed ({exc}) — {detail}."
227 ) from exc
228 if not TufService.same_version(removed, predicted):
229 # Unlikely divergence: also record the version actually
230 # withdrawn so the guard covers both spellings.
231 logger.warning(
232 "TUF removal mismatch: tree announced %s, releaser removed %s",
233 predicted,
234 removed,
235 )
236 TufService.record_withdrawn(repo_dir, removed)
237 logger.info("TUF version withdrawn: %s (sauvegarde : %s)", removed, backup)
238 return removed
240 @staticmethod
241 def _backup_tree(repo_dir: Path) -> Path:
242 """Copy the signed tree and the register to a timestamped directory.
244 Returns:
245 Path: The backup directory, under ``<repo_dir>/.backup``.
247 Raises:
248 ReleaseError: If the copy fails — nothing must be removed without
249 a restorable snapshot.
250 """
251 stamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ")
252 backup = repo_dir / BACKUP_DIR / f"remove-latest-{stamp}"
253 try:
254 backup.mkdir(parents=True, exist_ok=True)
255 for name in TUF_PUBLIC_DIRS:
256 source = repo_dir / name
257 if source.is_dir():
258 shutil.copytree(source, backup / name, dirs_exist_ok=True)
259 register = repo_dir / WITHDRAWN_FILE
260 if register.is_file():
261 shutil.copy2(register, backup / WITHDRAWN_FILE)
262 except OSError as e:
263 raise ReleaseError(
264 f"Cannot back up the TUF tree into {backup}: {e}. "
265 "No withdrawal was performed."
266 ) from e
267 return backup
269 @staticmethod
270 def _restore_tree(repo_dir: Path, backup: Path) -> bool:
271 """Put ``metadata/``, ``targets/`` and the register back from ``backup``.
273 Returns:
274 bool: True when the restore succeeded; the caller names the backup
275 directory in its error message when it did not.
276 """
277 try:
278 for name in TUF_PUBLIC_DIRS:
279 source = backup / name
280 if source.is_dir():
281 shutil.rmtree(repo_dir / name, ignore_errors=True)
282 shutil.copytree(source, repo_dir / name)
283 register = backup / WITHDRAWN_FILE
284 if register.is_file():
285 shutil.copy2(register, repo_dir / WITHDRAWN_FILE)
286 else:
287 (repo_dir / WITHDRAWN_FILE).unlink(missing_ok=True)
288 except OSError:
289 return False
290 return True
292 # ////////////////////////////////////////////////
293 # STATUS
294 # ////////////////////////////////////////////////
296 @staticmethod
297 def status(config: CompilerConfig) -> TufStatus:
298 """Read the local tree: versions, flags, expirations, withdrawals.
300 Raises:
301 ReleaseError: If the tree is not initialized or a metadata file
302 cannot be read.
303 """
304 repo_dir = TufService.repo_dir(config)
305 meta_dir = repo_dir / "metadata"
306 if not (meta_dir / "root.json").is_file():
307 raise ReleaseError(
308 f"No TUF tree initialized in {repo_dir} "
309 "(metadata/root.json missing). Run `ezcompiler tuf init`."
310 )
312 signed = {role: TufService._read_signed(meta_dir, role) for role in TUF_ROLES}
313 try:
314 expirations = {
315 role: TufService._parse_expires(signed[role]["expires"])
316 for role in TUF_ROLES
317 }
318 targets = signed["targets"].get("targets", {})
319 if not isinstance(targets, dict):
320 raise TypeError("'targets' is not an object")
321 versions = TufService._versions(config.project_name, targets)
322 except (KeyError, ValueError, AttributeError, TypeError) as e:
323 raise ReleaseError(f"Unreadable TUF metadata in {meta_dir}: {e}") from e
325 return TufStatus(
326 repo_dir=repo_dir,
327 versions=versions,
328 expirations=expirations,
329 withdrawn=tuple(TufService.withdrawn_versions(repo_dir)),
330 )
332 @staticmethod
333 def _parse_expires(raw: object) -> datetime:
334 """ISO 8601 expiry; a value without offset is read as UTC."""
335 expires = datetime.fromisoformat(str(raw).replace("Z", "+00:00"))
336 if expires.tzinfo is None:
337 expires = expires.replace(tzinfo=UTC)
338 return expires
340 @staticmethod
341 def _read_signed(meta_dir: Path, role: str) -> dict[str, Any]:
342 path = meta_dir / f"{role}.json"
343 try:
344 signed = json.loads(path.read_text(encoding="utf-8"))["signed"]
345 except (OSError, ValueError, KeyError, TypeError) as e:
346 raise ReleaseError(f"{path} is unreadable: {e}") from e
347 if not isinstance(signed, dict): 347 ↛ 348line 347 didn't jump to line 348 because the condition on line 347 was never true
348 raise ReleaseError(f"{path} is unreadable: 'signed' is not an object")
349 return signed
351 @staticmethod
352 def _versions(app_name: str, targets: dict[str, Any]) -> tuple[TufVersion, ...]:
353 archive = re.compile(rf"^{re.escape(app_name)}-(.+)\.tar\.gz$")
354 patch = re.compile(rf"^{re.escape(app_name)}-(.+)\.patch$")
355 patched = {m.group(1) for name in targets if (m := patch.match(name))}
356 found: list[tuple[Version, TufVersion]] = []
357 for name, info in targets.items():
358 match = archive.match(name)
359 if match is None:
360 continue
361 raw = match.group(1)
362 try:
363 parsed = Version(raw)
364 except InvalidVersion:
365 continue
366 if not isinstance(info, dict):
367 raise TypeError(f"entry {name!r} is not an object")
368 custom = info.get("custom") or {}
369 required = bool((custom.get("tufup") or {}).get("required", False))
370 found.append((parsed, TufVersion(raw, required, raw in patched)))
371 found.sort(key=lambda item: item[0], reverse=True)
372 return tuple(v for _, v in found)
374 # ////////////////////////////////////////////////
375 # TREE VERSION
376 # ////////////////////////////////////////////////
378 @overload
379 @staticmethod
380 def read_tree_version( 380 ↛ anywhereline 380 didn't jump anywhere: it always raised an exception.
381 config: CompilerConfig, *, allow_empty: Literal[False] = False
382 ) -> str: ...
384 @overload
385 @staticmethod
386 def read_tree_version( 386 ↛ anywhereline 386 didn't jump anywhere: it always raised an exception.
387 config: CompilerConfig, *, allow_empty: bool
388 ) -> str | None: ...
390 @staticmethod
391 def read_tree_version(
392 config: CompilerConfig, *, allow_empty: bool = False
393 ) -> str | None:
394 """Return the highest version signed in the local TUF tree.
396 The recap of ``publish update`` names this version, not
397 ``config.version``: a config bumped without a rebuild would otherwise
398 announce a version the tree does not carry.
400 Args:
401 config: Current configuration (repo dir and project name).
402 allow_empty: True after the only version was withdrawn: the
403 empty tree is publishable and ``None`` is then returned.
405 Returns:
406 str | None: Highest version among the ``<app>-<version>`` targets,
407 or ``None`` if ``allow_empty`` and the tree holds no archive.
409 Raises:
410 ReleaseError: If no signed ``targets.json`` exists or it names no
411 archive of the project.
412 """
413 repo_dir = TufService.repo_dir(config)
414 targets_meta = repo_dir / "metadata" / "targets.json"
415 if not targets_meta.is_file():
416 raise ReleaseError(
417 f"No signed TUF tree in {repo_dir} "
418 f"({targets_meta.name} missing). "
419 "Run the build pipeline first (`ezcompiler compile`)."
420 )
421 latest = TufService.latest_tree_version(repo_dir, config.project_name)
422 if latest is None and not allow_empty:
423 raise ReleaseError(
424 f"{targets_meta} references no archive of "
425 f"{config.project_name}. Run the build pipeline first."
426 )
427 return latest
429 @staticmethod
430 def latest_tree_version(repo_dir: Path, app_name: str) -> str | None:
431 """Highest ``<app>-<version>`` archive signed in ``repo_dir``.
433 Returns:
434 str | None: The raw version spelling tufup used, or ``None`` when
435 no signed ``targets.json`` exists or it names no archive.
437 Raises:
438 ReleaseError: If ``targets.json`` exists but cannot be read.
439 """
440 targets_meta = repo_dir / "metadata" / "targets.json"
441 if not targets_meta.is_file(): 441 ↛ 442line 441 didn't jump to line 442 because the condition on line 441 was never true
442 return None
443 try:
444 doc = json.loads(targets_meta.read_text(encoding="utf-8"))
445 names = doc["signed"]["targets"]
446 except (OSError, ValueError, KeyError, TypeError) as e:
447 raise ReleaseError(f"{targets_meta} is unreadable: {e}") from e
449 pattern = re.compile(rf"^{re.escape(app_name)}-(.+)\.tar\.gz$")
450 # tufup names the archive with the raw config string: return it as
451 # is (1.2.3-rc.1, not its normalized form 1.2.3rc1).
452 versions: list[tuple[Version, str]] = []
453 for name in names:
454 match = pattern.match(name)
455 if match is None:
456 continue
457 try:
458 versions.append((Version(match.group(1)), match.group(1)))
459 except InvalidVersion:
460 continue
461 return max(versions)[1] if versions else None
463 @staticmethod
464 def needs_full_archive(repo_dir: Path, app_name: str) -> bool:
465 """True for the first release after a withdrawal.
467 tufup builds the patch from the latest archive still in the tree.
468 Clients on a withdrawn version above it hold another archive: the
469 patch would fail its hash check, so the release ships without one.
470 """
471 withdrawn = TufService.withdrawn_versions(repo_dir)
472 if not withdrawn:
473 return False
474 latest = TufService.latest_tree_version(repo_dir, app_name)
475 if latest is None:
476 return True
477 return any(not TufService._not_above(w, latest) for w in withdrawn)
479 @staticmethod
480 def same_version(left: str, right: str) -> bool:
481 """Compare two version strings by PEP 440 meaning, not spelling.
483 ``1.2.3-rc.1`` and ``1.2.3rc1`` are the same version. Strings that
484 are not PEP 440 versions fall back to an exact comparison.
485 """
486 try:
487 return Version(left) == Version(right)
488 except InvalidVersion:
489 return left == right