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 2021-2026, the TUF contributors
2# SPDX-License-Identifier: MIT OR Apache-2.0
4"""Trusted collection of client-side TUF Metadata.
6``TrustedMetadataSet`` keeps track of the current valid set of metadata for the
7client, and handles almost every step of the "Detailed client workflow" (
8https://theupdateframework.github.io/specification/latest#detailed-client-workflow)
9in the TUF specification: the remaining steps are related to filesystem and
10network IO, which are not handled here.
12Loaded metadata can be accessed via index access with rolename as key
13(``trusted_set[Root.type]``) or, in the case of top-level metadata, using the
14helper properties (``trusted_set.root``).
16Signatures are verified and discarded upon inclusion into the trusted set.
18The rules that ``TrustedMetadataSet`` follows for top-level metadata are
19 * Metadata must be loaded in order:
20 root -> timestamp -> snapshot -> targets -> (delegated targets).
21 * Metadata can be loaded even if it is expired (or in the snapshot case if the
22 meta info does not match): this is called "intermediate metadata".
23 * Intermediate metadata can _only_ be used to load newer versions of the
24 same metadata: As an example an expired root can be used to load a new root.
25 * Metadata is loadable only if metadata before it in loading order is loaded
26 (and is not intermediate): As an example timestamp can be loaded if a
27 final (non-expired) root has been loaded.
28 * Metadata is not loadable if any metadata after it in loading order has been
29 loaded: As an example new roots cannot be loaded if timestamp is loaded.
31Exceptions are raised if metadata fails to load in any way.
33Example of loading root, timestamp and snapshot:
35>>> # Load local root (RepositoryErrors here stop the update)
36>>> with open(root_path, "rb") as f:
37>>> trusted_set = TrustedMetadataSet(f.read(), EnvelopeType.METADATA)
38>>>
39>>> # update root from remote until no more are available
40>>> with download(Root.type, trusted_set.root.version + 1) as f:
41>>> trusted_set.update_root(f.read())
42>>>
43>>> # load local timestamp, then update from remote
44>>> try:
45>>> with open(timestamp_path, "rb") as f:
46>>> trusted_set.update_timestamp(f.read())
47>>> except (RepositoryError, OSError):
48>>> pass # failure to load a local file is ok
49>>>
50>>> with download(Timestamp.type) as f:
51>>> trusted_set.update_timestamp(f.read())
52>>>
53>>> # load local snapshot, then update from remote if needed
54>>> try:
55>>> with open(snapshot_path, "rb") as f:
56>>> trusted_set.update_snapshot(f.read())
57>>> except (RepositoryError, OSError):
58>>> # local snapshot is not valid, load from remote
59>>> # (RepositoryErrors here stop the update)
60>>> with download(Snapshot.type, version) as f:
61>>> trusted_set.update_snapshot(f.read())
62"""
64from __future__ import annotations
66import datetime
67import logging
68from collections import abc
69from typing import TYPE_CHECKING, cast
71from tuf.api import exceptions
72from tuf.api.dsse import SimpleEnvelope
73from tuf.api.metadata import (
74 Metadata,
75 Root,
76 Signed,
77 Snapshot,
78 T,
79 Targets,
80 Timestamp,
81)
82from tuf.ngclient.config import EnvelopeType
84if TYPE_CHECKING:
85 from collections.abc import Iterator
87 from securesystemslib.signer import Signature
89logger = logging.getLogger(__name__)
91Delegator = Root | Targets
94class TrustedMetadataSet(abc.Mapping):
95 """Internal class to keep track of trusted metadata in ``Updater``.
97 ``TrustedMetadataSet`` ensures that the collection of metadata in it is
98 valid and trusted through the whole client update workflow. It provides
99 easy ways to update the metadata with the caller making decisions on
100 what is updated.
101 """
103 def __init__(self, root_data: bytes, envelope_type: EnvelopeType):
104 """Initialize ``TrustedMetadataSet`` by loading trusted root metadata.
106 Args:
107 root_data: Trusted root metadata as bytes. Note that this metadata
108 will only be verified by itself: it is the source of trust for
109 all metadata in the ``TrustedMetadataSet``
110 envelope_type: Configures deserialization and verification mode of
111 TUF metadata.
113 Raises:
114 RepositoryError: Metadata failed to load or verify. The actual
115 error type and content will contain more details.
116 """
117 self._trusted_set: dict[str, Signed] = {}
118 self._trusted_delegations: set[tuple[str, str]] = set()
119 self.reference_time = datetime.datetime.now(datetime.timezone.utc)
121 if envelope_type is EnvelopeType.SIMPLE:
122 self._load_data = _load_from_simple_envelope
123 else:
124 self._load_data = _load_from_metadata
126 # Load and validate the local root metadata. Valid initial trusted root
127 # metadata is required
128 logger.debug("Updating initial trusted root")
129 self._load_trusted_root(root_data)
131 def contains(self, role: str, delegator: str) -> bool:
132 """Check if ``role`` is in ``TrustedMetadataSet`` and was verified
133 against ``delegator``.
134 """
135 return (delegator, role) in self._trusted_delegations
137 def _add(self, role: str, signed: Signed, delegator: str) -> None:
138 """Add role to trusted set, keep track of delegator(s)"""
139 self._trusted_set[role] = signed
140 self._trusted_delegations.add((delegator, role))
142 def __getitem__(self, role: str) -> Signed:
143 """Return current ``Signed`` for ``role``."""
144 return self._trusted_set[role]
146 def __len__(self) -> int:
147 """Return number of ``Signed`` objects in ``TrustedMetadataSet``."""
148 return len(self._trusted_set)
150 def __iter__(self) -> Iterator[Signed]:
151 """Return iterator over ``Signed`` objects in
152 ``TrustedMetadataSet``.
153 """
154 return iter(self._trusted_set.values())
156 # Helper properties for top level metadata
157 @property
158 def root(self) -> Root:
159 """Get current root."""
160 return cast("Root", self._trusted_set[Root.type])
162 @property
163 def timestamp(self) -> Timestamp:
164 """Get current timestamp."""
165 return cast("Timestamp", self._trusted_set[Timestamp.type])
167 @property
168 def snapshot(self) -> Snapshot:
169 """Get current snapshot."""
170 return cast("Snapshot", self._trusted_set[Snapshot.type])
172 @property
173 def targets(self) -> Targets:
174 """Get current top-level targets."""
175 return cast("Targets", self._trusted_set[Targets.type])
177 # Methods for updating metadata
178 def update_root(self, data: bytes) -> Root:
179 """Verify and load ``data`` as new root metadata.
181 Note that an expired intermediate root is considered valid: expiry is
182 only checked for the final root in ``update_timestamp()``.
184 Args:
185 data: Unverified new root metadata as bytes
187 Raises:
188 RuntimeError: This function is called after updating timestamp.
189 RepositoryError: Metadata failed to load or verify. The actual
190 error type and content will contain more details.
192 Returns:
193 Deserialized and verified ``Root`` object
194 """
195 if Timestamp.type in self._trusted_set:
196 raise RuntimeError("Cannot update root after timestamp")
197 logger.debug("Updating root")
199 new_root, new_root_bytes, new_root_signatures = self._load_data(
200 Root, data, self.root
201 )
202 if new_root.version != self.root.version + 1:
203 raise exceptions.BadVersionNumberError(
204 f"Expected root version {self.root.version + 1}"
205 f" instead got version {new_root.version}"
206 )
208 # Verify that new root is signed by itself
209 new_root.verify_delegate(Root.type, new_root_bytes, new_root_signatures)
211 self._add(Root.type, new_root, Root.type)
212 logger.debug("Updated root v%d", new_root.version)
214 return new_root
216 def update_timestamp(self, data: bytes) -> Timestamp:
217 """Verify and load ``data`` as new timestamp metadata.
219 Note that an intermediate timestamp is allowed to be expired:
220 ``TrustedMetadataSet`` will throw an ``ExpiredMetadataError`` in
221 this case but the intermediate timestamp will be loaded. This way
222 a newer timestamp can still be loaded (and the intermediate
223 timestamp will be used for rollback protection). Expired timestamp
224 will prevent loading snapshot metadata.
226 Args:
227 data: Unverified new timestamp metadata as bytes
229 Raises:
230 RuntimeError: This function is called after updating snapshot.
231 RepositoryError: Metadata failed to load or verify as final
232 timestamp. The actual error type and content will contain
233 more details.
235 Returns:
236 Deserialized and verified ``Timestamp`` object
237 """
238 if Snapshot.type in self._trusted_set:
239 raise RuntimeError("Cannot update timestamp after snapshot")
241 # client workflow 5.3.10: Make sure final root is not expired.
242 if self.root.is_expired(self.reference_time):
243 raise exceptions.ExpiredMetadataError("Final root.json is expired")
244 # No need to check for 5.3.11 (fast forward attack recovery):
245 # timestamp/snapshot can not yet be loaded at this point
247 new_timestamp, _, _ = self._load_data(Timestamp, data, self.root)
249 # If an existing trusted timestamp is updated,
250 # check for a rollback attack
251 if Timestamp.type in self._trusted_set:
252 # Prevent rolling back timestamp version
253 if new_timestamp.version < self.timestamp.version:
254 raise exceptions.BadVersionNumberError(
255 f"New timestamp version {new_timestamp.version} must"
256 f" be >= {self.timestamp.version}"
257 )
258 # Keep using old timestamp if versions are equal.
259 if new_timestamp.version == self.timestamp.version:
260 raise exceptions.EqualVersionNumberError
262 # Prevent rolling back snapshot version
263 snapshot_meta = self.timestamp.snapshot_meta
264 new_snapshot_meta = new_timestamp.snapshot_meta
265 if new_snapshot_meta.version < snapshot_meta.version:
266 raise exceptions.BadVersionNumberError(
267 f"New snapshot version must be >= {snapshot_meta.version}"
268 f", got version {new_snapshot_meta.version}"
269 )
271 # expiry not checked to allow old timestamp to be used for rollback
272 # protection of new timestamp: expiry is checked in update_snapshot()
274 self._add(Timestamp.type, new_timestamp, Root.type)
275 logger.debug("Updated timestamp v%d", new_timestamp.version)
277 # timestamp is loaded: raise if it is not valid _final_ timestamp
278 self._check_final_timestamp()
280 return new_timestamp
282 def _check_final_timestamp(self) -> None:
283 """Raise if timestamp is expired."""
285 if self.timestamp.is_expired(self.reference_time):
286 raise exceptions.ExpiredMetadataError("timestamp.json is expired")
288 def update_snapshot(
289 self, data: bytes, trusted: bool | None = False
290 ) -> Snapshot:
291 """Verify and load ``data`` as new snapshot metadata.
293 Note that an intermediate snapshot is allowed to be expired and version
294 is allowed to not match timestamp meta version: ``TrustedMetadataSet``
295 will throw an ``ExpiredMetadataError``/``BadVersionNumberError`` in
296 these cases but the intermediate snapshot will be loaded. This way a
297 newer snapshot can still be loaded (and the intermediate snapshot will
298 be used for rollback protection). Expired snapshot or snapshot that
299 does not match timestamp meta version will prevent loading targets.
301 Args:
302 data: Unverified new snapshot metadata as bytes
303 trusted: ``True`` if data has at some point been verified by
304 ``TrustedMetadataSet`` as a valid snapshot. Purpose of trusted
305 is to allow loading of locally stored snapshot as intermediate
306 snapshot even if hashes in current timestamp meta no longer
307 match data. Default is False.
309 Raises:
310 RuntimeError: This function is called before updating timestamp
311 or after updating targets.
312 RepositoryError: Data failed to load or verify as final snapshot.
313 The actual error type and content will contain more details.
315 Returns:
316 Deserialized and verified ``Snapshot`` object
317 """
319 if Timestamp.type not in self._trusted_set:
320 raise RuntimeError("Cannot update snapshot before timestamp")
321 if Targets.type in self._trusted_set:
322 raise RuntimeError("Cannot update snapshot after targets")
323 logger.debug("Updating snapshot")
325 # Snapshot cannot be loaded if final timestamp is expired
326 self._check_final_timestamp()
328 snapshot_meta = self.timestamp.snapshot_meta
330 # Verify non-trusted data against the hashes in timestamp, if any.
331 # Trusted snapshot data has already been verified once.
332 if not trusted:
333 snapshot_meta.verify_length_and_hashes(data)
335 new_snapshot, _, _ = self._load_data(Snapshot, data, self.root)
337 # version not checked against meta version to allow old snapshot to be
338 # used in rollback protection: it is checked when targets is updated
340 # If an existing trusted snapshot is updated, check for rollback attack
341 if Snapshot.type in self._trusted_set:
342 for filename, fileinfo in self.snapshot.meta.items():
343 new_fileinfo = new_snapshot.meta.get(filename)
345 # Prevent removal of any metadata in meta
346 if new_fileinfo is None:
347 raise exceptions.RepositoryError(
348 f"New snapshot is missing info for '{filename}'"
349 )
351 # Prevent rollback of any metadata versions
352 if new_fileinfo.version < fileinfo.version:
353 raise exceptions.BadVersionNumberError(
354 f"Expected {filename} version "
355 f"{new_fileinfo.version}, got {fileinfo.version}."
356 )
358 # expiry not checked to allow old snapshot to be used for rollback
359 # protection of new snapshot: it is checked when targets is updated
361 self._add(Snapshot.type, new_snapshot, Root.type)
362 logger.debug("Updated snapshot v%d", new_snapshot.version)
364 # snapshot is loaded, but we raise if it's not valid _final_ snapshot
365 self._check_final_snapshot()
367 return new_snapshot
369 def _check_final_snapshot(self) -> None:
370 """Raise if snapshot is expired or meta version does not match."""
372 if self.snapshot.is_expired(self.reference_time):
373 raise exceptions.ExpiredMetadataError("snapshot.json is expired")
374 snapshot_meta = self.timestamp.snapshot_meta
375 if self.snapshot.version != snapshot_meta.version:
376 raise exceptions.BadVersionNumberError(
377 f"Expected snapshot version {snapshot_meta.version}, "
378 f"got {self.snapshot.version}"
379 )
381 def update_targets(self, data: bytes) -> Targets:
382 """Verify and load ``data`` as new top-level targets metadata.
384 Args:
385 data: Unverified new targets metadata as bytes
387 Raises:
388 RepositoryError: Metadata failed to load or verify. The actual
389 error type and content will contain more details.
391 Returns:
392 Deserialized and verified `Targets`` object
393 """
394 return self.update_delegated_targets(data, Targets.type, Root.type)
396 def update_delegated_targets(
397 self, data: bytes, role_name: str, delegator_name: str
398 ) -> Targets:
399 """Verify and load ``data`` as new metadata for target ``role_name``.
401 Args:
402 data: Unverified new metadata as bytes
403 role_name: Role name of the new metadata
404 delegator_name: Name of the role delegating to the new metadata
406 Raises:
407 RuntimeError: This function is called before updating snapshot.
408 RepositoryError: Metadata failed to load or verify. The actual
409 error type and content will contain more details.
411 Returns:
412 Deserialized and verified ``Targets`` object
413 """
414 if Snapshot.type not in self._trusted_set:
415 raise RuntimeError("Cannot load targets before snapshot")
417 # Targets cannot be loaded if final snapshot is expired or its version
418 # does not match meta version in timestamp
419 self._check_final_snapshot()
421 delegator: Delegator | None = self.get(delegator_name)
422 if delegator is None:
423 raise RuntimeError("Cannot load targets before delegator")
425 logger.debug("Updating %s delegated by %s", role_name, delegator_name)
427 # Verify against the hashes in snapshot, if any
428 meta = self.snapshot.meta.get(f"{role_name}.json")
429 if meta is None:
430 raise exceptions.RepositoryError(
431 f"Snapshot does not contain information for '{role_name}'"
432 )
434 meta.verify_length_and_hashes(data)
436 new_delegate, _, _ = self._load_data(
437 Targets, data, delegator, role_name
438 )
440 version = new_delegate.version
441 if version != meta.version:
442 raise exceptions.BadVersionNumberError(
443 f"Expected {role_name} v{meta.version}, got v{version}."
444 )
446 if new_delegate.is_expired(self.reference_time):
447 raise exceptions.ExpiredMetadataError(f"New {role_name} is expired")
449 self._add(role_name, new_delegate, delegator_name)
450 logger.debug("Updated %s v%d", role_name, version)
452 return new_delegate
454 def _load_trusted_root(self, data: bytes) -> None:
455 """Verify and load ``data`` as trusted root metadata.
457 Note that an expired initial root is considered valid: expiry is
458 only checked for the final root in ``update_timestamp()``.
459 """
460 new_root, new_root_bytes, new_root_signatures = self._load_data(
461 Root, data
462 )
463 new_root.verify_delegate(Root.type, new_root_bytes, new_root_signatures)
465 self._add(Root.type, new_root, Root.type)
466 logger.debug("Loaded trusted root v%d", new_root.version)
469def _load_from_metadata(
470 role: type[T],
471 data: bytes,
472 delegator: Delegator | None = None,
473 role_name: str | None = None,
474) -> tuple[T, bytes, dict[str, Signature]]:
475 """Load traditional metadata bytes, and extract and verify payload.
477 If no delegator is passed, verification is skipped. Returns a tuple of
478 deserialized payload, signed payload bytes, and signatures.
479 """
480 md = Metadata[T].from_bytes(data)
482 if md.signed.type != role.type:
483 raise exceptions.RepositoryError(
484 f"Expected '{role.type}', got '{md.signed.type}'"
485 )
487 if delegator:
488 if role_name is None:
489 role_name = role.type
491 delegator.verify_delegate(role_name, md.signed_bytes, md.signatures)
493 return md.signed, md.signed_bytes, md.signatures
496def _load_from_simple_envelope(
497 role: type[T],
498 data: bytes,
499 delegator: Delegator | None = None,
500 role_name: str | None = None,
501) -> tuple[T, bytes, dict[str, Signature]]:
502 """Load simple envelope bytes, and extract and verify payload.
504 If no delegator is passed, verification is skipped. Returns a tuple of
505 deserialized payload, signed payload bytes, and signatures.
506 """
508 envelope = SimpleEnvelope[T].from_bytes(data)
510 if envelope.payload_type != SimpleEnvelope.DEFAULT_PAYLOAD_TYPE:
511 raise exceptions.RepositoryError(
512 f"Expected '{SimpleEnvelope.DEFAULT_PAYLOAD_TYPE}', "
513 f"got '{envelope.payload_type}'"
514 )
516 if delegator:
517 if role_name is None:
518 role_name = role.type
519 delegator.verify_delegate(
520 role_name, envelope.pae(), envelope.signatures
521 )
523 signed = envelope.get_signed()
524 if signed.type != role.type:
525 raise exceptions.RepositoryError(
526 f"Expected '{role.type}', got '{signed.type}'"
527 )
529 return signed, envelope.pae(), envelope.signatures