Sign and verify

securesystemslib/exceptions.py

@@ -109,7 +109,13 @@ class InvalidConfigurationError(Error):
   """If a configuration object does not match the expected format."""
   pass
 
+
 class StorageError(Error):
-  """Indicate an error occured during interaction with an abstracted storage
+  """Indicate an error occurred during interaction with an abstracted storage
      backend."""
   pass
+
+
+class SignatureVerificationError(Error):
+  """Indicate an error occurred during signature verification of DSSE."""
+  pass

securesystemslib/key.py

@@ -13,6 +13,20 @@ class Key:
     """Key interface created to support multiple verify implementations."""
 
     __metaclass__ = abc.ABCMeta
+    keyid: str
+
+    @abc.abstractmethod
+    def match_keyid(self, keyid: str) -> bool:
+        """Matches a keyid with the keyid present in the Key instance.
+
+        Arguments:
+            keyid: Key identifier
+
+        Returns:
+            Boolean. True if keyid matches for the Key, False otherwise.
+        """
+
+        raise NotImplementedError
 
     @abc.abstractmethod
     def verify(self, signature: Signature, payload: bytes) -> bool:
@@ -54,7 +68,7 @@ class SSlibKey(Key):
 
     keytype: str
     scheme: str
-    keyval: Dict[str, str]
+    keyval: Dict[str, Dict]
     keyid: str
     unrecognized_fields: Dict[str, Any] = field(default_factory=dict)
 
@@ -122,6 +136,18 @@ def to_securesystemslib_key(self) -> Dict[str, Any]:
             "keyval": self.keyval,
         }
 
+    def match_keyid(self, keyid: str) -> bool:
+        """Matches a keyid with the keyid present in the Key instance.
+
+        Arguments:
+            keyid: Key identifier
+
+        Returns:
+            Boolean. True if keyid matches for the Key, False otherwise.
+        """
+
+        return self.keyid == keyid
+
     def verify(self, signature: Signature, payload: bytes) -> bool:
         """Verifies a given payload by the key assigned to the SSlibKey instance.
 
@@ -227,6 +253,22 @@ def from_keyring(cls, keyid, homedir=None):
         pubkey_dict = gpg.export_pubkey(keyid, homedir)
         return cls.from_dict(pubkey_dict)
 
+    def match_keyid(self, keyid: str) -> bool:
+        """Matches a keyid with the keyids present in the Key instance.
+
+        Arguments:
+            keyid: Key identifier
+
+        Returns:
+            Boolean. True if keyid matches for the Key, False otherwise.
+        """
+
+        if self.keyid == keyid:
+            return True
+        if self.subkeys and keyid in self.subkeys:
+            return True
+        return False
+
     def verify(self, signature: GPGSignature, payload: bytes) -> bool:
         """Verifies a given payload by the key assigned to the GPGKey instance.
 

securesystemslib/metadata.py

@@ -1,36 +1,33 @@
 """Dead Simple Signing Envelope
 """
 
-from typing import Any, List
+import logging
+from typing import Any, Dict, List
 
 from securesystemslib import exceptions, formats
-from securesystemslib.signer import GPGSignature, Signature
+from securesystemslib.key import Key
+from securesystemslib.signer import GPGSignature, Signature, Signer
 from securesystemslib.util import b64dec, b64enc
 
+logger = logging.getLogger(__name__)
+
 
 class Envelope:
-    """
-    DSSE Envelope to provide interface for signing arbitrary data.
+    """DSSE Envelope to provide interface for signing arbitrary data.
 
     Attributes:
-        payload: Arbitrary byte sequence of serialized body
-        payload_type: string that identifies how to interpret payload
-        signatures: List of Signature and GPG Signature
-
-    Methods:
-        from_dict(cls, data):
-            Creates a Signature object from its JSON/dict representation.
-
-        to_dict(self):
-            Returns the JSON-serializable dictionary representation of self.
+        payload: Arbitrary byte sequence of serialized body.
+        payload_type: string that identifies how to interpret payload.
+        signatures: list of Signature and GPGSignature.
 
     """
 
-    payload: bytes
-    payload_type: str
-    signatures: List[Signature]
-
-    def __init__(self, payload, payload_type, signatures):
+    def __init__(
+        self,
+        payload: bytes,
+        payload_type: str,
+        signatures: List[Signature]
+    ):
         self.payload = payload
         self.payload_type = payload_type
         self.signatures = signatures
@@ -84,7 +81,6 @@ def to_dict(self) -> dict:
             "signatures": [signature.to_dict() for signature in self.signatures],
         }
 
-    @property
     def pae(self) -> bytes:
         """Pre-Auth-Encoding byte sequence of self."""
 
@@ -94,3 +90,74 @@ def pae(self) -> bytes:
             len(self.payload),
             self.payload,
         )
+
+    def sign(self, signer: Signer) -> Signature:
+        """Sign the payload and create the signature.
+
+        Arguments:
+            signer: A "Signer" class instance.
+
+        Returns:
+            A "Signature" instance.
+        """
+
+        signature = signer.sign(self.pae())
+        self.signatures.append(signature)
+
+        return signature
+
+    def verify(self, keys: List[Key], threshold: int) -> Dict[str, Key]:
+        """Verify the payload with the provided Keys.
+
+        Arguments:
+            keys: A list of public keys to verify the signatures.
+            threshold: Number of signatures needed to pass the verification.
+
+        Raises:
+            ValueError: If "threshold" is not valid.
+            SignatureVerificationError: If the enclosed signatures do not pass
+                the verification.
+
+        Note:
+            Mandating keyid in signatures and matching them with keyid of Key
+            in order to consider them for verification, is not DSSE spec
+            compliant (Issue #416).
+
+        Returns:
+            accepted_keys: A dict of unique public keys.
+        """
+
+        accepted_keys = {}
+        pae = self.pae()
+
+        # checks for threshold value.
+        if threshold <= 0:
+            raise ValueError("Threshold must be greater than 0")
+
+        if len(keys) < threshold:
+            raise ValueError("Number of keys can't be less than threshold")
+
+        for signature in self.signatures:
+            for key in keys:
+                # If Signature keyid doesn't match with Key, skip.
+                if not key.match_keyid(signature.keyid):
+                    continue
+
+                # If a key verifies the signature, we exit and use the result.
+                # TODO: Exception handling in `Key.verify`.
+                # See https://github.com/in-toto/securesystemslib/issues/8.
+                if key.verify(signature, pae):
+                    accepted_keys[key.keyid] = key
+                    break
+
+            # Break, if amount of recognized_signer are more than threshold.
+            if len(accepted_keys) >= threshold:
+                break
+
+        if threshold > len(accepted_keys):
+            raise exceptions.SignatureVerificationError(
+                "Accepted signatures do not match threshold,"
+                f" Found: {len(accepted_keys)}, Expected {threshold}"
+            )
+
+        return accepted_keys

tests/test_gpg.py

@@ -25,6 +25,8 @@
 import tempfile
 import unittest
 
+from typing import List
+
 if sys.version_info >= (3, 3):
   from unittest.mock import patch # pylint: disable=no-name-in-module,import-error
 else:
@@ -55,8 +57,12 @@
 from securesystemslib.gpg.exceptions import (PacketParsingError,
     PacketVersionNotSupportedError, SignatureAlgorithmNotSupportedError,
     KeyNotFoundError, CommandError, KeyExpirationError)
+from securesystemslib.exceptions import SignatureVerificationError
 from securesystemslib.formats import (GPG_PUBKEY_SCHEMA,
     ANY_PUBKEY_DICT_SCHEMA)
+from securesystemslib.key import GPGKey, Key
+from securesystemslib.metadata import Envelope
+from securesystemslib.signer import GPGSignature, GPGSigner
 
 
 class GPGTestUtils:
@@ -649,6 +655,31 @@ def test_verify_signature_with_expired_key(self):
         "\nexpected: {}"
         "\ngot:      {}".format(expected, ctx.exception))
 
+  def test_dsse_envelope(self):
+    """Test signing and verifying DSSE signatures."""
+
+    # Create the DSSE Envelope.
+    envelope = Envelope(
+      payload=b"hello world",
+      payload_type="http://example.com/HelloWorld",
+      signatures=[],
+    )
+
+    # Create a GPGSigner and create a DSSE signature.
+    gpg_signer = GPGSigner(homedir=self.gnupg_home)
+    gpg_signature = envelope.sign(gpg_signer)
+    self.assertIsInstance(gpg_signature, GPGSignature)
+
+    # Create a GPGKey and verify the DSSE signature.
+    gpgkey = GPGKey.from_keyring(keyid=self.default_keyid, homedir=self.gnupg_home)
+    key_list: List[Key] = [gpgkey]
+    envelope.verify(key_list, 1)
+
+    # Duplicate GPGKey.
+    new_key_list = key_list + key_list
+    with self.assertRaises(SignatureVerificationError):
+      envelope.verify(new_key_list, 2)
+
 
 @unittest.skipIf(not HAVE_GPG, "gpg not found")
 class TestGPGDSA(unittest.TestCase):
@@ -733,6 +764,26 @@ def test_gpg_sign_and_verify_object(self):
     self.assertTrue(verify_signature(signature, key_data, test_data))
     self.assertFalse(verify_signature(signature, key_data, wrong_data))
 
+  def test_dsse_envelope(self):
+    """Test signing and verifying DSSE signatures."""
+
+    # Create the DSSE Envelope.
+    envelope = Envelope(
+      payload=b"hello world",
+      payload_type="http://example.com/HelloWorld",
+      signatures=[],
+    )
+
+    # Create a GPGSigner and create a DSSE signature.
+    gpg_signer = GPGSigner(homedir=self.gnupg_home)
+    gpg_signature = envelope.sign(gpg_signer)
+    self.assertIsInstance(gpg_signature, GPGSignature)
+
+    # Create a GPGKey and verify the DSSE signature.
+    gpgkey = GPGKey.from_keyring(keyid=self.default_keyid, homedir=self.gnupg_home)
+    key_list: List[Key] = [gpgkey]
+    envelope.verify(key_list, 1)
+
 
 
 @unittest.skipIf(not HAVE_GPG, "gpg not found")
@@ -808,6 +859,26 @@ def test_verify_short_signature(self):
     key = export_pubkey(self.default_keyid, homedir=self.gnupg_home)
     self.assertTrue(verify_signature(signature, key, test_data))
 
+  def test_dsse_envelope(self):
+    """Test signing and verifying DSSE signatures."""
+
+    # Create the DSSE Envelope.
+    envelope = Envelope(
+      payload=b"hello world",
+      payload_type="http://example.com/HelloWorld",
+      signatures=[],
+    )
+
+    # Create a GPGSigner and create a DSSE signature.
+    gpg_signer = GPGSigner(homedir=self.gnupg_home)
+    gpg_signature = envelope.sign(gpg_signer)
+    self.assertIsInstance(gpg_signature, GPGSignature)
+
+    # Create a GPGKey and verify the DSSE signature.
+    gpgkey = GPGKey.from_keyring(keyid=self.default_keyid, homedir=self.gnupg_home)
+    key_list: List[Key] = [gpgkey]
+    envelope.verify(key_list, 1)
+
 
 if __name__ == "__main__":
   unittest.main()