Merge pull request #341 from MVrachev/gpg-signer

Add  GPGSigner implementation

Author: lukpueh

securesystemslib/gpg/functions.py

@@ -50,7 +50,7 @@ def create_signature(content, keyid=None, homedir=None):
     identified by the passed keyid from the gpg keyring at the passed homedir.
 
     The executed base command is defined in
-    securesystemslib.gpgp.constants.GPG_SIGN_COMMAND.
+    securesystemslib.gpg.constants.GPG_SIGN_COMMAND.
 
     NOTE: On not fully supported versions of GPG, i.e. versions below
     securesystemslib.gpg.constants.FULLY_SUPPORTED_MIN_VERSION the returned

securesystemslib/signer.py

@@ -7,6 +7,7 @@
 
 import abc
 import securesystemslib.keys as sslib_keys
+import securesystemslib.gpg.functions as gpg
 from typing import Any, Dict, Optional, Mapping
 
 
@@ -88,6 +89,60 @@ def to_dict(self) -> Dict:
 
 
 
+class GPGSignature(Signature):
+    """A container class containing information about a gpg signature.
+
+    Besides the signature, it also contains other meta information
+    needed to uniquely identify the key used to generate the signature.
+
+    Attributes:
+        keyid: HEX string used as a unique identifier of the key.
+        signature: HEX string representing the signature.
+        other_headers: HEX representation of additional GPG headers.
+    """
+    def __init__(
+        self,
+        keyid: str,
+        signature: str,
+        other_headers: str,
+    ):
+        super().__init__(keyid, signature)
+        self.other_headers = other_headers
+
+
+    @classmethod
+    def from_dict(cls, signature_dict: Dict) -> "Signature":
+        """Creates a GPGSignature object from its JSON/dict representation.
+
+        Args:
+            signature_dict: Dict containing valid "keyid", "signature" and
+                "other_fields" fields.
+
+        Raises:
+            KeyError: If any of the "keyid", "sig" or "other_headers" fields
+                are missing from the signature_dict.
+
+        Returns:
+            GPGSignature instance.
+        """
+
+        return cls(
+            signature_dict["keyid"],
+            signature_dict["sig"],
+            signature_dict["other_headers"]
+        )
+
+
+    def to_dict(self) -> Dict:
+        """Returns the JSON-serializable dictionary representation of self."""
+        return {
+            "keyid": self.keyid,
+            "signature": self.signature,
+            "other_headers": self.other_headers
+        }
+
+
+
 class Signer:
     """Signer interface created to support multiple signing implementations."""
 
@@ -160,3 +215,62 @@ def sign(self, payload: bytes) -> "Signature":
 
         sig_dict = sslib_keys.create_signature(self.key_dict, payload)
         return Signature(**sig_dict)
+
+
+
+class GPGSigner(Signer):
+    """A securesystemslib gpg implementation of the "Signer" interface.
+
+    Provides a sign method to generate a cryptographic signature with gpg, using
+    an RSA, DSA or EdDSA private key identified by the keyid on the instance.
+
+    Args:
+        keyid: The keyid of the gpg signing keyid. If not passed the default
+              key in the keyring is used.
+
+        homedir: Path to the gpg keyring. If not passed the default keyring
+            is used.
+
+    """
+    def __init__(
+            self, keyid: Optional[str] = None, homedir: Optional[str] = None
+    ):
+        self.keyid = keyid
+        self.homedir = homedir
+
+
+    def sign(self, payload: bytes) -> "GPGSignature":
+        """Signs a given payload by the key assigned to the GPGSigner instance.
+
+        Calls the gpg command line utility to sign the passed content with the
+        key identified by the passed keyid from the gpg keyring at the passed
+        homedir.
+
+        The executed base command is defined in
+        securesystemslib.gpg.constants.GPG_SIGN_COMMAND.
+
+        Arguments:
+            payload: The bytes to be signed.
+
+        Raises:
+            securesystemslib.exceptions.FormatError:
+                If the keyid was passed and does not match
+                securesystemslib.formats.KEYID_SCHEMA.
+
+            ValueError: the gpg command failed to create a valid signature.
+            OSError: the gpg command is not present or non-executable.
+            securesystemslib.exceptions.UnsupportedLibraryError: the gpg
+                command is not available, or the cryptography library is
+                not installed.
+            securesystemslib.gpg.exceptions.CommandError: the gpg command
+                returned a non-zero exit code.
+            securesystemslib.gpg.exceptions.KeyNotFoundError: the used gpg
+                version is not fully supported and no public key can be found
+                for short keyid.
+
+        Returns:
+            Returns a "GPGSignature" class instance.
+        """
+
+        sig_dict = gpg.create_signature(payload, self.keyid, self.homedir)
+        return GPGSignature(**sig_dict)

tests/test_signer.py

@@ -3,14 +3,20 @@
 """Test cases for "signer.py". """
 
 import copy
-import sys
+import os
 import unittest
+import tempfile
+import shutil
 
-import unittest
 import securesystemslib.formats
 import securesystemslib.keys as KEYS
 from securesystemslib.exceptions import FormatError, UnsupportedAlgorithmError
-from securesystemslib.signer import Signature, SSlibSigner
+from securesystemslib.signer import Signature, SSlibSigner, GPGSigner
+from securesystemslib.gpg.constants import HAVE_GPG
+from securesystemslib.gpg.functions import (
+    export_pubkey,
+    verify_signature as verify_sig
+)
 
 
 class TestSSlibSigner(unittest.TestCase):
@@ -95,6 +101,63 @@ def test_signature_eq_(self):
         sig_obj_2 = None
         self.assertNotEqual(sig_obj, sig_obj_2)
 
+@unittest.skipIf(not HAVE_GPG, "gpg not found")
+class TestGPGRSA(unittest.TestCase):
+    """Test RSA gpg signature creation and verification."""
+
+    @classmethod
+    def setUpClass(cls):
+        cls.default_keyid = "8465A1E2E0FB2B40ADB2478E18FB3F537E0C8A17"
+        cls.signing_subkey_keyid = "C5A0ABE6EC19D0D65F85E2C39BE9DF5131D924E9"
+
+        # Create directory to run the tests without having everything blow up.
+        cls.working_dir = os.getcwd()
+        cls.test_data = b'test_data'
+        cls.wrong_data = b'something malicious'
+
+        # Find demo files.
+        gpg_keyring_path = os.path.join(
+            os.path.dirname(os.path.realpath(__file__)), "gpg_keyrings", "rsa")
+
+        cls.test_dir = os.path.realpath(tempfile.mkdtemp())
+        cls.gnupg_home = os.path.join(cls.test_dir, "rsa")
+        shutil.copytree(gpg_keyring_path, cls.gnupg_home)
+        os.chdir(cls.test_dir)
+
+
+    @classmethod
+    def tearDownClass(cls):
+        """Change back to initial working dir and remove temp test directory."""
+
+        os.chdir(cls.working_dir)
+        shutil.rmtree(cls.test_dir)
+
+
+    def test_gpg_sign_and_verify_object_with_default_key(self):
+        """Create a signature using the default key on the keyring. """
+
+        signer = GPGSigner(homedir=self.gnupg_home)
+        signature = signer.sign(self.test_data)
+
+        signature_dict = signature.to_dict()
+        key_data = export_pubkey(self.default_keyid, self.gnupg_home)
+
+        self.assertTrue(verify_sig(signature_dict, key_data, self.test_data))
+        self.assertFalse(verify_sig(signature_dict, key_data, self.wrong_data))
+
+
+    def test_gpg_sign_and_verify_object(self):
+        """Create a signature using a specific key on the keyring. """
+
+        signer = GPGSigner(self.signing_subkey_keyid, self.gnupg_home)
+        signature = signer.sign(self.test_data)
+
+        signature_dict = signature.to_dict()
+        key_data = export_pubkey(self.signing_subkey_keyid, self.gnupg_home)
+
+        self.assertTrue(verify_sig(signature_dict, key_data, self.test_data))
+        self.assertFalse(verify_sig(signature_dict, key_data, self.wrong_data))
+
 
 # Run the unit tests.
 if __name__ == "__main__":