Merge pull request #1424 from jku/enforce-unique-sigs

Metadata API: Store signatures as dict

Author: Jussi Kukkonen

tests/test_api.py

@@ -175,7 +175,7 @@ def test_sign_verify(self):
         metadata_obj = Metadata.from_file(path)
 
         # ... it has a single existing signature,
-        self.assertTrue(len(metadata_obj.signatures) == 1)
+        self.assertEqual(len(metadata_obj.signatures), 1)
         # ... which is valid for the correct key.
         targets_key.verify_signature(metadata_obj)
         with self.assertRaises(tuf.exceptions.UnsignedMetadataError):
@@ -185,7 +185,7 @@ def test_sign_verify(self):
         # Append a new signature with the unrelated key and assert that ...
         metadata_obj.sign(sslib_signer, append=True)
         # ... there are now two signatures, and
-        self.assertTrue(len(metadata_obj.signatures) == 2)
+        self.assertEqual(len(metadata_obj.signatures), 2)
         # ... both are valid for the corresponding keys.
         targets_key.verify_signature(metadata_obj)
         snapshot_key.verify_signature(metadata_obj)
@@ -194,7 +194,7 @@ def test_sign_verify(self):
         # Create and assign (don't append) a new signature and assert that ...
         metadata_obj.sign(sslib_signer, append=False)
         # ... there now is only one signature,
-        self.assertTrue(len(metadata_obj.signatures) == 1)
+        self.assertEqual(len(metadata_obj.signatures), 1)
         # ... valid for that key.
         timestamp_key.verify_signature(metadata_obj)
         with self.assertRaises(tuf.exceptions.UnsignedMetadataError):
@@ -236,6 +236,12 @@ def test_metadata_base(self):
         self.assertFalse(is_expired)
         md.signed.expires = expires
 
+        # Test deserializing metadata with non-unique signatures:
+        data = md.to_dict()
+        data["signatures"].append({"keyid": data["signatures"][0]["keyid"], "sig": "foo"})
+        with self.assertRaises(ValueError):
+            Metadata.from_dict(data)
+
 
     def test_metafile_class(self):
         # Test from_dict and to_dict with all attributes.

tuf/api/metadata.py

@@ -17,6 +17,7 @@
 """
 import abc
 import tempfile
+from collections import OrderedDict
 from datetime import datetime, timedelta
 from typing import Any, ClassVar, Dict, List, Mapping, Optional, Tuple, Type
 
@@ -48,12 +49,13 @@ class Metadata:
         signed: A subclass of Signed, which has the actual metadata payload,
             i.e. one of Targets, Snapshot, Timestamp or Root.
 
-        signatures: A list of Securesystemslib Signature objects, each signing
-            the canonical serialized representation of 'signed'.
-
+        signatures: An ordered dictionary of keyids to Signature objects, each
+            signing the canonical serialized representation of 'signed'.
     """
 
-    def __init__(self, signed: "Signed", signatures: List[Signature]) -> None:
+    def __init__(
+        self, signed: "Signed", signatures: "OrderedDict[str, Signature]"
+    ):
         self.signed = signed
         self.signatures = signatures
 
@@ -89,10 +91,15 @@ def from_dict(cls, metadata: Dict[str, Any]) -> "Metadata":
         else:
             raise ValueError(f'unrecognized metadata type "{_type}"')
 
-        signatures = []
-        for signature in metadata.pop("signatures"):
-            signature_obj = Signature.from_dict(signature)
-            signatures.append(signature_obj)
+        # Make sure signatures are unique
+        signatures: "OrderedDict[str, Signature]" = OrderedDict()
+        for sig_dict in metadata.pop("signatures"):
+            sig = Signature.from_dict(sig_dict)
+            if sig.keyid in signatures:
+                raise ValueError(
+                    f"Multiple signatures found for keyid {sig.keyid}"
+                )
+            signatures[sig.keyid] = sig
 
         return cls(
             signed=inner_cls.from_dict(metadata.pop("signed")),
@@ -164,9 +171,7 @@ def from_bytes(
     def to_dict(self) -> Dict[str, Any]:
         """Returns the dict representation of self."""
 
-        signatures = []
-        for sig in self.signatures:
-            signatures.append(sig.to_dict())
+        signatures = [sig.to_dict() for sig in self.signatures.values()]
 
         return {"signatures": signatures, "signed": self.signed.to_dict()}
 
@@ -244,10 +249,10 @@ def sign(
 
         signature = signer.sign(signed_serializer.serialize(self.signed))
 
-        if append:
-            self.signatures.append(signature)
-        else:
-            self.signatures = [signature]
+        if not append:
+            self.signatures.clear()
+
+        self.signatures[signature.keyid] = signature
 
         return signature
 
@@ -453,9 +458,8 @@ def verify_signature(
                 level components: Issue #1351
         """
         try:
-            sigs = metadata.signatures
-            signature = next(sig for sig in sigs if sig.keyid == self.keyid)
-        except StopIteration:
+            signature = metadata.signatures[self.keyid]
+        except KeyError:
             raise exceptions.UnsignedMetadataError(
                 f"no signature for key {self.keyid} found in metadata",
                 metadata.signed,