Update ngclient to return loaded metadata

This changes `TrustedMetadataSet` to return new trusted Metadata
on successful calls of the `update_<role>` functions and also
changes `Updater._load_targets` to return loaded metadata as well

Signed-off-by: Ivana Atanasova <[email protected]>

Author: Ivana Atanasova

tests/test_trusted_metadata_set.py

@@ -129,6 +129,21 @@ def test_update(self):
 
         self.assertTrue(count, 6)
 
+    def test_update_metadata_output(self):
+        timestamp = self.trusted_set.update_timestamp(self.metadata["timestamp"])
+        snapshot = self.trusted_set.update_snapshot(self.metadata["snapshot"])
+        self.trusted_set.update_targets(self.metadata["targets"])
+        delegeted_targets_1 = self.trusted_set.update_delegated_targets(
+            self.metadata["role1"], "role1", "targets"
+        )
+        delegeted_targets_2 = self.trusted_set.update_delegated_targets(
+            self.metadata["role2"], "role2", "role1"
+        )
+        self.assertIsInstance(timestamp, Metadata)
+        self.assertIsInstance(snapshot, Metadata)
+        self.assertIsInstance(delegeted_targets_1, Metadata)
+        self.assertIsInstance(delegeted_targets_2, Metadata)
+
     def test_out_of_order_ops(self):
         # Update snapshot before timestamp
         with self.assertRaises(RuntimeError):

tuf/ngclient/_internal/trusted_metadata_set.py

@@ -141,7 +141,7 @@ def targets(self) -> Optional[Metadata[Targets]]:
         return self._trusted_set.get("targets")
 
     # Methods for updating metadata
-    def update_root(self, data: bytes) -> None:
+    def update_root(self, data: bytes) -> Metadata:
         """Verifies and loads 'data' as new root metadata.
 
         Note that an expired intermediate root is considered valid: expiry is
@@ -182,7 +182,9 @@ def update_root(self, data: bytes) -> None:
         self._trusted_set["root"] = new_root
         logger.info("Updated root v%d", new_root.signed.version)
 
-    def update_timestamp(self, data: bytes) -> None:
+        return new_root
+
+    def update_timestamp(self, data: bytes) -> Metadata:
         """Verifies and loads 'data' as new timestamp metadata.
 
         Note that an intermediate timestamp is allowed to be expired:
@@ -251,6 +253,8 @@ def update_timestamp(self, data: bytes) -> None:
         # timestamp is loaded: raise if it is not valid _final_ timestamp
         self._check_final_timestamp()
 
+        return new_timestamp
+
     def _check_final_timestamp(self) -> None:
         """Raise if timestamp is expired"""
 
@@ -260,7 +264,7 @@ def _check_final_timestamp(self) -> None:
 
     def update_snapshot(
         self, data: bytes, trusted: Optional[bool] = False
-    ) -> None:
+    ) -> Metadata:
         """Verifies and loads 'data' as new snapshot metadata.
 
         Note that an intermediate snapshot is allowed to be expired and version
@@ -347,6 +351,8 @@ def update_snapshot(
         # snapshot is loaded, but we raise if it's not valid _final_ snapshot
         self._check_final_snapshot()
 
+        return new_snapshot
+
     def _check_final_snapshot(self) -> None:
         """Raise if snapshot is expired or meta version does not match"""
 
@@ -375,7 +381,7 @@ def update_targets(self, data: bytes) -> None:
 
     def update_delegated_targets(
         self, data: bytes, role_name: str, delegator_name: str
-    ) -> None:
+    ) -> Metadata:
         """Verifies and loads 'data' as new metadata for target 'role_name'.
 
         Args:
@@ -438,6 +444,8 @@ def update_delegated_targets(
         self._trusted_set[role_name] = new_delegate
         logger.info("Updated %s v%d", role_name, version)
 
+        return new_delegate
+
     def _load_trusted_root(self, data: bytes) -> None:
         """Verifies and loads 'data' as trusted root metadata.
 

tuf/ngclient/updater.py

@@ -72,7 +72,7 @@
 from securesystemslib import util as sslib_util
 
 from tuf import exceptions
-from tuf.api.metadata import TargetFile, Targets
+from tuf.api.metadata import Metadata, TargetFile
 from tuf.ngclient._internal import requests_fetcher, trusted_metadata_set
 from tuf.ngclient.config import UpdaterConfig
 from tuf.ngclient.fetcher import FetcherInterface
@@ -331,7 +331,7 @@ def _load_root(self) -> None:
                 # 404/403 means current root is newest available
                 break
 
-    def _load_timestamp(self) -> None:
+    def _load_timestamp(self) -> Metadata:
         """Load local and remote timestamp metadata"""
         try:
             data = self._load_local_metadata("timestamp")
@@ -344,15 +344,18 @@ def _load_timestamp(self) -> None:
         data = self._download_metadata(
             "timestamp", self.config.timestamp_max_length
         )
-        self._trusted_set.update_timestamp(data)
+        timestamp = self._trusted_set.update_timestamp(data)
         self._persist_metadata("timestamp", data)
 
-    def _load_snapshot(self) -> None:
+        return timestamp
+
+    def _load_snapshot(self) -> Metadata:
         """Load local (and if needed remote) snapshot metadata"""
         try:
             data = self._load_local_metadata("snapshot")
-            self._trusted_set.update_snapshot(data, trusted=True)
+            snapshot = self._trusted_set.update_snapshot(data, trusted=True)
             logger.debug("Local snapshot is valid: not downloading new one")
+            return snapshot
         except (OSError, exceptions.RepositoryError) as e:
             # Local snapshot does not exist or is invalid: update from remote
             logger.debug("Local snapshot not valid as final: %s", e)
@@ -365,15 +368,20 @@ def _load_snapshot(self) -> None:
                 version = snapshot_meta.version
 
             data = self._download_metadata("snapshot", length, version)
-            self._trusted_set.update_snapshot(data)
+            snapshot = self._trusted_set.update_snapshot(data)
             self._persist_metadata("snapshot", data)
 
-    def _load_targets(self, role: str, parent_role: str) -> None:
+            return snapshot
+
+    def _load_targets(self, role: str, parent_role: str) -> Metadata:
         """Load local (and if needed remote) metadata for 'role'."""
         try:
             data = self._load_local_metadata(role)
-            self._trusted_set.update_delegated_targets(data, role, parent_role)
+            delegated_targets = self._trusted_set.update_delegated_targets(
+                data, role, parent_role
+            )
             logger.debug("Local %s is valid: not downloading new one", role)
+            return delegated_targets
         except (OSError, exceptions.RepositoryError) as e:
             # Local 'role' does not exist or is invalid: update from remote
             logger.debug("Failed to load local %s: %s", role, e)
@@ -386,9 +394,13 @@ def _load_targets(self, role: str, parent_role: str) -> None:
                 version = metainfo.version
 
             data = self._download_metadata(role, length, version)
-            self._trusted_set.update_delegated_targets(data, role, parent_role)
+            delegated_targets = self._trusted_set.update_delegated_targets(
+                data, role, parent_role
+            )
             self._persist_metadata(role, data)
 
+            return delegated_targets
+
     def _preorder_depth_first_walk(
         self, target_filepath: str
     ) -> Optional[TargetFile]:
@@ -417,10 +429,9 @@ def _preorder_depth_first_walk(
 
             # The metadata for 'role_name' must be downloaded/updated before
             # its targets, delegations, and child roles can be inspected.
-            self._load_targets(role_name, parent_role)
+            role_metadata = self._load_targets(role_name, parent_role)
 
-            role_metadata: Targets = self._trusted_set[role_name].signed
-            target = role_metadata.targets.get(target_filepath)
+            target = role_metadata.signed.targets.get(target_filepath)
 
             if target is not None:
                 logger.debug("Found target in current role %s", role_name)
@@ -432,11 +443,13 @@ def _preorder_depth_first_walk(
             # And also decrement number of visited roles.
             number_of_delegations -= 1
 
-            if role_metadata.delegations is not None:
+            if role_metadata.signed.delegations is not None:
                 child_roles_to_visit = []
                 # NOTE: This may be a slow operation if there are many
                 # delegated roles.
-                for child_role in role_metadata.delegations.roles.values():
+                for (
+                    child_role
+                ) in role_metadata.signed.delegations.roles.values():
                     if child_role.is_delegated_path(target_filepath):
                         logger.debug("Adding child role %s", child_role.name)