ngclient constants and configuration

tuf/ngclient/__init__.py

@@ -4,5 +4,6 @@
 """TUF client public API
 """
 
+from tuf.ngclient.config import UpdaterConfig
 from tuf.ngclient.fetcher import FetcherInterface
 from tuf.ngclient.updater import Updater

tuf/ngclient/_internal/requests_fetcher.py

@@ -7,14 +7,15 @@
 
 import logging
 import time
+from typing import Optional
 from urllib import parse
 
 # Imports
 import requests
 import urllib3.exceptions
 
 import tuf
-from tuf import exceptions, settings
+from tuf import exceptions
 from tuf.ngclient.fetcher import FetcherInterface
 
 # Globals
@@ -47,6 +48,11 @@ def __init__(self):
         # Some cookies may not be HTTP-safe.
         self._sessions = {}
 
+        # Default settings
+        self.socket_timeout: int = 4  # seconds
+        self.chunk_size: int = 400000  # bytes
+        self.sleep_before_round: Optional[int] = None
+
     def fetch(self, url, required_length):
         """Fetches the contents of HTTP/HTTPS url from a remote server.
 
@@ -75,9 +81,7 @@ def fetch(self, url, required_length):
         # requests as:
         #  - connect timeout (max delay before first byte is received)
         #  - read (gap) timeout (max delay between bytes received)
-        response = session.get(
-            url, stream=True, timeout=settings.SOCKET_TIMEOUT
-        )
+        response = session.get(url, stream=True, timeout=self.socket_timeout)
         # Check response status.
         try:
             response.raise_for_status()
@@ -99,11 +103,11 @@ def chunks():
                     # large file in one shot. Before beginning the round, sleep
                     # (if set) for a short amount of time so that the CPU is not
                     # hogged in the while loop.
-                    if settings.SLEEP_BEFORE_ROUND:
-                        time.sleep(settings.SLEEP_BEFORE_ROUND)
+                    if self.sleep_before_round:
+                        time.sleep(self.sleep_before_round)
 
                     read_amount = min(
-                        settings.CHUNK_SIZE,
+                        self.chunk_size,
                         required_length - bytes_received,
                     )
 

tuf/ngclient/config.py

@@ -0,0 +1,17 @@
+# Copyright 2021, New York University and the TUF contributors
+# SPDX-License-Identifier: MIT OR Apache-2.0
+
+"""Configuration options for Updater class
+"""
+
+from dataclasses import dataclass
+
+
+@dataclass
+class UpdaterConfig:
+    max_root_rotations: int = 32
+    max_delegations: int = 32
+    root_max_length: int = 512000  # bytes
+    timestamp_max_length: int = 16384  # bytes
+    snapshot_max_length: int = 2000000  # bytes
+    targets_max_length: int = 5000000  # bytes

tuf/ngclient/updater.py

@@ -20,16 +20,9 @@
     requests_fetcher,
     trusted_metadata_set,
 )
+from tuf.ngclient.config import UpdaterConfig
 from tuf.ngclient.fetcher import FetcherInterface
 
-# Globals
-MAX_ROOT_ROTATIONS = 32
-MAX_DELEGATIONS = 32
-DEFAULT_ROOT_MAX_LENGTH = 512000  # bytes
-DEFAULT_TIMESTAMP_MAX_LENGTH = 16384  # bytes
-DEFAULT_SNAPSHOT_MAX_LENGTH = 2000000  # bytes
-DEFAULT_TARGETS_MAX_LENGTH = 5000000  # bytes
-
 logger = logging.getLogger(__name__)
 
 
@@ -45,6 +38,7 @@ def __init__(
         metadata_base_url: str,
         target_base_url: Optional[str] = None,
         fetcher: Optional[FetcherInterface] = None,
+        config: Optional[UpdaterConfig] = None,
     ):
         """
         Args:
@@ -76,6 +70,8 @@ def __init__(
         else:
             self._fetcher = fetcher
 
+        self.config = config or UpdaterConfig()
+
     def refresh(self) -> None:
         """
         This method downloads, verifies, and loads metadata for the top-level
@@ -251,12 +247,12 @@ def _load_root(self) -> None:
 
         # Update the root role
         lower_bound = self._trusted_set.root.signed.version + 1
-        upper_bound = lower_bound + MAX_ROOT_ROTATIONS
+        upper_bound = lower_bound + self.config.max_root_rotations
 
         for next_version in range(lower_bound, upper_bound):
             try:
                 data = self._download_metadata(
-                    "root", DEFAULT_ROOT_MAX_LENGTH, next_version
+                    "root", self.config.root_max_length, next_version
                 )
                 self._trusted_set.update_root(data)
                 self._persist_metadata("root", data)
@@ -281,7 +277,7 @@ def _load_timestamp(self) -> None:
 
         # Load from remote (whether local load succeeded or not)
         data = self._download_metadata(
-            "timestamp", DEFAULT_TIMESTAMP_MAX_LENGTH
+            "timestamp", self.config.timestamp_max_length
         )
         self._trusted_set.update_timestamp(data)
         self._persist_metadata("timestamp", data)
@@ -297,7 +293,7 @@ def _load_snapshot(self) -> None:
             logger.debug("Failed to load local snapshot %s", e)
 
             metainfo = self._trusted_set.timestamp.signed.meta["snapshot.json"]
-            length = metainfo.length or DEFAULT_SNAPSHOT_MAX_LENGTH
+            length = metainfo.length or self.config.snapshot_max_length
             version = None
             if self._trusted_set.root.signed.consistent_snapshot:
                 version = metainfo.version
@@ -317,7 +313,7 @@ def _load_targets(self, role: str, parent_role: str) -> None:
             logger.debug("Failed to load local %s: %s", role, e)
 
             metainfo = self._trusted_set.snapshot.signed.meta[f"{role}.json"]
-            length = metainfo.length or DEFAULT_TARGETS_MAX_LENGTH
+            length = metainfo.length or self.config.targets_max_length
             version = None
             if self._trusted_set.root.signed.consistent_snapshot:
                 version = metainfo.version
@@ -336,7 +332,7 @@ def _preorder_depth_first_walk(self, target_filepath) -> Dict:
         target = None
         role_names = [("targets", "root")]
         visited_role_names = set()
-        number_of_delegations = MAX_DELEGATIONS
+        number_of_delegations = self.config.max_delegations
 
         # Preorder depth-first traversal of the graph of target delegations.
         while (
@@ -417,7 +413,7 @@ def _preorder_depth_first_walk(self, target_filepath) -> Dict:
         ):
             msg = (
                 f"{len(role_names)} roles left to visit, but allowed to ",
-                f"visit at most {MAX_DELEGATIONS} delegations.",
+                f"visit at most {self.config.max_delegations} delegations.",
             )
             logger.debug(msg)