Merge pull request #2378 from jku/move-verify-delegate-v2

Lukas Pühringer · web-flow · commit a871f648e571 · 2023-08-21T13:27:01.000+02:00
Move verify_delegate() to Root/Targets
diff --git a/examples/repository/_simplerepo.py b/examples/repository/_simplerepo.py
@@ -177,8 +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")
 
-            targets_md = self.open("targets")
-            targets_md.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}")
diff --git a/tests/test_api.py b/tests/test_api.py
@@ -49,7 +49,7 @@
     Timestamp,
 )
 from tuf.api.serialization import DeserializationError, SerializationError
-from tuf.api.serialization.json import CanonicalJSONSerializer, JSONSerializer
+from tuf.api.serialization.json import JSONSerializer
 
 logger = logging.getLogger(__name__)
 
@@ -208,7 +208,7 @@ def test_sign_verify(self) -> None:
         # Load sample metadata (targets) and assert ...
         md_obj = Metadata.from_file(os.path.join(path, "targets.json"))
         sig = md_obj.signatures[targets_keyid]
-        data = CanonicalJSONSerializer().serialize(md_obj.signed)
+        data = md_obj.signed_bytes
 
         # ... it has a single existing signature,
         self.assertEqual(len(md_obj.signatures), 1)
@@ -274,7 +274,7 @@ def test_key_verify_failures(self) -> None:
         path = os.path.join(self.repo_dir, "metadata", "timestamp.json")
         md_obj = Metadata.from_file(path)
         sig = md_obj.signatures[timestamp_keyid]
-        data = CanonicalJSONSerializer().serialize(md_obj.signed)
+        data = md_obj.signed_bytes
 
         # Test failure on unknown scheme (securesystemslib
         # UnsupportedAlgorithmError)
@@ -362,44 +362,113 @@ def test_metadata_verify_delegate(self) -> None:
         with self.assertRaises(ValueError):
             role2.verify_delegate("role1", role1)
 
+    def test_signed_verify_delegate(self) -> None:
+        # pylint: disable=too-many-locals,too-many-statements
+        root_path = os.path.join(self.repo_dir, "metadata", "root.json")
+        root_md = Metadata[Root].from_file(root_path)
+        root = root_md.signed
+        snapshot_path = os.path.join(self.repo_dir, "metadata", "snapshot.json")
+        snapshot_md = Metadata[Snapshot].from_file(snapshot_path)
+        snapshot = snapshot_md.signed
+        targets_path = os.path.join(self.repo_dir, "metadata", "targets.json")
+        targets_md = Metadata[Targets].from_file(targets_path)
+        targets = targets_md.signed
+        role1_path = os.path.join(self.repo_dir, "metadata", "role1.json")
+        role1_md = Metadata[Targets].from_file(role1_path)
+        role1 = role1_md.signed
+        role2_path = os.path.join(self.repo_dir, "metadata", "role2.json")
+        role2_md = Metadata[Targets].from_file(role2_path)
+        role2 = role2_md.signed
+
+        # test the expected delegation tree
+        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.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.signed_bytes, role1_md.signatures
+            )
+        with self.assertRaises(ValueError):
+            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.signed_bytes, role1_md.signatures
+            )
+
         # verify fails when delegate content is modified
-        expires = snapshot.signed.expires
-        snapshot.signed.expires = expires + timedelta(days=1)
+        expires = snapshot.expires
+        snapshot.expires = expires + timedelta(days=1)
         with self.assertRaises(exceptions.UnsignedMetadataError):
-            root.verify_delegate(Snapshot.type, snapshot)
-        snapshot.signed.expires = expires
+            root.verify_delegate(
+                Snapshot.type, snapshot_md.signed_bytes, snapshot_md.signatures
+            )
+        snapshot.expires = expires
 
         # verify fails if sslib verify fails with VerificationError
         # (in this case signature is malformed)
-        keyid = next(iter(root.signed.roles[Snapshot.type].keyids))
-        good_sig = snapshot.signatures[keyid].signature
-        snapshot.signatures[keyid].signature = "foo"
+        keyid = next(iter(root.roles[Snapshot.type].keyids))
+        good_sig = snapshot_md.signatures[keyid].signature
+        snapshot_md.signatures[keyid].signature = "foo"
         with self.assertRaises(exceptions.UnsignedMetadataError):
-            root.verify_delegate(Snapshot.type, snapshot)
-        snapshot.signatures[keyid].signature = good_sig
+            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)
+            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.signed.roles[Timestamp.type].keyids))
-        root.signed.add_key(root.signed.keys[ts_keyid], Snapshot.type)
-        snapshot.signatures[ts_keyid] = Signature(ts_keyid, "ff" * 64)
+        ts_keyid = next(iter(root.roles[Timestamp.type].keyids))
+        root.add_key(root.keys[ts_keyid], Snapshot.type)
+        snapshot_md.signatures[ts_keyid] = Signature(ts_keyid, "ff" * 64)
 
         # verify succeeds if threshold is reached even if some signatures
         # fail to verify
-        root.verify_delegate(Snapshot.type, snapshot)
+        root.verify_delegate(
+            Snapshot.type, snapshot_md.signed_bytes, snapshot_md.signatures
+        )
 
         # verify fails if threshold of signatures is not reached
-        root.signed.roles[Snapshot.type].threshold = 2
+        root.roles[Snapshot.type].threshold = 2
         with self.assertRaises(exceptions.UnsignedMetadataError):
-            root.verify_delegate(Snapshot.type, snapshot)
+            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.sign(SSlibSigner(self.keystore[Timestamp.type]), append=True)
-        root.verify_delegate(Snapshot.type, snapshot)
+        snapshot_md.sign(
+            SSlibSigner(self.keystore[Timestamp.type]), append=True
+        )
+        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
diff --git a/tuf/api/metadata.py b/tuf/api/metadata.py
@@ -150,6 +150,16 @@ def __eq__(self, other: Any) -> bool:
             and self.unrecognized_fields == other.unrecognized_fields
         )
 
+    @property
+    def signed_bytes(self) -> bytes:
+        """Default canonical json byte representation of ``self.signed``."""
+
+        # Use local scope import to avoid circular import errors
+        # pylint: disable=import-outside-toplevel
+        from tuf.api.serialization.json import CanonicalJSONSerializer
+
+        return CanonicalJSONSerializer().serialize(self.signed)
+
     @classmethod
     def from_dict(cls, metadata: Dict[str, Any]) -> "Metadata[T]":
         """Create ``Metadata`` object from its json/dict representation.
@@ -366,13 +376,9 @@ def sign(
         """
 
         if signed_serializer is None:
-            # Use local scope import to avoid circular import errors
-            # pylint: disable=import-outside-toplevel
-            from tuf.api.serialization.json import CanonicalJSONSerializer
-
-            signed_serializer = CanonicalJSONSerializer()
-
-        bytes_data = signed_serializer.serialize(self.signed)
+            bytes_data = self.signed_bytes
+        else:
+            bytes_data = signed_serializer.serialize(self.signed)
 
         try:
             signature = signer.sign(bytes_data)
@@ -393,58 +399,24 @@ def verify_delegate(
         signed_serializer: Optional[SignedSerializer] = None,
     ) -> None:
         """Verify that ``delegated_metadata`` is signed with the required
-        threshold of keys for the delegated role ``delegated_role``.
+        threshold of keys for ``delegated_role``.
 
-        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``.
-
-        Raises:
-            UnsignedMetadataError: ``delegated_role`` was not signed with
-                required threshold of keys for ``role_name``.
-            ValueError: no delegation was found for ``delegated_role``.
-            TypeError: called this function on non-delegating metadata class.
+        .. deprecated:: 3.1.0
+           Please use ``Root.verify_delegate()`` or ``Targets.verify_delegate()``.
         """
 
         if self.signed.type not in ["root", "targets"]:
             raise TypeError("Call is valid only on delegator metadata")
 
         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.signed.get_delegated_role(delegated_role)
-
-        # verify that delegated_metadata is signed by threshold of unique keys
-        signing_keys = set()
-        for keyid in role.keyids:
-            try:
-                key = self.signed.get_key(keyid)
-            except ValueError:
-                logger.info("No key for keyid %s", keyid)
-                continue
+            payload = delegated_metadata.signed_bytes
 
-            if keyid not in delegated_metadata.signatures:
-                logger.info("No signature for keyid %s", keyid)
-                continue
-
-            sig = delegated_metadata.signatures[keyid]
-            try:
-                key.verify_signature(sig, data)
-                signing_keys.add(keyid)
-            except sslib_exceptions.UnverifiedSignatureError:
-                logger.info("Key %s failed to verify %s", keyid, delegated_role)
+        else:
+            payload = signed_serializer.serialize(delegated_metadata.signed)
 
-        if len(signing_keys) < role.threshold:
-            raise UnsignedMetadataError(
-                f"{delegated_role} was signed by {len(signing_keys)}/"
-                f"{role.threshold} keys",
-            )
+        self.signed.verify_delegate(
+            delegated_role, payload, delegated_metadata.signatures
+        )
 
 
 class Signed(metaclass=abc.ABCMeta):
@@ -674,7 +646,77 @@ def to_dict(self) -> Dict[str, Any]:
         }
 
 
-class Root(Signed):
+class _DelegatorMixin(metaclass=abc.ABCMeta):
+    """Class that implements verify_delegate() for Root and Targets"""
+
+    @abc.abstractmethod
+    def get_delegated_role(self, delegated_role: str) -> Role:
+        """Return the role object for the given delegated role.
+
+        Raises ValueError if delegated_role is not actually delegated.
+        """
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def get_key(self, keyid: str) -> Key:
+        """Return the key object for the given keyid.
+
+        Raises ValueError if key is not found.
+        """
+        raise NotImplementedError
+
+    def verify_delegate(
+        self,
+        delegated_role: str,
+        payload: bytes,
+        signatures: Dict[str, Signature],
+    ) -> None:
+        """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
+            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``.
+        """
+        role = self.get_delegated_role(delegated_role)
+
+        # verify that delegated_metadata is signed by threshold of unique keys
+        signing_keys = set()
+        for keyid in role.keyids:
+            try:
+                key = self.get_key(keyid)
+            except ValueError:
+                logger.info("No key for keyid %s", keyid)
+                continue
+
+            if keyid not in signatures:
+                logger.info("No signature for keyid %s", keyid)
+                continue
+
+            sig = signatures[keyid]
+            try:
+                key.verify_signature(sig, payload)
+                signing_keys.add(keyid)
+            except sslib_exceptions.UnverifiedSignatureError:
+                logger.info("Key %s failed to verify %s", keyid, delegated_role)
+
+        if len(signing_keys) < role.threshold:
+            raise UnsignedMetadataError(
+                f"{delegated_role} was signed by {len(signing_keys)}/"
+                f"{role.threshold} keys",
+            )
+
+
+class Root(Signed, _DelegatorMixin):
     """A container for the signed part of root metadata.
 
     Parameters listed below are also instance attributes.
@@ -823,11 +865,7 @@ def get_delegated_role(self, delegated_role: str) -> Role:
 
         return self.roles[delegated_role]
 
-    def get_key(self, keyid: str) -> Key:
-        """Return the key object for the given keyid.
-
-        Raises ValueError if key is not found.
-        """
+    def get_key(self, keyid: str) -> Key:  # noqa: D102
         if keyid not in self.keys:
             raise ValueError(f"Key {keyid} not found")
 
@@ -1778,7 +1816,7 @@ def get_prefixed_paths(self) -> List[str]:
         return paths
 
 
-class Targets(Signed):
+class Targets(Signed, _DelegatorMixin):
     """A container for the signed part of targets metadata.
 
     Targets contains verifying information about target files and also
@@ -1952,11 +1990,7 @@ def get_delegated_role(self, delegated_role: str) -> Role:
 
         return role
 
-    def get_key(self, keyid: str) -> Key:
-        """Return the key object for the given keyid.
-
-        Raises ValueError if keyid is not found.
-        """
+    def get_key(self, keyid: str) -> Key:  # noqa: D102
         if self.delegations is None:
             raise ValueError("No delegations found")
         if keyid not in self.delegations.keys:
diff --git a/tuf/ngclient/_internal/trusted_metadata_set.py b/tuf/ngclient/_internal/trusted_metadata_set.py
