diff --git a/tests/test_updater_consistent_snapshot.py b/tests/test_updater_consistent_snapshot.py index cf272fcb42..df0d5ac84a 100644 --- a/tests/test_updater_consistent_snapshot.py +++ b/tests/test_updater_consistent_snapshot.py @@ -20,7 +20,7 @@ TargetFile, Targets, ) -from tuf.ngclient import Updater +from tuf.ngclient import Updater, UpdaterConfig if TYPE_CHECKING: from collections.abc import Iterable @@ -76,7 +76,7 @@ def _init_repo( return sim - def _init_updater(self) -> Updater: + def _init_updater(self, config: UpdaterConfig | None = None) -> Updater: """Create a new Updater instance""" return Updater( self.metadata_dir, @@ -84,6 +84,7 @@ def _init_updater(self) -> Updater: self.targets_dir, "https://example.com/targets/", self.sim, + config=config, bootstrap=self.sim.signed_roots[-1], ) @@ -221,8 +222,11 @@ def test_download_targets(self, test_case_data: dict[str, Any]) -> None: self.sim.add_target("targets", b"content", targetpath) self.sim.update_snapshot() - updater = self._init_updater() - updater.config.prefix_targets_with_hash = prefix_targets_with_hash + updater = self._init_updater( + config=UpdaterConfig( + prefix_targets_with_hash=prefix_targets_with_hash + ) + ) updater.refresh() for path in targetpaths: diff --git a/tests/test_updater_delegation_graphs.py b/tests/test_updater_delegation_graphs.py index 3c1bae5637..3adacb977a 100644 --- a/tests/test_updater_delegation_graphs.py +++ b/tests/test_updater_delegation_graphs.py @@ -22,7 +22,7 @@ DelegatedRole, Targets, ) -from tuf.ngclient import Updater +from tuf.ngclient import Updater, UpdaterConfig if TYPE_CHECKING: from collections.abc import Iterable @@ -118,7 +118,7 @@ def _init_repo(self, test_case: DelegationsTestCase) -> None: self.sim.targets.version += 1 self.sim.update_snapshot() - def _init_updater(self) -> Updater: + def _init_updater(self, config: UpdaterConfig | None = None) -> Updater: """Create a new Updater instance""" return Updater( self.metadata_dir, @@ -126,6 +126,7 @@ def _init_updater(self) -> Updater: self.targets_dir, "https://example.com/targets/", self.sim, + config=config, bootstrap=self.sim.signed_roots[0], ) @@ -270,9 +271,8 @@ def test_graph_traversal(self, test_data: DelegationsTestCase) -> None: self._init_repo(test_data) self.setup_subtest() - updater = self._init_updater() # restrict the max number of delegations to simplify the test - updater.config.max_delegations = 4 + updater = self._init_updater(config=UpdaterConfig(max_delegations=4)) # Call explicitly refresh to simplify the expected_calls list updater.refresh() self.sim.fetch_tracker.metadata.clear() diff --git a/tests/test_updater_top_level_update.py b/tests/test_updater_top_level_update.py index 76c74d4b57..3c287ff740 100644 --- a/tests/test_updater_top_level_update.py +++ b/tests/test_updater_top_level_update.py @@ -11,6 +11,7 @@ import sys import tempfile import unittest +from dataclasses import FrozenInstanceError, replace from datetime import timezone from pathlib import Path from typing import TYPE_CHECKING @@ -37,7 +38,7 @@ Targets, Timestamp, ) -from tuf.ngclient import Updater +from tuf.ngclient import Updater, UpdaterConfig if TYPE_CHECKING: from collections.abc import Iterable @@ -76,7 +77,11 @@ def _run_refresh(self, skip_bootstrap: bool = False) -> Updater: updater.refresh() return updater - def _init_updater(self, skip_bootstrap: bool = False) -> Updater: + def _init_updater( + self, + skip_bootstrap: bool = False, + config: UpdaterConfig | None = None, + ) -> Updater: """Create a new Updater instance""" if self.dump_dir is not None: self.sim.write() @@ -87,6 +92,7 @@ def _init_updater(self, skip_bootstrap: bool = False) -> Updater: self.targets_dir, "https://example.com/targets/", self.sim, + config=config, bootstrap=None if skip_bootstrap else self.sim.signed_roots[0], ) @@ -183,8 +189,7 @@ def test_trusted_root_unsigned_without_bootstrap(self) -> None: def test_max_root_rotations(self) -> None: # Root must stop looking for new versions after Y number of # intermediate files were downloaded. - updater = self._init_updater() - updater.config.max_root_rotations = 3 + updater = self._init_updater(config=UpdaterConfig(max_root_rotations=3)) # Create some number of roots greater than 'max_root_rotations' while self.sim.root.version < updater.config.max_root_rotations + 3: @@ -201,6 +206,30 @@ def test_max_root_rotations(self) -> None: Root.type, initial_root_version + updater.config.max_root_rotations ) + def test_configuration_is_immutable(self) -> None: + config = UpdaterConfig(max_root_rotations=3) + updater = self._init_updater(config=config) + self.assertIs(updater.config, config) + field_name = "max_root_rotations" + config_attribute = "config" + + for refresh in (False, True): + with self.subTest(refresh=refresh): + if refresh: + updater.refresh() + with self.assertRaises(FrozenInstanceError): + setattr(config, field_name, 10) + with self.assertRaises(FrozenInstanceError): + setattr(updater.config, field_name, 10) + with self.assertRaises(AttributeError): + setattr(updater, config_attribute, UpdaterConfig()) + self.assertEqual(updater.config.max_root_rotations, 3) + + # A new configuration can be derived without changing the old one. + new_config = replace(config, max_root_rotations=10) + self.assertEqual(new_config.max_root_rotations, 10) + self.assertEqual(config.max_root_rotations, 3) + def test_intermediate_root_incorrectly_signed(self) -> None: # Check for an arbitrary software attack @@ -855,16 +884,15 @@ def test_max_metadata_lengths(self) -> None: self.sim.root.version += 1 self.sim.publish_root() - config_vars = [ - "root_max_length", - "timestamp_max_length", - "snapshot_max_length", - "targets_max_length", + configs = [ + UpdaterConfig(root_max_length=100), + UpdaterConfig(timestamp_max_length=100), + UpdaterConfig(snapshot_max_length=100), + UpdaterConfig(targets_max_length=100), ] # make sure going over any length limit raises DownloadLengthMismatchError - for var_name in config_vars: - updater = self._init_updater() - setattr(updater.config, var_name, 100) + for config in configs: + updater = self._init_updater(config=config) with self.assertRaises(DownloadLengthMismatchError): updater.refresh() diff --git a/tuf/ngclient/config.py b/tuf/ngclient/config.py index 3a41fad451..61e60afdb9 100644 --- a/tuf/ngclient/config.py +++ b/tuf/ngclient/config.py @@ -22,10 +22,14 @@ class EnvelopeType(Flag): SIMPLE = 2 -@dataclass +@dataclass(frozen=True) class UpdaterConfig: """Used to store ``Updater`` configuration. + Configuration is immutable. Set options in the constructor or use + :func:`dataclasses.replace` to create a modified configuration for a new + ``Updater`` instance. + Args: max_root_rotations: Maximum number of root rotations. max_delegations: Maximum number of delegations. diff --git a/tuf/ngclient/updater.py b/tuf/ngclient/updater.py index 0ae0fabe51..f70c10b3c0 100644 --- a/tuf/ngclient/updater.py +++ b/tuf/ngclient/updater.py @@ -91,7 +91,8 @@ class Updater: fetcher: ``Optional``; ``FetcherInterface`` implementation used to download both metadata and targets. Default is ``Urllib3Fetcher`` config: ``Optional``; ``UpdaterConfig`` could be used to setup common - configuration options. + configuration options. Configuration cannot be changed after + initialization: create a new ``Updater`` to use different options. bootstrap: Initial root metadata bytes. This argument is required. Pass the embedded root metadata bytes for secure initialization. Pass ``None`` only if you explicitly want to use the cached @@ -122,7 +123,7 @@ def __init__( # noqa: PLR0917 else: self._target_base_url = _ensure_trailing_slash(target_base_url) - self.config = config or UpdaterConfig() + self._config = config or UpdaterConfig() if fetcher is not None: self._fetcher = fetcher else: @@ -147,6 +148,11 @@ def __init__( # noqa: PLR0917 self._persist_root(self._trusted_set.root.version, bootstrap) self._update_root_symlink() + @property + def config(self) -> UpdaterConfig: + """Read-only configuration used for this Updater's lifetime.""" + return self._config + def refresh(self) -> None: """Refresh top-level metadata.