Merge pull request #1436 from jku/verify-delegate

Metadata API: Implement threshold verification

Author: joshuagl

tests/repository_data/repository/metadata/role2.json

@@ -2,15 +2,11 @@
  "signatures": [
   {
    "keyid": "c8022fa1e9b9cb239a6b362bbdffa9649e61ad2cb699d2e4bc4fdf7930a0e64a",
-   "sig": "6c32f8cc2c642803a7b3b022ede0cf727e82964c1aa934571ef366bd5050ed02cfe3fdfe5477c08d0cbcc2dd17bb786d37ab1ce2b27e01ad79faf087594e0300"
+   "sig": "75b196a224fd200e46e738b1216b3316c5384f61083872f8d14b8b0a378b2344e64b1a6f1a89a711206a66a0b199d65ac0e30fe15ddbc4de89fa8ff645f99403"
   }
  ],
  "signed": {
   "_type": "targets",
-  "delegations": {
-   "keys": {},
-   "roles": []
-  },
   "expires": "2030-01-01T00:00:00Z",
   "spec_version": "1.0.0",
   "targets": {},

tests/test_api.py

@@ -51,7 +51,8 @@
 )
 
 from securesystemslib.signer import (
-    SSlibSigner
+    SSlibSigner,
+    Signature
 )
 
 logger = logging.getLogger(__name__)
@@ -333,6 +334,71 @@ def test_metadata_timestamp(self):
         )
 
 
+    def test_metadata_verify_delegate(self):
+        root_path = os.path.join(self.repo_dir, 'metadata', 'root.json')
+        root = Metadata.from_file(root_path)
+        snapshot_path = os.path.join(
+                self.repo_dir, 'metadata', 'snapshot.json')
+        snapshot = Metadata.from_file(snapshot_path)
+        targets_path = os.path.join(
+                self.repo_dir, 'metadata', 'targets.json')
+        targets = Metadata.from_file(targets_path)
+        role1_path = os.path.join(
+                self.repo_dir, 'metadata', 'role1.json')
+        role1 = Metadata.from_file(role1_path)
+        role2_path = os.path.join(
+                self.repo_dir, 'metadata', 'role2.json')
+        role2 = Metadata.from_file(role2_path)
+
+        # test the expected delegation tree
+        root.verify_delegate('root', root)
+        root.verify_delegate('snapshot', snapshot)
+        root.verify_delegate('targets', targets)
+        targets.verify_delegate('role1', role1)
+        role1.verify_delegate('role2', role2)
+
+        # only root and targets can verify delegates
+        with self.assertRaises(TypeError):
+            snapshot.verify_delegate('snapshot', snapshot)
+        # verify fails for roles that are not delegated by delegator
+        with self.assertRaises(ValueError):
+            root.verify_delegate('role1', role1)
+        with self.assertRaises(ValueError):
+            targets.verify_delegate('targets', targets)
+        # verify fails when delegator has no delegations
+        with self.assertRaises(ValueError):
+            role2.verify_delegate('role1', role1)
+
+        # verify fails when delegate content is modified
+        expires = snapshot.signed.expires
+        snapshot.signed.bump_expiration()
+        with self.assertRaises(exceptions.UnsignedMetadataError):
+            root.verify_delegate('snapshot', snapshot)
+        snapshot.signed.expires = expires
+
+        # verify fails if roles keys do not sign the metadata
+        with self.assertRaises(exceptions.UnsignedMetadataError):
+            root.verify_delegate('timestamp', snapshot)
+
+        # Add a key to snapshot role, make sure the new sig fails to verify
+        ts_keyid = next(iter(root.signed.roles["timestamp"].keyids))
+        root.signed.add_key("snapshot", root.signed.keys[ts_keyid])
+        snapshot.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', snapshot)
+
+        # verify fails if threshold of signatures is not reached
+        root.signed.roles['snapshot'].threshold = 2
+        with self.assertRaises(exceptions.UnsignedMetadataError):
+            root.verify_delegate('snapshot', snapshot)
+
+        # verify succeeds when we correct the new signature and reach the
+        # threshold of 2 keys
+        snapshot.sign(SSlibSigner(self.keystore['timestamp']), append=True)
+        root.verify_delegate('snapshot', snapshot)
+
 
     def test_key_class(self):
         keys = {

tuf/api/metadata.py

@@ -17,6 +17,7 @@
 """
 import abc
 import io
+import logging
 import tempfile
 from collections import OrderedDict
 from datetime import datetime, timedelta
@@ -49,6 +50,8 @@
 
 # pylint: disable=too-many-lines
 
+logger = logging.getLogger(__name__)
+
 # We aim to support SPECIFICATION_VERSION and require the input metadata
 # files to have the same major version (the first number) as ours.
 SPECIFICATION_VERSION = ["1", "0", "19"]
@@ -266,6 +269,63 @@ def sign(
 
         return signature
 
+    def verify_delegate(
+        self,
+        delegated_role: str,
+        delegated_metadata: "Metadata",
+        signed_serializer: Optional[SignedSerializer] = None,
+    ) -> None:
+        """Verifies that 'delegated_metadata' is signed with the required
+        threshold of keys for the delegated role 'delegated_role'.
+
+        Args:
+            delegated_role: Name of the delegated role to verify
+            delegated_metadata: The Metadata object for the delegated role
+            signed_serializer: Optional; serializer used for delegate
+                serialization. Default is CanonicalJSONSerializer.
+
+        Raises:
+            UnsignedMetadataError: 'delegate' was not signed with required
+                threshold of keys for 'role_name'
+        """
+
+        # Find the keys and role in delegator metadata
+        role = None
+        if isinstance(self.signed, Root):
+            keys = self.signed.keys
+            role = self.signed.roles.get(delegated_role)
+        elif isinstance(self.signed, Targets):
+            if self.signed.delegations is None:
+                raise ValueError(f"No delegation found for {delegated_role}")
+
+            keys = self.signed.delegations.keys
+            roles = self.signed.delegations.roles
+            # Assume role names are unique in delegations.roles: #1426
+            # Find first role in roles with matching name (or None if no match)
+            role = next((r for r in roles if r.name == delegated_role), None)
+        else:
+            raise TypeError("Call is valid only on delegator metadata")
+
+        if role is None:
+            raise ValueError(f"No delegation found for {delegated_role}")
+
+        # verify that delegated_metadata is signed by threshold of unique keys
+        signing_keys = set()
+        for keyid in role.keyids:
+            key = keys[keyid]
+            try:
+                key.verify_signature(delegated_metadata, signed_serializer)
+                signing_keys.add(key.keyid)
+            except exceptions.UnsignedMetadataError:
+                logger.info("Key %s failed to verify %s", keyid, delegated_role)
+
+        if len(signing_keys) < role.threshold:
+            raise exceptions.UnsignedMetadataError(
+                f"{delegated_role} was signed by {len(signing_keys)}/"
+                f"{role.threshold} keys",
+                delegated_metadata.signed,
+            )
+
 
 class Signed(metaclass=abc.ABCMeta):
     """A base class for the signed part of TUF metadata.
@@ -967,7 +1027,7 @@ class Delegations:
 
     def __init__(
         self,
-        keys: Mapping[str, Key],
+        keys: Dict[str, Key],
         roles: List[DelegatedRole],
         unrecognized_fields: Optional[Mapping[str, Any]] = None,
     ) -> None: