ng client: support for prefix_targets_with_hash when downloading targets

tests/test_updater_ng.py

@@ -15,7 +15,10 @@
 import tuf.unittest_toolbox as unittest_toolbox
 
 from tests import utils
+from tuf.api.metadata import Metadata
 from tuf import ngclient
+from securesystemslib.signer import SSlibSigner
+from securesystemslib.interface import import_rsa_privatekey_from_file
 
 logger = logging.getLogger(__name__)
 
@@ -94,13 +97,13 @@ def setUp(self):
     url_prefix = 'http://' + utils.TEST_HOST_ADDRESS + ':' \
         + str(self.server_process_handler.port) + repository_basepath
 
-    metadata_url = f"{url_prefix}/metadata/"
-    targets_url = f"{url_prefix}/targets/"
+    self.metadata_url = f"{url_prefix}/metadata/"
+    self.targets_url = f"{url_prefix}/targets/"
     # Creating a repository instance.  The test cases will use this client
     # updater to refresh metadata, fetch target files, etc.
     self.repository_updater = ngclient.Updater(self.client_directory,
-                                              metadata_url,
-                                              targets_url)
+                                              self.metadata_url,
+                                              self.targets_url)
 
   def tearDown(self):
     # We are inheriting from custom class.
@@ -109,14 +112,91 @@ def tearDown(self):
     # Logs stdout and stderr from the sever subprocess.
     self.server_process_handler.flush_log()
 
+  def _create_consistent_target(self, targetname: str, target_hash:str) -> None:
+    """Create consistent targets copies of their non-consistent counterparts
+    inside the repository directory.
+
+    Args:
+      targetname: A string denoting the name of the target file.
+      target_hash: A string denoting the hash of the target.
+
+    """
+    consistent_target_name = f"{target_hash}.{targetname}"
+    source_path = os.path.join(self.repository_directory, "targets", targetname)
+    destination_path = os.path.join(
+      self.repository_directory, "targets", consistent_target_name
+    )
+    shutil.copy(source_path, destination_path)
+
+
+  def _make_root_file_with_consistent_snapshot_true(self) -> None:
+    """Swap the existing root file inside the client directory with a new root
+    file where the consistent_snapshot is set to true."""
+    root_path = os.path.join(self.client_directory, "root.json")
+    root = Metadata.from_file(root_path)
+    root.signed.consistent_snapshot = True
+    root_key_path = os.path.join(self.keystore_directory, "root_key")
+    root_key_dict = import_rsa_privatekey_from_file(
+            root_key_path, password="password"
+    )
+    root_signer = SSlibSigner(root_key_dict)
+    root.sign(root_signer)
+    # Remove the old root file and replace it with the newer root file.
+    os.remove(root_path)
+    root.to_file(root_path)
+
+
+  def test_refresh_on_consistent_targets(self):
+    # Generate a new root file where consistent_snapshot is set to true and
+    # replace the old root metadata file with it.
+    self._make_root_file_with_consistent_snapshot_true()
+    self.repository_updater = ngclient.Updater(self.client_directory,
+                                              self.metadata_url,
+                                              self.targets_url)
+    # All metadata is in local directory already
+    self.repository_updater.refresh()
+
+    # Get targetinfo for "file1.txt" listed in targets
+    targetinfo1 = self.repository_updater.get_one_valid_targetinfo("file1.txt")
+    # Get targetinfo for "file3.txt" listed in the delegated role1
+    targetinfo3 = self.repository_updater.get_one_valid_targetinfo("file3.txt")
+
+    # Create consistent targets with file path HASH.FILENAME.EXT
+    target1_hash = list(targetinfo1["fileinfo"].hashes.values())[0]
+    target3_hash = list(targetinfo3["fileinfo"].hashes.values())[0]
+    self._create_consistent_target("file1.txt", target1_hash)
+    self._create_consistent_target("file3.txt", target3_hash)
+
+    destination_directory = self.make_temp_directory()
+    updated_targets = self.repository_updater.updated_targets(
+      [targetinfo1, targetinfo3], destination_directory
+    )
+
+    self.assertListEqual(updated_targets, [targetinfo1, targetinfo3])
+    self.repository_updater.download_target(targetinfo1, destination_directory)
+    updated_targets = self.repository_updater.updated_targets(
+      updated_targets, destination_directory
+    )
+
+    self.assertListEqual(updated_targets, [targetinfo3])
+
+    self.repository_updater.download_target(targetinfo3, destination_directory)
+    updated_targets = self.repository_updater.updated_targets(
+      updated_targets, destination_directory
+    )
+
+    self.assertListEqual(updated_targets, [])
+
   def test_refresh(self):
+    # Test refresh without consistent targets - targets without hash prefixes.
+
     # All metadata is in local directory already
     self.repository_updater.refresh()
 
     # Get targetinfo for 'file1.txt' listed in targets
-    targetinfo1 = self.repository_updater.get_one_valid_targetinfo('file1.txt')
+    targetinfo1 = self.repository_updater.get_one_valid_targetinfo("file1.txt")
     # Get targetinfo for 'file3.txt' listed in the delegated role1
-    targetinfo3= self.repository_updater.get_one_valid_targetinfo('file3.txt')
+    targetinfo3 = self.repository_updater.get_one_valid_targetinfo("file3.txt")
 
     destination_directory = self.make_temp_directory()
     updated_targets = self.repository_updater.updated_targets([targetinfo1, targetinfo3],
@@ -146,7 +226,7 @@ def test_refresh_with_only_local_root(self):
     self.repository_updater.refresh()
 
     # Get targetinfo for 'file3.txt' listed in the delegated role1
-    targetinfo3= self.repository_updater.get_one_valid_targetinfo('file3.txt')
+    targetinfo3 = self.repository_updater.get_one_valid_targetinfo('file3.txt')
 
 if __name__ == '__main__':
   utils.configure_test_logging(sys.argv)

tuf/ngclient/config.py

@@ -9,9 +9,27 @@
 
 @dataclass
 class UpdaterConfig:
+    """Used to store Updater configuration.
+
+    Arguments:
+        max_root_rotations: The maximum number of root rotations.
+        max_delegations: The maximum number of delegations.
+        root_max_length: The maxmimum length of a root metadata file.
+        timestamp_max_length: The maximum length of a timestamp metadata file.
+        snapshot_max_length: The maximum length of a snapshot metadata file.
+        targets_max_length: The maximum length of a targets metadata file.
+        prefix_targets_with_hash: When consistent snapshots are used
+            (see https://theupdateframework.github.io/specification/latest/#consistent-snapshots), #pylint: disable=line-too-long
+            target download URLs are formed by prefixing the filename with a
+            hash digest of file content by default. This can be overridden by
+            setting prefix_targets_with_hash to False.
+
+    """
+
     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
+    prefix_targets_with_hash: bool = True

tuf/ngclient/updater.py

@@ -252,8 +252,12 @@ def download_target(
         else:
             target_base_url = _ensure_trailing_slash(target_base_url)
 
-        target_filepath = targetinfo["filepath"]
         target_fileinfo: "TargetFile" = targetinfo["fileinfo"]
+        target_filepath = targetinfo["filepath"]
+        consistent_snapshot = self._trusted_set.root.signed.consistent_snapshot
+        if consistent_snapshot and self.config.prefix_targets_with_hash:
+            hashes = list(target_fileinfo.hashes.values())
+            target_filepath = f"{hashes[0]}.{target_filepath}"
         full_url = parse.urljoin(target_base_url, target_filepath)
 
         with self._fetcher.download_file(
@@ -266,8 +270,11 @@ def download_target(
                     f"{target_filepath} length or hashes do not match"
                 ) from e
 
-            filepath = os.path.join(destination_directory, target_filepath)
-            sslib_util.persist_temp_file(target_file, filepath)
+            # Store the target file name without the HASH prefix.
+            local_filepath = os.path.join(
+                destination_directory, targetinfo["filepath"]
+            )
+            sslib_util.persist_temp_file(target_file, local_filepath)
 
     def _download_metadata(
         self, rolename: str, length: int, version: Optional[int] = None