ngclient: support dsse in Updater

* Add `use_dsse` updater config parameter which indicates, if an updater
  instance expects metadata to come in a DSSE envelope.

* Update TrustedMetadataSet to take an Unwrapper instance.

* Update Updater, to pass an EnvelopeUnwrapper to TrustedMetadataSet, if
  configured with `use_dsse`.

Signed-off-by: Lukas Puehringer <lukas.puehringer@nyu.edu>

tests/test_trusted_metadata_set.py

@@ -198,17 +198,17 @@ def test_root_with_invalid_json(self) -> None:
         for test_func in [TrustedMetadataSet, self.trusted_set.update_root]:
             # root is not json
             with self.assertRaises(exceptions.RepositoryError):
-                test_func(b"")
+                test_func(b"")  # type: ignore[operator]
 
             # root is invalid
             root = Metadata.from_bytes(self.metadata[Root.type])
             root.signed.version += 1
             with self.assertRaises(exceptions.UnsignedMetadataError):
-                test_func(root.to_bytes())
+                test_func(root.to_bytes())  # type: ignore[operator]
 
             # metadata is of wrong type
             with self.assertRaises(exceptions.RepositoryError):
-                test_func(self.metadata[Snapshot.type])
+                test_func(self.metadata[Snapshot.type])  # type: ignore[operator]
 
     def test_top_level_md_with_invalid_json(self) -> None:
         top_level_md: List[Tuple[bytes, Callable[[bytes], Signed]]] = [

tuf/ngclient/_internal/trusted_metadata_set.py

@@ -68,7 +68,7 @@
 
 from tuf.api import exceptions
 from tuf.api.metadata import Root, Signed, Snapshot, Targets, Timestamp
-from tuf.ngclient._internal.wrapping import MetadataUnwrapper
+from tuf.ngclient._internal.wrapping import MetadataUnwrapper, Unwrapper
 
 logger = logging.getLogger(__name__)
 
@@ -82,19 +82,24 @@ class TrustedMetadataSet(abc.Mapping):
     what is updated.
     """
 
-    def __init__(self, root_data: bytes):
+    def __init__(self, root_data: bytes, unwrapper: Optional[Unwrapper] = None):
         """Initialize ``TrustedMetadataSet`` by loading trusted root metadata.
 
         Args:
             root_data: Trusted root metadata as bytes. Note that this metadata
                 will only be verified by itself: it is the source of trust for
                 all metadata in the ``TrustedMetadataSet``
+            unwrapper: Used to unwrap and verify metadata. Default is
+                MetadataUnwrapper.
 
         Raises:
             RepositoryError: Metadata failed to load or verify. The actual
                 error type and content will contain more details.
         """
-        self._unwrapper = MetadataUnwrapper()
+        if unwrapper is None:
+            unwrapper = MetadataUnwrapper()
+        self._unwrapper = unwrapper
+
         self._trusted_set: Dict[str, Signed] = {}
         self.reference_time = datetime.datetime.utcnow()
 

tuf/ngclient/config.py

@@ -23,7 +23,8 @@ class UpdaterConfig:
             are used, 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``.
-
+        use_dsse: If true, expect metadata in a DSSE Envelope. Use
+            traditional Metadata (canonical json) otherwise.
     """
 
     max_root_rotations: int = 32
@@ -33,3 +34,4 @@ class UpdaterConfig:
     snapshot_max_length: int = 2000000  # bytes
     targets_max_length: int = 5000000  # bytes
     prefix_targets_with_hash: bool = True
+    use_dsse: bool = False

tuf/ngclient/updater.py

@@ -46,7 +46,11 @@
 
 from tuf.api import exceptions
 from tuf.api.metadata import Root, Snapshot, TargetFile, Targets, Timestamp
-from tuf.ngclient._internal import requests_fetcher, trusted_metadata_set
+from tuf.ngclient._internal import (
+    requests_fetcher,
+    trusted_metadata_set,
+    wrapping,
+)
 from tuf.ngclient.config import UpdaterConfig
 from tuf.ngclient.fetcher import FetcherInterface
 
@@ -94,10 +98,17 @@ def __init__(
 
         # Read trusted local root metadata
         data = self._load_local_metadata(Root.type)
-        self._trusted_set = trusted_metadata_set.TrustedMetadataSet(data)
         self._fetcher = fetcher or requests_fetcher.RequestsFetcher()
         self.config = config or UpdaterConfig()
 
+        unwrapper: Optional[wrapping.Unwrapper] = None
+        if self.config.use_dsse:
+            unwrapper = wrapping.EnvelopeUnwrapper()
+
+        self._trusted_set = trusted_metadata_set.TrustedMetadataSet(
+            data, unwrapper
+        )
+
     def refresh(self) -> None:
         """Refresh top-level metadata.