Make updater configuration immutable

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,14 +76,15 @@ 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,
             "https://example.com/metadata/",
             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:

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,14 +118,15 @@ 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,
             "https://example.com/metadata/",
             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()

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()
 

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.

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.