#!/usr/bin/env python """ test_gpg.py Santiago Torres-Arias Lukas Puehringer Nov 15, 2017 See LICENSE for licensing information. Test gpg/pgp-related functions. """ import os import sys import shutil import tempfile import unittest if sys.version_info >= (3, 3): from unittest.mock import patch # pylint: disable=no-name-in-module,import-error else: from mock import patch # pylint: disable=import-error from six import string_types from copy import deepcopy from collections import OrderedDict import cryptography.hazmat.primitives.serialization as serialization import cryptography.hazmat.backends as backends import cryptography.hazmat.primitives.hashes as hashing from securesystemslib import process from securesystemslib.gpg.functions import (create_signature, export_pubkey, verify_signature, export_pubkeys) from securesystemslib.gpg.util import (get_version, is_version_fully_supported, get_hashing_class, parse_packet_header, parse_subpacket_header) from securesystemslib.gpg.rsa import create_pubkey as rsa_create_pubkey from securesystemslib.gpg.dsa import create_pubkey as dsa_create_pubkey from securesystemslib.gpg.eddsa import create_pubkey as eddsa_create_pubkey from securesystemslib.gpg.eddsa import ED25519_SIG_LENGTH from securesystemslib.gpg.common import (parse_pubkey_payload, parse_pubkey_bundle, get_pubkey_bundle, _assign_certified_key_info, _get_verified_subkeys, parse_signature_packet) from securesystemslib.gpg.constants import (SHA1, SHA256, SHA512, GPG_EXPORT_PUBKEY_COMMAND, PACKET_TYPE_PRIMARY_KEY, PACKET_TYPE_USER_ID, PACKET_TYPE_USER_ATTR, PACKET_TYPE_SUB_KEY, HAVE_GPG) from securesystemslib.gpg.exceptions import (PacketParsingError, PacketVersionNotSupportedError, SignatureAlgorithmNotSupportedError, KeyNotFoundError, CommandError, KeyExpirationError) from securesystemslib.formats import (GPG_PUBKEY_SCHEMA, ANY_PUBKEY_DICT_SCHEMA) @unittest.skipIf(not HAVE_GPG, "gpg not found") class TestUtil(unittest.TestCase): """Test util functions. """ def test_version_utils_return_types(self): """Run dummy tests for coverage. """ self.assertTrue(isinstance(get_version(), string_types)) self.assertTrue(isinstance(is_version_fully_supported(), bool)) def test_get_hashing_class(self): # Assert return expected hashing class expected_hashing_class = [hashing.SHA1, hashing.SHA256, hashing.SHA512] for idx, hashing_id in enumerate([SHA1, SHA256, SHA512]): result = get_hashing_class(hashing_id) self.assertEqual(result, expected_hashing_class[idx]) # Assert raises ValueError with non-supported hashing id with self.assertRaises(ValueError): get_hashing_class("bogus_hashing_id") def test_parse_packet_header(self): """Test parse_packet_header with manually crafted data. """ data_list = [ ## New format packet length with mock packet type 100001 # one-octet length, header len: 2, body len: 0 to 191 [0b01100001, 0], [0b01100001, 191], # two-octet length, header len: 3, body len: 192 to 8383 [0b01100001, 192, 0], [0b01100001, 223, 255], # five-octet length, header len: 6, body len: 0 to 4,294,967,295 [0b01100001, 255, 0, 0, 0, 0], [0b01100001, 255, 255, 255, 255, 255], ## Old format packet lengths with mock packet type 1001 # one-octet length, header len: 2, body len: 0 to 255 [0b00100100, 0], [0b00100100, 255], # two-octet length, header len: 3, body len: 0 to 65,535 [0b00100101, 0, 0], [0b00100101, 255, 255], # four-octet length, header len: 5, body len: 0 to 4,294,967,295 [0b00100110, 0, 0, 0, 0, 0], [0b00100110, 255, 255, 255, 255, 255], ] # packet_type | header_len | body_len | packet_len expected = [ (33, 2, 0, 2), (33, 2, 191, 193), (33, 3, 192, 195), (33, 3, 8383, 8386), (33, 6, 0, 6), (33, 6, 4294967295, 4294967301), (9, 2, 0, 2), (9, 2, 255, 257), (9, 3, 0, 3), (9, 3, 65535, 65538), (9, 5, 0, 5), (9, 5, 4294967295, 4294967300), ] for idx, data in enumerate(data_list): result = parse_packet_header(bytearray(data)) self.assertEqual(result, expected[idx]) # New Format Packet Lengths with Partial Body Lengths range for second_octet in [224, 254]: with self.assertRaises(PacketParsingError): parse_packet_header(bytearray([0b01100001, second_octet])) # Old Format Packet Lengths with indeterminate length (length type 3) with self.assertRaises(PacketParsingError): parse_packet_header(bytearray([0b00100111])) # Get expected type parse_packet_header(bytearray([0b01100001, 0]), expected_type=33) # Raise with unexpected type with self.assertRaises(PacketParsingError): parse_packet_header(bytearray([0b01100001, 0]), expected_type=34) def test_parse_subpacket_header(self): """Test parse_subpacket_header with manually crafted data. """ # All items until last item encode the length of the subpacket, # the last item encodes the mock subpacket type. data_list = [ # length of length 1, subpacket length 0 to 191 [0, 33], # NOTE: Nonsense 0-length [191, 33], # # length of length 2, subpacket length 192 to 16,319 [192, 0, 33], [254, 255, 33], # # length of length 5, subpacket length 0 to 4,294,967,295 [255, 0, 0, 0, 0, 33], # NOTE: Nonsense 0-length [255, 255, 255, 255, 255, 33], ] # packet_type | header_len | body_len | packet_len expected = [ (33, 2, -1, 1), # NOTE: Nonsense negative payload (33, 2, 190, 192), (33, 3, 191, 194), (33, 3, 16318, 16321), (33, 6, -1, 5), # NOTE: Nonsense negative payload (33, 6, 4294967294, 4294967300) ] for idx, data in enumerate(data_list): result = parse_subpacket_header(bytearray(data)) self.assertEqual(result, expected[idx]) @unittest.skipIf(not HAVE_GPG, "gpg not found") class TestCommon(unittest.TestCase): """Test common functions of the securesystemslib.gpg module. """ @classmethod def setUpClass(self): gpg_keyring_path = os.path.join( os.path.dirname(os.path.realpath(__file__)), "gpg_keyrings", "rsa") homearg = "--homedir {}".format(gpg_keyring_path).replace("\\", "/") # Load test raw public key bundle from rsa keyring, used to construct # erroneous gpg data in tests below. keyid = "F557D0FF451DEF45372591429EA70BD13D883381" cmd = GPG_EXPORT_PUBKEY_COMMAND.format(keyid=keyid, homearg=homearg) proc = process.run(cmd, stdout=process.PIPE, stderr=process.PIPE) self.raw_key_data = proc.stdout self.raw_key_bundle = parse_pubkey_bundle(self.raw_key_data) # Export pubkey bundle with expired key for key expiration tests keyid = "E8AC80C924116DABB51D4B987CB07D6D2C199C7C" cmd = GPG_EXPORT_PUBKEY_COMMAND.format(keyid=keyid, homearg=homearg) proc = process.run(cmd, stdout=process.PIPE, stderr=process.PIPE) self.raw_expired_key_bundle = parse_pubkey_bundle(proc.stdout) def test_parse_pubkey_payload_errors(self): """ Test parse_pubkey_payload errors with manually crafted data. """ # passed data | expected error | expected error message test_data = [ (None, ValueError, "empty pubkey"), (bytearray([0x03]), PacketVersionNotSupportedError, "packet version '3' not supported"), (bytearray([0x04, 0, 0, 0, 0, 255]), SignatureAlgorithmNotSupportedError, "Signature algorithm '255' not supported") ] for data, error, error_str in test_data: with self.assertRaises(error) as ctx: parse_pubkey_payload(data) self.assertTrue(error_str in str(ctx.exception)) def test_parse_pubkey_bundle_errors(self): """Test parse_pubkey_bundle errors with manually crafted data partially based on real gpg key data (see self.raw_key_bundle). """ # Extract sample (legitimate) user ID packet and pass as first packet to # raise first packet must be primary key error user_id_packet = list(self.raw_key_bundle[PACKET_TYPE_USER_ID].keys())[0] # Extract sample (legitimate) primary key packet and pass as first two # packets to raise unexpected second primary key error primary_key_packet = self.raw_key_bundle[PACKET_TYPE_PRIMARY_KEY]["packet"] # Create incomplete packet to re-raise header parsing IndexError as # PacketParsingError incomplete_packet = bytearray([0b01111111]) # passed data | expected error message test_data = [ (None, "empty gpg data"), (user_id_packet, "must be a primary key"), (primary_key_packet + primary_key_packet, "Unexpected primary key"), (incomplete_packet, "index out of range") ] for data, error_str in test_data: with self.assertRaises(PacketParsingError) as ctx: parse_pubkey_bundle(data) self.assertTrue(error_str in str(ctx.exception)) # Create empty packet of unsupported type 66 (bit 0-5) and length 0 and # pass as second packet to provoke skipping of unsupported packet unsupported_packet = bytearray([0b01111111, 0]) with patch("securesystemslib.gpg.common.log") as mock_log: parse_pubkey_bundle(primary_key_packet + unsupported_packet) self.assertTrue("Ignoring gpg key packet '63'" in mock_log.info.call_args[0][0]) def test_parse_pubkey_bundle(self): """Assert presence of packets expected returned from `parse_pubkey_bundle` for specific test key). See ``` gpg --homedir tests/gpg_keyrings/rsa/ --export 9EA70BD13D883381 | \ gpg --list-packets ``` """ # Expect parsed primary key matching GPG_PUBKEY_SCHEMA self.assertTrue(GPG_PUBKEY_SCHEMA.matches( self.raw_key_bundle[PACKET_TYPE_PRIMARY_KEY]["key"])) # Parse corresponding raw packet for comparison _, header_len, _, _ = parse_packet_header( self.raw_key_bundle[PACKET_TYPE_PRIMARY_KEY]["packet"]) # pylint: disable=unsubscriptable-object parsed_raw_packet = parse_pubkey_payload(bytearray( self.raw_key_bundle[PACKET_TYPE_PRIMARY_KEY]["packet"][header_len:])) # And compare self.assertDictEqual( self.raw_key_bundle[PACKET_TYPE_PRIMARY_KEY]["key"], parsed_raw_packet) # Expect one primary key signature (revocation signature) self.assertEqual( len(self.raw_key_bundle[PACKET_TYPE_PRIMARY_KEY]["signatures"]), 1) # Expect one User ID packet, one User Attribute packet and one Subkey, # each with correct data for _type in [PACKET_TYPE_USER_ID, PACKET_TYPE_USER_ATTR, PACKET_TYPE_SUB_KEY]: # Of each type there is only one packet self.assertTrue(len(self.raw_key_bundle[_type]) == 1) # The raw packet is stored as key in the per-packet type collection raw_packet = next(iter(self.raw_key_bundle[_type])) # Its values are the raw packets header and body length self.assertEqual(len(raw_packet), self.raw_key_bundle[_type][raw_packet]["header_len"] + self.raw_key_bundle[_type][raw_packet]["body_len"]) # and one self-signature self.assertEqual( len(self.raw_key_bundle[_type][raw_packet]["signatures"]), 1) def test_assign_certified_key_info_errors(self): """Test _assign_certified_key_info errors with manually crafted data based on real gpg key data (see self.raw_key_bundle). """ # Replace legitimate user certifacte with a bogus packet wrong_cert_bundle = deepcopy(self.raw_key_bundle) packet, packet_data = wrong_cert_bundle[PACKET_TYPE_USER_ID].popitem() packet_data["signatures"] = [bytearray([0b01111111, 0])] wrong_cert_bundle[PACKET_TYPE_USER_ID][packet] = packet_data # Replace primary key id with a non-associated keyid wrong_keyid_bundle = deepcopy(self.raw_key_bundle) wrong_keyid_bundle[PACKET_TYPE_PRIMARY_KEY]["key"]["keyid"] = \ "8465A1E2E0FB2B40ADB2478E18FB3F537E0C8A17" # Remove a byte in user id packet to make signature verification fail invalid_cert_bundle = deepcopy(self.raw_key_bundle) packet, packet_data = invalid_cert_bundle[PACKET_TYPE_USER_ID].popitem() packet = packet[:-1] invalid_cert_bundle[PACKET_TYPE_USER_ID][packet] = packet_data test_data = [ # Skip and log parse_signature_packet error (wrong_cert_bundle, "Expected packet 2, but got 63 instead"), # Skip and log signature packet that doesn't match primary key id (wrong_keyid_bundle, "Ignoring User ID certificate issued by"), # Skip and log invalid signature (invalid_cert_bundle, "Ignoring invalid User ID self-certificate") ] for bundle, expected_msg in test_data: with patch("securesystemslib.gpg.common.log") as mock_log: _assign_certified_key_info(bundle) msg = str(mock_log.info.call_args[0][0]) self.assertTrue(expected_msg in msg, "'{}' not in '{}'".format(expected_msg, msg)) def test_assign_certified_key_info_expiration(self): """Test assignment of key expiration date in gpg.common._assign_certified_key_info using real gpg data (with ambiguity resolution / prioritization). # FIXME: Below tests are missing proper assertions for which User ID self-certificate is considered for the expiration date. Reasons are: - gpg does not let you (easily) modify individual expiration dates of User IDs (changing one changes all), hence we cannot assert the chosen packet by the particular date - _assign_certified_key_info first verifies all self-certificates and then only considers successfully verified ones, hence we cannot modify the certificate data, before passing it to _assign_certified_key_info IMO the best solution is a better separation of concerns, e.g. separate self-certificate verification and packet prioritization. """ # Test ambiguity resolution scheme with 3 User IDs # :user ID packet: "Test Expiration I " # :user ID packet: "Test Expiration II " # :user ID packet: "Test Expiration III " # User ID packets are ordered by their creation time in ascending order. # "Test Expiration II" has the primary user ID flag set and therefor has # the highest priority. key = _assign_certified_key_info(self.raw_expired_key_bundle) self.assertTrue(key["validity_period"] == 87901) # ~ 1 day # Test ambiguity resolution scheme with 2 User IDs # :user ID packet: "Test Expiration III " # :user ID packet: "Test Expiration I " # User ID packets are ordered by their creation time in descending order. # Neither packet has the primary user ID flag set. # "Test Expiration III" has the highest priority. raw_key_bundle = deepcopy(self.raw_expired_key_bundle) user_id_items = list(reversed(raw_key_bundle[PACKET_TYPE_USER_ID].items())) del user_id_items[1] raw_key_bundle[PACKET_TYPE_USER_ID] = OrderedDict(user_id_items) key = _assign_certified_key_info(raw_key_bundle) self.assertTrue(key["validity_period"] == 87901) # ~ 1 day def test_get_verified_subkeys_errors(self): """Test _get_verified_subkeys errors with manually crafted data based on real gpg key data (see self.raw_key_bundle). """ # Tamper with subkey (change version number) to trigger key parsing error bad_subkey_bundle = deepcopy(self.raw_key_bundle) packet, packet_data = bad_subkey_bundle[PACKET_TYPE_SUB_KEY].popitem() packet = bytes(packet[:packet_data["header_len"]] + bytearray([0x03]) + packet[packet_data["header_len"]+1:]) bad_subkey_bundle[PACKET_TYPE_SUB_KEY][packet] = packet_data # Add bogus sig to trigger sig parsing error wrong_sig_bundle = deepcopy(self.raw_key_bundle) packet, packet_data = wrong_sig_bundle[PACKET_TYPE_SUB_KEY].popitem() # NOTE: We can't only pass the bogus sig, because that would also trigger # the not enough sigs error (see not_enough_sigs_bundle) and mock only # lets us assert for the most recent log statement packet_data["signatures"].append(bytearray([0b01111111, 0])) wrong_sig_bundle[PACKET_TYPE_SUB_KEY][packet] = packet_data # Remove sigs to trigger not enough sigs error not_enough_sigs_bundle = deepcopy(self.raw_key_bundle) packet, packet_data = not_enough_sigs_bundle[PACKET_TYPE_SUB_KEY].popitem() packet_data["signatures"] = [] not_enough_sigs_bundle[PACKET_TYPE_SUB_KEY][packet] = packet_data # Duplicate sig to trigger wrong amount signatures too_many_sigs_bundle = deepcopy(self.raw_key_bundle) packet, packet_data = too_many_sigs_bundle[PACKET_TYPE_SUB_KEY].popitem() packet_data["signatures"] = packet_data["signatures"] * 2 too_many_sigs_bundle[PACKET_TYPE_SUB_KEY][packet] = packet_data # Tamper with primary key to trigger signature verification error invalid_sig_bundle = deepcopy(self.raw_key_bundle) invalid_sig_bundle[PACKET_TYPE_PRIMARY_KEY]["packet"] = \ invalid_sig_bundle[PACKET_TYPE_PRIMARY_KEY]["packet"][:-1] test_data = [ (bad_subkey_bundle, "Pubkey packet version '3' not supported"), (wrong_sig_bundle, "Expected packet 2, but got 63 instead"), (not_enough_sigs_bundle, "wrong amount of key binding signatures (0)"), (too_many_sigs_bundle, "wrong amount of key binding signatures (2)"), (invalid_sig_bundle, "invalid key binding signature"), ] for bundle, expected_msg in test_data: with patch("securesystemslib.gpg.common.log") as mock_log: _get_verified_subkeys(bundle) msg = str(mock_log.info.call_args[0][0]) self.assertTrue(expected_msg in msg, "'{}' not in '{}'".format(expected_msg, msg)) def test_get_verified_subkeys(self): """Test correct assignment of subkey expiration date in gpg.common._get_verified_subkeys using real gpg data. """ subkeys = _get_verified_subkeys(self.raw_expired_key_bundle) # Test subkey with validity period 175451, i.e. ~ 2 days self.assertTrue(subkeys["0ce427fa3f0f50bc83a4a760ed95e1581691db4d"].get( "validity_period") == 175451) # Test subkey without validity period, i.e. it does not expire self.assertTrue(subkeys["70cfabf1e2f1dc60ac5c7bca10cd20d3d5bcb6ef"].get( "validity_period") == None) def test_get_pubkey_bundle_errors(self): """Test correct error raising in get_pubkey_bundle. """ # Call without key data with self.assertRaises(KeyNotFoundError): get_pubkey_bundle(None, "deadbeef") # Pass wrong keyid with valid gpg data to trigger KeyNotFoundError. not_associated_keyid = "8465A1E2E0FB2B40ADB2478E18FB3F537E0C8A17" with self.assertRaises(KeyNotFoundError): get_pubkey_bundle(self.raw_key_data, not_associated_keyid) def test_parse_signature_packet_errors(self): """Test parse_signature_packet errors with manually crafted data. """ # passed data | expected error message test_data = [ (bytearray([0b01000010, 1, 255]), "Signature version '255' not supported"), (bytearray([0b01000010, 2, 4, 255]), "Signature type '255' not supported"), (bytearray([0b01000010, 3, 4, 0, 255]), "Signature algorithm '255' not supported"), (bytearray([0b01000010, 4, 4, 0, 1, 255]), "Hash algorithm '255' not supported"), ] for data, expected_error_str in test_data: with self.assertRaises(ValueError) as ctx: parse_signature_packet(data) self.assertTrue(expected_error_str in str(ctx.exception), "'{}' not in '{}'".format(expected_error_str, str(ctx.exception))) @unittest.skipIf(not HAVE_GPG, "gpg not found") class TestGPGRSA(unittest.TestCase): """Test signature creation, verification and key export from the gpg module""" default_keyid = "8465A1E2E0FB2B40ADB2478E18FB3F537E0C8A17" signing_subkey_keyid = "C5A0ABE6EC19D0D65F85E2C39BE9DF5131D924E9" encryption_subkey_keyid = "6A112FD3390B2E53AFC2E57F8FC8E12099AECEEA" unsupported_subkey_keyid = "611A9B648E16F54E8A7FAD5DA51E8CDF3B06524F" expired_key_keyid = "E8AC80C924116DABB51D4B987CB07D6D2C199C7C" keyid_768C43 = "7B3ABB26B97B655AB9296BD15B0BD02E1C768C43" @classmethod def setUpClass(self): # Create directory to run the tests without having everything blow up self.working_dir = os.getcwd() # Find demo files gpg_keyring_path = os.path.join( os.path.dirname(os.path.realpath(__file__)), "gpg_keyrings", "rsa") self.test_dir = os.path.realpath(tempfile.mkdtemp()) self.gnupg_home = os.path.join(self.test_dir, "rsa") shutil.copytree(gpg_keyring_path, self.gnupg_home) os.chdir(self.test_dir) @classmethod def tearDownClass(self): """Change back to initial working dir and remove temp test directory. """ os.chdir(self.working_dir) shutil.rmtree(self.test_dir) def test_export_pubkey_error(self): """Test correct error is raised if function called incorrectly. """ with self.assertRaises(ValueError): export_pubkey("not-a-key-id") def test_export_pubkey(self): """ export a public key and make sure the parameters are the right ones: since there's very little we can do to check rsa key parameters are right we pre-exported the public key to an ssh key, which we can load with cryptography for the sake of comparison """ # export our gpg key, using our functions key_data = export_pubkey(self.default_keyid, homedir=self.gnupg_home) our_exported_key = rsa_create_pubkey(key_data) # load the equivalent ssh key, and make sure that we get the same RSA key # parameters ssh_key_basename = "{}.ssh".format(self.default_keyid) ssh_key_path = os.path.join(self.gnupg_home, ssh_key_basename) with open(ssh_key_path, "rb") as fp: keydata = fp.read() ssh_key = serialization.load_ssh_public_key(keydata, backends.default_backend()) self.assertEqual(ssh_key.public_numbers().n, our_exported_key.public_numbers().n) self.assertEqual(ssh_key.public_numbers().e, our_exported_key.public_numbers().e) subkey_keyids = list(key_data["subkeys"].keys()) # We export the whole master key bundle which must contain the subkeys self.assertTrue(self.signing_subkey_keyid.lower() in subkey_keyids) # Currently we do not exclude encryption subkeys self.assertTrue(self.encryption_subkey_keyid.lower() in subkey_keyids) # However we do exclude subkeys, whose algorithm we do not support self.assertFalse(self.unsupported_subkey_keyid.lower() in subkey_keyids) # When passing the subkey keyid we also export the whole keybundle key_data2 = export_pubkey(self.signing_subkey_keyid, homedir=self.gnupg_home) self.assertDictEqual(key_data, key_data2) def test_export_pubkeys(self): """Test export multiple pubkeys at once. """ key_dict = export_pubkeys([self.default_keyid, self.keyid_768C43], homedir=self.gnupg_home) ANY_PUBKEY_DICT_SCHEMA.check_match(key_dict) self.assertListEqual( sorted([self.default_keyid.lower(), self.keyid_768C43.lower()]), sorted(key_dict.keys())) def test_gpg_sign_and_verify_object_with_default_key(self): """Create a signature using the default key on the keyring """ test_data = b'test_data' wrong_data = b'something malicious' signature = create_signature(test_data, homedir=self.gnupg_home) key_data = export_pubkey(self.default_keyid, homedir=self.gnupg_home) self.assertTrue(verify_signature(signature, key_data, test_data)) self.assertFalse(verify_signature(signature, key_data, wrong_data)) def test_gpg_sign_and_verify_object(self): """Create a signature using a specific key on the keyring """ test_data = b'test_data' wrong_data = b'something malicious' signature = create_signature(test_data, keyid=self.default_keyid, homedir=self.gnupg_home) key_data = export_pubkey(self.default_keyid, homedir=self.gnupg_home) self.assertTrue(verify_signature(signature, key_data, test_data)) self.assertFalse(verify_signature(signature, key_data, wrong_data)) def test_gpg_sign_and_verify_object_default_keyring(self): """Sign/verify using keyring from envvar. """ test_data = b'test_data' gnupg_home_backup = os.environ.get("GNUPGHOME") os.environ["GNUPGHOME"] = self.gnupg_home signature = create_signature(test_data, keyid=self.default_keyid) key_data = export_pubkey(self.default_keyid) self.assertTrue(verify_signature(signature, key_data, test_data)) # Reset GNUPGHOME if gnupg_home_backup: os.environ["GNUPGHOME"] = gnupg_home_backup else: del os.environ["GNUPGHOME"] def test_create_signature_with_expired_key(self): """Test signing with expired key raises gpg CommandError. """ with self.assertRaises(CommandError) as ctx: create_signature(b"livestock", keyid=self.expired_key_keyid, homedir=self.gnupg_home) expected = "returned non-zero exit status '2'" self.assertTrue(expected in str(ctx.exception), "{} not in {}".format( expected, ctx.exception)) def test_verify_signature_with_expired_key(self): """Test sig verification with expired key raises KeyExpirationError. """ signature = { "keyid": self.expired_key_keyid, "other_headers": "deadbeef", "signature": "deadbeef", } content = b"livestock" key = export_pubkey(self.expired_key_keyid, homedir=self.gnupg_home) with self.assertRaises(KeyExpirationError) as ctx: verify_signature(signature, key, content) expected = ("GPG key 'e8ac80c924116dabb51d4b987cb07d6d2c199c7c' " "created on '2019-03-25 12:46 UTC' with validity period '1 day, " "0:25:01' expired on '2019-03-26 13:11 UTC'.") self.assertTrue(expected == str(ctx.exception), "\nexpected: {}" "\ngot: {}".format(expected, ctx.exception)) @unittest.skipIf(not HAVE_GPG, "gpg not found") class TestGPGDSA(unittest.TestCase): """ Test signature creation, verification and key export from the gpg module """ default_keyid = "C242A830DAAF1C2BEF604A9EF033A3A3E267B3B1" @classmethod def setUpClass(self): # Create directory to run the tests without having everything blow up self.working_dir = os.getcwd() self.test_dir = os.path.realpath(tempfile.mkdtemp()) self.gnupg_home = os.path.join(self.test_dir, "dsa") # Find keyrings keyrings = os.path.join( os.path.dirname(os.path.realpath(__file__)), "gpg_keyrings", "dsa") shutil.copytree(keyrings, self.gnupg_home) os.chdir(self.test_dir) @classmethod def tearDownClass(self): """Change back to initial working dir and remove temp test directory. """ os.chdir(self.working_dir) shutil.rmtree(self.test_dir) def test_export_pubkey(self): """ export a public key and make sure the parameters are the right ones: since there's very little we can do to check key parameters are right we pre-exported the public key to an x.509 SubjectPublicKeyInfo key, which we can load with cryptography for the sake of comparison """ # export our gpg key, using our functions key_data = export_pubkey(self.default_keyid, homedir=self.gnupg_home) our_exported_key = dsa_create_pubkey(key_data) # load same key, pre-exported with 3rd-party tooling pem_key_basename = "{}.pem".format(self.default_keyid) pem_key_path = os.path.join(self.gnupg_home, pem_key_basename) with open(pem_key_path, "rb") as fp: keydata = fp.read() pem_key = serialization.load_pem_public_key(keydata, backends.default_backend()) # make sure keys match self.assertEqual(pem_key.public_numbers().y, our_exported_key.public_numbers().y) self.assertEqual(pem_key.public_numbers().parameter_numbers.g, our_exported_key.public_numbers().parameter_numbers.g) self.assertEqual(pem_key.public_numbers().parameter_numbers.q, our_exported_key.public_numbers().parameter_numbers.q) self.assertEqual(pem_key.public_numbers().parameter_numbers.p, our_exported_key.public_numbers().parameter_numbers.p) def test_gpg_sign_and_verify_object_with_default_key(self): """Create a signature using the default key on the keyring """ test_data = b'test_data' wrong_data = b'something malicious' signature = create_signature(test_data, homedir=self.gnupg_home) key_data = export_pubkey(self.default_keyid, homedir=self.gnupg_home) self.assertTrue(verify_signature(signature, key_data, test_data)) self.assertFalse(verify_signature(signature, key_data, wrong_data)) def test_gpg_sign_and_verify_object(self): """Create a signature using a specific key on the keyring """ test_data = b'test_data' wrong_data = b'something malicious' signature = create_signature(test_data, keyid=self.default_keyid, homedir=self.gnupg_home) key_data = export_pubkey(self.default_keyid, homedir=self.gnupg_home) self.assertTrue(verify_signature(signature, key_data, test_data)) self.assertFalse(verify_signature(signature, key_data, wrong_data)) @unittest.skipIf(not HAVE_GPG, "gpg not found") class TestGPGEdDSA(unittest.TestCase): """ Test signature creation, verification and key export from the gpg module """ default_keyid = "4E630F84838BF6F7447B830B22692F5FEA9E2DD2" @classmethod def setUpClass(self): # Create directory to run the tests without having everything blow up self.working_dir = os.getcwd() self.test_dir = os.path.realpath(tempfile.mkdtemp()) self.gnupg_home = os.path.join(self.test_dir, "dsa") # Find keyrings keyrings = os.path.join( os.path.dirname(os.path.realpath(__file__)), "gpg_keyrings", "eddsa") shutil.copytree(keyrings, self.gnupg_home) os.chdir(self.test_dir) @classmethod def tearDownClass(self): """Change back to initial working dir and remove temp test directory. """ os.chdir(self.working_dir) shutil.rmtree(self.test_dir) def test_gpg_sign_and_verify_object_with_default_key(self): """Create a signature using the default key on the keyring """ test_data = b'test_data' wrong_data = b'something malicious' signature = create_signature(test_data, homedir=self.gnupg_home) key_data = export_pubkey(self.default_keyid, homedir=self.gnupg_home) self.assertTrue(verify_signature(signature, key_data, test_data)) self.assertFalse(verify_signature(signature, key_data, wrong_data)) def test_gpg_sign_and_verify_object_with_specific_key(self): """Create a signature using a specific key on the keyring """ test_data = b'test_data' wrong_data = b'something malicious' signature = create_signature(test_data, keyid=self.default_keyid, homedir=self.gnupg_home) key_data = export_pubkey(self.default_keyid, homedir=self.gnupg_home) self.assertTrue(verify_signature(signature, key_data, test_data)) self.assertFalse(verify_signature(signature, key_data, wrong_data)) def test_verify_short_signature(self): """Correctly verify a special-crafted short signature. """ test_data = b"hello" signature_path = os.path.join(self.gnupg_home, "short.sig") # Read special-crafted raw gpg signature that is one byte too short with open(signature_path, "rb") as f: signature_data = f.read() # Check that the signature is padded upon parsing # NOTE: The returned signature is a hex string and thus twice as long signature = parse_signature_packet(signature_data) self.assertTrue(len(signature["signature"]) == (ED25519_SIG_LENGTH * 2)) # Check that the signature can be successfully verified key = export_pubkey(self.default_keyid, homedir=self.gnupg_home) self.assertTrue(verify_signature(signature, key, test_data)) if __name__ == "__main__": unittest.main()