Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/tuf/ngclient/updater.py: 16%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# Copyright 2020, New York University and the TUF contributors
2# SPDX-License-Identifier: MIT OR Apache-2.0
4"""Client update workflow implementation.
6The ``Updater`` class provides an implementation of the
7`TUF client workflow
8<https://theupdateframework.github.io/specification/latest/#detailed-client-workflow>`_.
9``Updater`` provides an API to query available targets and to download them in a
10secure manner: All downloaded files are verified by signed metadata.
12High-level description of ``Updater`` functionality:
13 * Initializing an ``Updater`` loads and validates the trusted local root
14 metadata: This root metadata is used as the source of trust for all other
15 metadata. Updater should always be initialized with the ``bootstrap``
16 argument: pass ``bootstrap=None`` only to explicitly opt into using the
17 cached root.json as the trust anchor.
18 * ``refresh()`` can optionally be called to update and load all top-level
19 metadata as described in the specification, using both locally cached
20 metadata and metadata downloaded from the remote repository. If refresh is
21 not done explicitly, it will happen automatically during the first target
22 info lookup.
23 * ``Updater`` can be used to download targets. For each target:
25 * ``Updater.get_targetinfo()`` is first used to find information about a
26 specific target. This will load new targets metadata as needed (from
27 local cache or remote repository).
28 * ``Updater.find_cached_target()`` can optionally be used to check if a
29 target file is already locally cached.
30 * ``Updater.download_target()`` downloads a target file and ensures it is
31 verified correct by the metadata.
33Note that applications using ``Updater`` should be 'single instance'
34applications: running multiple instances that use the same cache directories at
35the same time is not supported.
37A simple example of using the Updater to implement a Python TUF client that
38downloads target files is available in `examples/client
39<https://github.com/theupdateframework/python-tuf/tree/develop/examples/client>`_.
41Notes on how Updater uses HTTP by default:
42 * urllib3 is the HTTP library
43 * Typically all requests are retried by urllib3 three times (in cases where
44 this seems useful)
45 * Operating system certificate store is used for TLS, in other words
46 ``certifi`` is not used as the certificate source
47 * Proxy use can be configured with ``https_proxy`` and other similar
48 environment variables
50All of the HTTP decisions can be changed with ``fetcher`` argument:
51Custom ``FetcherInterface`` implementations are possible. The alternative
52``RequestsFetcher`` implementation is also provided (although deprecated).
53"""
55from __future__ import annotations
57import contextlib
58import logging
59import os
60import shutil
61import tempfile
62from pathlib import Path
63from typing import TYPE_CHECKING, cast
64from urllib import parse
66from tuf.api import exceptions
67from tuf.api.metadata import Root, Snapshot, TargetFile, Targets, Timestamp
68from tuf.ngclient._internal.trusted_metadata_set import TrustedMetadataSet
69from tuf.ngclient.config import EnvelopeType, UpdaterConfig
70from tuf.ngclient.urllib3_fetcher import Urllib3Fetcher
72if TYPE_CHECKING:
73 from tuf.ngclient.fetcher import FetcherInterface
75logger = logging.getLogger(__name__)
78class Updater:
79 """Creates a new ``Updater`` instance and loads trusted root metadata.
81 Args:
82 metadata_dir: Local metadata directory. Directory must be
83 writable. If ``bootstrap`` is ``None``, this directory must contain
84 a trusted root.json file.
85 metadata_base_url: Base URL for all remote metadata downloads
86 target_dir: Local targets directory. Directory must be writable. It
87 will be used as the default target download directory by
88 ``find_cached_target()`` and ``download_target()``
89 target_base_url: ``Optional``; Default base URL for all remote target
90 downloads. Can be individually set in ``download_target()``
91 fetcher: ``Optional``; ``FetcherInterface`` implementation used to
92 download both metadata and targets. Default is ``Urllib3Fetcher``
93 config: ``Optional``; ``UpdaterConfig`` could be used to setup common
94 configuration options.
95 bootstrap: Initial root metadata bytes. This argument is required.
96 Pass the embedded root metadata bytes for secure initialization.
97 Pass ``None`` only if you explicitly want to use the cached
98 root.json as the trust anchor (not recommended for most
99 deployments).
101 Raises:
102 OSError: Local root.json cannot be read
103 RepositoryError: Local root.json is invalid
104 """
106 def __init__( # noqa: PLR0917
107 self,
108 metadata_dir: str,
109 metadata_base_url: str,
110 target_dir: str | None = None,
111 target_base_url: str | None = None,
112 fetcher: FetcherInterface | None = None,
113 config: UpdaterConfig | None = None,
114 *,
115 bootstrap: bytes | None,
116 ):
117 self._dir = metadata_dir
118 self._metadata_base_url = _ensure_trailing_slash(metadata_base_url)
119 self.target_dir = target_dir
120 if target_base_url is None:
121 self._target_base_url = None
122 else:
123 self._target_base_url = _ensure_trailing_slash(target_base_url)
125 self.config = config or UpdaterConfig()
126 if fetcher is not None:
127 self._fetcher = fetcher
128 else:
129 self._fetcher = Urllib3Fetcher(
130 app_user_agent=self.config.app_user_agent
131 )
132 supported_envelopes = [EnvelopeType.METADATA, EnvelopeType.SIMPLE]
133 if self.config.envelope_type not in supported_envelopes:
134 raise ValueError(
135 f"config: envelope_type must be one of {supported_envelopes}, "
136 f"got '{self.config.envelope_type}'"
137 )
139 if bootstrap is None:
140 # if no root was provided, use the cached non-versioned root.json
141 bootstrap = self._load_local_metadata(Root.type)
143 # Load the initial root, make sure it's cached
144 self._trusted_set = TrustedMetadataSet(
145 bootstrap, self.config.envelope_type
146 )
147 self._persist_root(self._trusted_set.root.version, bootstrap)
148 self._update_root_symlink()
150 def refresh(self) -> None:
151 """Refresh top-level metadata.
153 Downloads, verifies, and loads metadata for the top-level roles in the
154 specified order (root -> timestamp -> snapshot -> targets) implementing
155 all the checks required in the TUF client workflow.
157 A ``refresh()`` can be done only once during the lifetime of an Updater.
158 If ``refresh()`` has not been explicitly called before the first
159 ``get_targetinfo()`` call, it will be done implicitly at that time.
161 The metadata for delegated roles is not updated by ``refresh()``:
162 that happens on demand during ``get_targetinfo()``. However, if the
163 repository uses `consistent_snapshot
164 <https://theupdateframework.github.io/specification/latest/#consistent-snapshots>`_,
165 then all metadata downloaded by the Updater will use the same consistent
166 repository state.
168 Raises:
169 OSError: New metadata could not be written to disk
170 RepositoryError: Metadata failed to verify in some way
171 DownloadError: Download of a metadata file failed in some way
172 """
174 self._load_root()
175 self._load_timestamp()
176 self._load_snapshot()
177 self._load_targets(Targets.type, Root.type)
179 def _generate_target_file_path(self, targetinfo: TargetFile) -> str:
180 if self.target_dir is None:
181 raise ValueError("target_dir must be set if filepath is not given")
183 # Use URL encoded target path as filename
184 filename = parse.quote(targetinfo.path, "")
185 return os.path.join(self.target_dir, filename)
187 def get_targetinfo(self, target_path: str) -> TargetFile | None:
188 """Return ``TargetFile`` instance with information for ``target_path``.
190 The return value can be used as an argument to
191 ``download_target()`` and ``find_cached_target()``.
193 If ``refresh()`` has not been called before calling
194 ``get_targetinfo()``, the refresh will be done implicitly.
196 As a side-effect this method downloads all the additional (delegated
197 targets) metadata it needs to return the target information.
199 Args:
200 target_path: `path-relative-URL string
201 <https://url.spec.whatwg.org/#path-relative-url-string>`_
202 that uniquely identifies the target within the repository.
204 Raises:
205 OSError: New metadata could not be written to disk
206 RepositoryError: Metadata failed to verify in some way
207 DownloadError: Download of a metadata file failed in some way
209 Returns:
210 ``TargetFile`` instance or ``None``.
211 """
213 if Targets.type not in self._trusted_set:
214 self.refresh()
215 return self._preorder_depth_first_walk(target_path)
217 def find_cached_target(
218 self,
219 targetinfo: TargetFile,
220 filepath: str | None = None,
221 ) -> str | None:
222 """Check whether a local file is an up to date target.
224 Args:
225 targetinfo: ``TargetFile`` from ``get_targetinfo()``.
226 filepath: Local path to file. If ``None``, a file path is
227 generated based on ``target_dir`` constructor argument.
229 Raises:
230 ValueError: Incorrect arguments
232 Returns:
233 Local file path if the file is an up to date target file.
234 ``None`` if file is not found or it is not up to date.
235 """
237 if filepath is None:
238 filepath = self._generate_target_file_path(targetinfo)
240 try:
241 with open(filepath, "rb") as target_file:
242 targetinfo.verify_length_and_hashes(target_file)
243 return filepath
244 except (OSError, exceptions.LengthOrHashMismatchError):
245 return None
247 def download_target(
248 self,
249 targetinfo: TargetFile,
250 filepath: str | None = None,
251 target_base_url: str | None = None,
252 ) -> str:
253 """Download the target file specified by ``targetinfo``.
255 Args:
256 targetinfo: ``TargetFile`` from ``get_targetinfo()``.
257 filepath: Local path to download into. If ``None``, the file is
258 downloaded into directory defined by ``target_dir`` constructor
259 argument using a generated filename. If file already exists,
260 it is overwritten.
261 target_base_url: Base URL used to form the final target
262 download URL. Default is the value provided in ``Updater()``
264 Raises:
265 ValueError: Invalid arguments
266 DownloadError: Download of the target file failed in some way
267 RepositoryError: Downloaded target failed to be verified in some way
268 OSError: Failed to write target to file
270 Returns:
271 Local path to downloaded file
272 """
274 if filepath is None:
275 filepath = self._generate_target_file_path(targetinfo)
276 Path(filepath).parent.mkdir(exist_ok=True, parents=True)
278 if target_base_url is None:
279 if self._target_base_url is None:
280 raise ValueError(
281 "target_base_url must be set in either "
282 "download_target() or constructor"
283 )
285 target_base_url = self._target_base_url
286 else:
287 target_base_url = _ensure_trailing_slash(target_base_url)
289 target_filepath = targetinfo.path
290 consistent_snapshot = self._trusted_set.root.consistent_snapshot
291 if consistent_snapshot and self.config.prefix_targets_with_hash:
292 hashes = list(targetinfo.hashes.values())
293 dirname, sep, basename = target_filepath.rpartition("/")
294 target_filepath = f"{dirname}{sep}{hashes[0]}.{basename}"
295 full_url = f"{target_base_url}{target_filepath}"
297 with self._fetcher.download_file(
298 full_url, targetinfo.length
299 ) as target_file:
300 targetinfo.verify_length_and_hashes(target_file)
302 target_file.seek(0)
303 with open(filepath, "wb") as destination_file:
304 shutil.copyfileobj(target_file, destination_file)
306 logger.debug("Downloaded target %s", targetinfo.path)
307 return filepath
309 def _download_metadata(
310 self, rolename: str, length: int, version: int | None = None
311 ) -> bytes:
312 """Download a metadata file and return it as bytes."""
313 encoded_name = parse.quote(rolename, "")
314 if version is None:
315 url = f"{self._metadata_base_url}{encoded_name}.json"
316 else:
317 url = f"{self._metadata_base_url}{version}.{encoded_name}.json"
318 return self._fetcher.download_bytes(url, length)
320 def _load_local_metadata(self, rolename: str) -> bytes:
321 encoded_name = parse.quote(rolename, "")
322 with open(os.path.join(self._dir, f"{encoded_name}.json"), "rb") as f:
323 return f.read()
325 def _persist_metadata(self, rolename: str, data: bytes) -> None:
326 """Write metadata to disk atomically to avoid data loss.
328 Use a filename _not_ prefixed with version (e.g. "timestamp.json")
329 . Encode the rolename to avoid issues with e.g. path separators
330 """
332 encoded_name = parse.quote(rolename, "")
333 filename = os.path.join(self._dir, f"{encoded_name}.json")
334 self._persist_file(filename, data)
336 def _persist_root(self, version: int, data: bytes) -> None:
337 """Write root metadata to disk atomically to avoid data loss.
339 The metadata is stored with version prefix (e.g.
340 "root_history/1.root.json").
341 """
342 rootdir = Path(self._dir, "root_history")
343 rootdir.mkdir(exist_ok=True, parents=True)
344 self._persist_file(str(rootdir / f"{version}.root.json"), data)
346 def _persist_file(self, filename: str, data: bytes) -> None:
347 """Write a file to disk atomically to avoid data loss."""
348 temp_file_name = None
350 try:
351 with tempfile.NamedTemporaryFile(
352 dir=self._dir, delete=False
353 ) as temp_file:
354 temp_file_name = temp_file.name
355 temp_file.write(data)
356 os.replace(temp_file.name, filename)
357 except OSError as e:
358 # remove tempfile if we managed to create one,
359 # then let the exception happen
360 if temp_file_name is not None:
361 with contextlib.suppress(FileNotFoundError):
362 os.remove(temp_file_name)
363 raise e
365 def _update_root_symlink(self) -> None:
366 """Symlink root.json to current trusted root version in root_history/"""
367 linkname = os.path.join(self._dir, "root.json")
368 version = self._trusted_set.root.version
369 current = os.path.join("root_history", f"{version}.root.json")
370 with contextlib.suppress(FileNotFoundError):
371 os.remove(linkname)
372 try:
373 os.symlink(current, linkname)
374 except OSError as e:
375 if getattr(e, "winerror", None) == 1314:
376 # Fallback for NTFS "required privilege is not held by client"
377 os.link(os.path.join(self._dir, current), linkname)
378 else:
379 raise
381 def _load_root(self) -> None:
382 """Load root metadata.
384 Sequentially load newer root metadata versions. First try to load from
385 local cache and if that does not work, from the remote repository.
387 If metadata is loaded from remote repository, store it in local cache.
388 """
390 # Update the root role
391 lower_bound = self._trusted_set.root.version + 1
392 upper_bound = lower_bound + self.config.max_root_rotations
394 try:
395 for next_version in range(lower_bound, upper_bound):
396 # look for next_version in local cache
397 try:
398 root_path = os.path.join(
399 self._dir, "root_history", f"{next_version}.root.json"
400 )
401 with open(root_path, "rb") as f:
402 self._trusted_set.update_root(f.read())
403 continue
404 except (OSError, exceptions.RepositoryError) as e:
405 # this root did not exist locally or is invalid
406 logger.debug("Local root is not valid: %s", e)
408 # next_version was not found locally, try remote
409 try:
410 data = self._download_metadata(
411 Root.type,
412 self.config.root_max_length,
413 next_version,
414 )
415 self._trusted_set.update_root(data)
416 self._persist_root(next_version, data)
418 except exceptions.DownloadHTTPError as exception:
419 if exception.status_code not in {403, 404}:
420 raise
421 # 404/403 means current root is newest available
422 break
423 finally:
424 # Make sure the non-versioned root.json links to current version
425 self._update_root_symlink()
427 def _load_timestamp(self) -> None:
428 """Load local and remote timestamp metadata."""
429 try:
430 data = self._load_local_metadata(Timestamp.type)
431 self._trusted_set.update_timestamp(data)
432 except (OSError, exceptions.RepositoryError) as e:
433 # Local timestamp does not exist or is invalid
434 logger.debug("Local timestamp not valid as final: %s", e)
436 # Load from remote (whether local load succeeded or not)
437 data = self._download_metadata(
438 Timestamp.type, self.config.timestamp_max_length
439 )
440 try:
441 self._trusted_set.update_timestamp(data)
442 except exceptions.EqualVersionNumberError:
443 # If the new timestamp version is the same as current, discard the
444 # new timestamp. This is normal and it shouldn't raise any error.
445 return
447 self._persist_metadata(Timestamp.type, data)
449 def _load_snapshot(self) -> None:
450 """Load local (and if needed remote) snapshot metadata."""
451 try:
452 data = self._load_local_metadata(Snapshot.type)
453 self._trusted_set.update_snapshot(data, trusted=True)
454 logger.debug("Local snapshot is valid: not downloading new one")
455 except (OSError, exceptions.RepositoryError) as e:
456 # Local snapshot does not exist or is invalid: update from remote
457 logger.debug("Local snapshot not valid as final: %s", e)
459 snapshot_meta = self._trusted_set.timestamp.snapshot_meta
460 length = snapshot_meta.length or self.config.snapshot_max_length
461 version = None
462 if self._trusted_set.root.consistent_snapshot:
463 version = snapshot_meta.version
465 data = self._download_metadata(Snapshot.type, length, version)
466 self._trusted_set.update_snapshot(data)
467 self._persist_metadata(Snapshot.type, data)
469 def _load_targets(self, role: str, parent_role: str) -> Targets:
470 """Load local (and if needed remote) metadata for ``role``."""
472 # Avoid loading 'role' more than once for the same parent_role
473 if self._trusted_set.contains(role, parent_role):
474 return cast("Targets", self._trusted_set[role])
476 try:
477 data = self._load_local_metadata(role)
478 delegated_targets = self._trusted_set.update_delegated_targets(
479 data, role, parent_role
480 )
481 logger.debug("Local %s is valid: not downloading new one", role)
482 return delegated_targets
483 except (OSError, exceptions.RepositoryError) as e:
484 # Local 'role' does not exist or is invalid: update from remote
485 logger.debug("Failed to load local %s: %s", role, e)
487 snapshot = self._trusted_set.snapshot
488 metainfo = snapshot.meta.get(f"{role}.json")
489 if metainfo is None:
490 raise exceptions.RepositoryError(
491 f"Role {role} was delegated but is not part of snapshot"
492 ) from None
494 length = metainfo.length or self.config.targets_max_length
495 version = None
496 if self._trusted_set.root.consistent_snapshot:
497 version = metainfo.version
499 data = self._download_metadata(role, length, version)
500 delegated_targets = self._trusted_set.update_delegated_targets(
501 data, role, parent_role
502 )
503 self._persist_metadata(role, data)
505 return delegated_targets
507 def _preorder_depth_first_walk(
508 self, target_filepath: str
509 ) -> TargetFile | None:
510 """
511 Interrogates the tree of target delegations in order of appearance
512 (which implicitly order trustworthiness), and returns the matching
513 target found in the most trusted role.
514 """
516 # List of delegations to be interrogated. A (role, parent role) pair
517 # is needed to load and verify the delegated targets metadata.
518 delegations_to_visit = [(Targets.type, Root.type)]
519 visited_role_names: set[str] = set()
521 # Preorder depth-first traversal of the graph of target delegations.
522 while (
523 len(visited_role_names) <= self.config.max_delegations
524 and len(delegations_to_visit) > 0
525 ):
526 # Pop the role name from the top of the stack.
527 role_name, parent_role = delegations_to_visit.pop(-1)
529 # Skip any visited current role to prevent cycles.
530 if role_name in visited_role_names:
531 logger.debug("Skipping visited current role %s", role_name)
532 continue
534 # The metadata for 'role_name' must be downloaded/updated before
535 # its targets, delegations, and child roles can be inspected.
536 targets = self._load_targets(role_name, parent_role)
538 target = targets.targets.get(target_filepath)
540 if target is not None:
541 logger.debug("Found target in current role %s", role_name)
542 return target
544 # After preorder check, add current role to set of visited roles.
545 visited_role_names.add(role_name)
547 if targets.delegations is not None:
548 child_roles_to_visit = []
549 # NOTE: This may be a slow operation if there are many
550 # delegated roles.
551 for (
552 child_name,
553 terminating,
554 ) in targets.delegations.get_roles_for_target(target_filepath):
555 logger.debug("Adding child role %s", child_name)
556 child_roles_to_visit.append((child_name, role_name))
557 if terminating:
558 logger.debug("Not backtracking to other roles")
559 delegations_to_visit = []
560 break
561 # Push 'child_roles_to_visit' in reverse order of appearance
562 # onto 'delegations_to_visit'. Roles are popped from the end of
563 # the list.
564 child_roles_to_visit.reverse()
565 delegations_to_visit.extend(child_roles_to_visit)
567 if len(delegations_to_visit) > 0:
568 logger.debug(
569 "%d roles left to visit, but allowed at most %d delegations",
570 len(delegations_to_visit),
571 self.config.max_delegations,
572 )
574 # If this point is reached then target is not found, return None
575 return None
578def _ensure_trailing_slash(url: str) -> str:
579 """Return url guaranteed to end in a slash."""
580 return url if url.endswith("/") else f"{url}/"