Metadata API: make new verify_delegate unaware of Metadata

Change new _Delegator.verify_delegate to take payload bytes and
signatures instead of a Metadata object and a payload serializer.

This allows using verify_delegate for payloads that do not come in
a Metadata container, but e.g. in a DSSE envelope (see #2385).

Usage becomes a bit more cumbersome, but still feels reasonable with the
recently added shortcut for default canonical bytes representation of
Metadata.signed.

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

Author: Lukas Puehringer

examples/repository/_simplerepo.py

@@ -177,7 +177,7 @@ def submit_role(self, role: str, data: bytes) -> bool:
                 if not targetpath.startswith(f"{role}/"):
                     raise ValueError(f"targets allowed under {role}/ only")
 
-            self.targets().verify_delegate(role, md)
+            self.targets().verify_delegate(role, md.signed_bytes, md.signatures)
 
             if md.signed.version != self.targets(role).version + 1:
                 raise ValueError("Invalid version {md.signed.version}")

tests/test_api.py

@@ -364,29 +364,49 @@ def test_signed_verify_delegate(self) -> None:
         role2 = role2_md.signed
 
         # test the expected delegation tree
-        root.verify_delegate(Root.type, root_md)
-        root.verify_delegate(Snapshot.type, snapshot_md)
-        root.verify_delegate(Targets.type, targets_md)
-        targets.verify_delegate("role1", role1_md)
-        role1.verify_delegate("role2", role2_md)
+        root.verify_delegate(
+            Root.type, root_md.signed_bytes, root_md.signatures
+        )
+        root.verify_delegate(
+            Snapshot.type, snapshot_md.signed_bytes, snapshot_md.signatures
+        )
+        root.verify_delegate(
+            Targets.type, targets_md.signed_bytes, targets_md.signatures
+        )
+        targets.verify_delegate(
+            "role1", role1_md.signed_bytes, role1_md.signatures
+        )
+        role1.verify_delegate(
+            "role2", role2_md.signed_bytes, role2_md.signatures
+        )
 
         # only root and targets can verify delegates
         with self.assertRaises(AttributeError):
-            snapshot.verify_delegate(Snapshot.type, snapshot_md)
+            snapshot.verify_delegate(
+                Snapshot.type, snapshot_md.signed_bytes, snapshot_md.signatures
+            )
         # verify fails for roles that are not delegated by delegator
         with self.assertRaises(ValueError):
-            root.verify_delegate("role1", role1_md)
+            root.verify_delegate(
+                "role1", role1_md.signed_bytes, role1_md.signatures
+            )
         with self.assertRaises(ValueError):
-            targets.verify_delegate(Targets.type, targets_md)
+            targets.verify_delegate(
+                Targets.type, targets_md.signed_bytes, targets_md.signatures
+            )
         # verify fails when delegator has no delegations
         with self.assertRaises(ValueError):
-            role2.verify_delegate("role1", role1_md)
+            role2.verify_delegate(
+                "role1", role1_md.signed_bytes, role1_md.signatures
+            )
 
         # verify fails when delegate content is modified
         expires = snapshot.expires
         snapshot.expires = expires + timedelta(days=1)
         with self.assertRaises(exceptions.UnsignedMetadataError):
-            root.verify_delegate(Snapshot.type, snapshot_md)
+            root.verify_delegate(
+                Snapshot.type, snapshot_md.signed_bytes, snapshot_md.signatures
+            )
         snapshot.expires = expires
 
         # verify fails if sslib verify fails with VerificationError
@@ -395,12 +415,16 @@ def test_signed_verify_delegate(self) -> None:
         good_sig = snapshot_md.signatures[keyid].signature
         snapshot_md.signatures[keyid].signature = "foo"
         with self.assertRaises(exceptions.UnsignedMetadataError):
-            root.verify_delegate(Snapshot.type, snapshot_md)
+            root.verify_delegate(
+                Snapshot.type, snapshot_md.signed_bytes, snapshot_md.signatures
+            )
         snapshot_md.signatures[keyid].signature = good_sig
 
         # verify fails if roles keys do not sign the metadata
         with self.assertRaises(exceptions.UnsignedMetadataError):
-            root.verify_delegate(Timestamp.type, snapshot_md)
+            root.verify_delegate(
+                Timestamp.type, snapshot_md.signed_bytes, snapshot_md.signatures
+            )
 
         # Add a key to snapshot role, make sure the new sig fails to verify
         ts_keyid = next(iter(root.roles[Timestamp.type].keyids))
@@ -409,19 +433,25 @@ def test_signed_verify_delegate(self) -> None:
 
         # verify succeeds if threshold is reached even if some signatures
         # fail to verify
-        root.verify_delegate(Snapshot.type, snapshot_md)
+        root.verify_delegate(
+            Snapshot.type, snapshot_md.signed_bytes, snapshot_md.signatures
+        )
 
         # verify fails if threshold of signatures is not reached
         root.roles[Snapshot.type].threshold = 2
         with self.assertRaises(exceptions.UnsignedMetadataError):
-            root.verify_delegate(Snapshot.type, snapshot_md)
+            root.verify_delegate(
+                Snapshot.type, snapshot_md.signed_bytes, snapshot_md.signatures
+            )
 
         # verify succeeds when we correct the new signature and reach the
         # threshold of 2 keys
         snapshot_md.sign(
             SSlibSigner(self.keystore[Timestamp.type]), append=True
         )
-        root.verify_delegate(Snapshot.type, snapshot_md)
+        root.verify_delegate(
+            Snapshot.type, snapshot_md.signed_bytes, snapshot_md.signatures
+        )
 
     def test_key_class(self) -> None:
         # Test if from_securesystemslib_key removes the private key from keyval

tuf/api/metadata.py

@@ -408,8 +408,14 @@ def verify_delegate(
         if self.signed.type not in ["root", "targets"]:
             raise TypeError("Call is valid only on delegator metadata")
 
+        if signed_serializer is None:
+            payload = delegated_metadata.signed_bytes
+
+        else:
+            payload = signed_serializer.serialize(delegated_metadata.signed)
+
         self.signed.verify_delegate(
-            delegated_role, delegated_metadata, signed_serializer
+            delegated_role, payload, delegated_metadata.signatures
         )
 
 
@@ -662,31 +668,25 @@ def get_key(self, keyid: str) -> Key:
     def verify_delegate(
         self,
         delegated_role: str,
-        delegated_metadata: Metadata,
-        signed_serializer: Optional[SignedSerializer] = None,
+        payload: bytes,
+        signatures: Dict[str, Signature],
     ) -> None:
-        """Verify that ``delegated_metadata`` is signed with the required
-        threshold of keys for ``delegated_role``.
+        """Verify signature threshold for delegated role.
+
+        Verify that there are enough valid ``signatures`` over ``payload``, to
+        meet the threshold of keys for ``delegated_role``, as defined by the
+        delegator (``self``).
 
         Args:
             delegated_role: Name of the delegated role to verify
-            delegated_metadata: ``Metadata`` object for the delegated role
-            signed_serializer: Serializer used for delegate
-                serialization. Default is ``CanonicalJSONSerializer``.
+            payload: Signed payload bytes for the delegated role
+            signatures: Signatures over payload bytes
 
         Raises:
             UnsignedMetadataError: ``delegated_role`` was not signed with
                 required threshold of keys for ``role_name``.
             ValueError: no delegation was found for ``delegated_role``.
         """
-
-        if signed_serializer is None:
-            # pylint: disable=import-outside-toplevel
-            from tuf.api.serialization.json import CanonicalJSONSerializer
-
-            signed_serializer = CanonicalJSONSerializer()
-
-        data = signed_serializer.serialize(delegated_metadata.signed)
         role = self.get_delegated_role(delegated_role)
 
         # verify that delegated_metadata is signed by threshold of unique keys
@@ -698,13 +698,13 @@ def verify_delegate(
                 logger.info("No key for keyid %s", keyid)
                 continue
 
-            if keyid not in delegated_metadata.signatures:
+            if keyid not in signatures:
                 logger.info("No signature for keyid %s", keyid)
                 continue
 
-            sig = delegated_metadata.signatures[keyid]
+            sig = signatures[keyid]
             try:
-                key.verify_signature(sig, data)
+                key.verify_signature(sig, payload)
                 signing_keys.add(keyid)
             except sslib_exceptions.UnverifiedSignatureError:
                 logger.info("Key %s failed to verify %s", keyid, delegated_role)

tuf/ngclient/_internal/trusted_metadata_set.py

@@ -161,7 +161,9 @@ def update_root(self, data: bytes) -> Metadata[Root]:
             )
 
         # Verify that new root is signed by trusted root
-        self.root.signed.verify_delegate(Root.type, new_root)
+        self.root.signed.verify_delegate(
+            Root.type, new_root.signed_bytes, new_root.signatures
+        )
 
         if new_root.signed.version != self.root.signed.version + 1:
             raise exceptions.BadVersionNumberError(
@@ -170,7 +172,9 @@ def update_root(self, data: bytes) -> Metadata[Root]:
             )
 
         # Verify that new root is signed by itself
-        new_root.signed.verify_delegate(Root.type, new_root)
+        new_root.signed.verify_delegate(
+            Root.type, new_root.signed_bytes, new_root.signatures
+        )
 
         self._trusted_set[Root.type] = new_root
         logger.debug("Updated root v%d", new_root.signed.version)
@@ -215,7 +219,9 @@ def update_timestamp(self, data: bytes) -> Metadata[Timestamp]:
                 f"Expected 'timestamp', got '{new_timestamp.signed.type}'"
             )
 
-        self.root.signed.verify_delegate(Timestamp.type, new_timestamp)
+        self.root.signed.verify_delegate(
+            Timestamp.type, new_timestamp.signed_bytes, new_timestamp.signatures
+        )
 
         # If an existing trusted timestamp is updated,
         # check for a rollback attack
@@ -310,7 +316,9 @@ def update_snapshot(
                 f"Expected 'snapshot', got '{new_snapshot.signed.type}'"
             )
 
-        self.root.signed.verify_delegate(Snapshot.type, new_snapshot)
+        self.root.signed.verify_delegate(
+            Snapshot.type, new_snapshot.signed_bytes, new_snapshot.signatures
+        )
 
         # version not checked against meta version to allow old snapshot to be
         # used in rollback protection: it is checked when targets is updated
@@ -418,7 +426,9 @@ def update_delegated_targets(
                 f"Expected 'targets', got '{new_delegate.signed.type}'"
             )
 
-        delegator.signed.verify_delegate(role_name, new_delegate)
+        delegator.signed.verify_delegate(
+            role_name, new_delegate.signed_bytes, new_delegate.signatures
+        )
 
         version = new_delegate.signed.version
         if version != meta.version:
@@ -447,7 +457,9 @@ def _load_trusted_root(self, data: bytes) -> None:
                 f"Expected 'root', got '{new_root.signed.type}'"
             )
 
-        new_root.signed.verify_delegate(Root.type, new_root)
+        new_root.signed.verify_delegate(
+            Root.type, new_root.signed_bytes, new_root.signatures
+        )
 
         self._trusted_set[Root.type] = new_root
         logger.debug("Loaded trusted root v%d", new_root.signed.version)