Permalink
Switch branches/tags
Nothing to show
Find file Copy path
Fetching contributors…
Cannot retrieve contributors at this time
398 lines (368 sloc) 17.1 KB
# -*- coding: utf-8 -*-
"""
A test harness for gnupg.py.
Copyright (C) 2008-2012 Vinay Sajip. All rights reserved.
"""
import doctest
import logging
import os.path
import os
import shutil
import sys
import tempfile
import unittest
import gnupg
__author__ = "Vinay Sajip"
__date__ = "$01-Mar-2010 12:30:04$"
ALL_TESTS = True
logger = logging.getLogger(__name__)
KEYS_TO_IMPORT = """-----BEGIN PGP PUBLIC KEY BLOCK-----
Version: GnuPG v1.4.9 (MingW32)
mQGiBEiH4QERBACm48JJsg2XGzWfL7f/fjp3wtrY+JIz6P07s7smr35kve+wl605
nqHtgjnIVpUVsbI9+xhIAPIkFIR6ZcQ7gRDhoT0bWKGkfdQ7YzXedVRPlQLdbpmR
K2pKKySpF35pJsPAYa73EVaxu2KrII4CyBxVQgNWfGwEbtL5FfzuHhVOZwCg6JF7
bgOMPmEwBLEHLmgiXbb5K48D/2xsXtWMkvgRp/ubcLxzbNjaHH6gSb2IfDi1+W/o
Bmfua6FksPnEDn7PWnBhCEO9rf1tV0FcrvkR9m2FGfx38tjssxDdLvX511gbfc/Q
DJxZ00A63BxI3xav8RiXlqpfQGXpLJmCLdeCh5DXOsVMCfepqRbWyJF0St7LDcq9
SmuXA/47dzb8puo9dNxA5Nj48I5g4ke3dg6nPn7aiBUQ35PfXjIktXB6/sQJtWWx
XNFX/GVUxqMM0/aCMPdtaoDkFtz1C6b80ngEz94vXzmON7PCgDY6LqZP1B1xbrkr
4jGSr68iq7ERT+7E/iF9xp+Ynl91KK7h8llY6zFw+yIe6vGlcLQvR2FyeSBHcm9z
cyAoQSB0ZXN0IHVzZXIpIDxnYXJ5Lmdyb3NzQGdhbW1hLmNvbT6IYAQTEQIAIAUC
SIfhAQIbAwYLCQgHAwIEFQIIAwQWAgMBAh4BAheAAAoJEJZ2Ekdc7S4UtEcAoJIA
iZurfuzIUE9Dtn86o6vC14qoAJ9P79mxR88wRr/ac9h5/BIf5cZKMbkCDQRIh+EB
EAgAyYCvtS43J/OfuGHPGPZT0q8C+Y15YLItSQ3H6IMZWFY+sX+ZocaIiM4noVRG
+mrEqzO9JNh4KP1OdFju1ZC8HZXpPVur48XlTNSm0yjmvvfmi+aGSuyQ0NkfLyi1
aBeRvB4na/oFUgl908l7vpSYWYn4EY3xpvwJdyTWHTh4o7+zvrR1fByDt49k2b3z
yTACoxYPVQfknt8gxqLqHZsbgn02Ml7HS17bSWr5Z7PlWqDlmsdqUikVU9d2RvIq
R+YIJbOdHSklbVQQDhr+xgHPi39e7nXMxR/rMjMbz7E5vSNkge45n8Pzim8iyqy+
MTMW8psV/OyrHUJzBEA7M6hA1wADBwgAnB0HzI1iyiQmIymO0Hj0BgqU6/avFw9R
ggBuE2v7KsvuLP6ohXDEhYopjw5hgeotobpg6tS15ynch+6L8uWsJ0rcY2X9dsJy
O8/5mjrNDHwCKiYRuZfmRZjzW03vO/9+rjtZ0NzoWYMP3UR8lUTVp2LTygefBA88
Zgw6dWBVzn+/c0vdwcF4Y3njYKE7eq4VrfcwqRgD0hDyIJd1OpqzHfXXnTtLlAsm
UwtdONzlwu7KkgafMo4vzKY6dCtUkR6pXAE/rLQfCTonwl9SnyusoYZgjDoj4Pvw
ePxIl2q05dcn96NJGS+SfS/5B4H4irbfaEYmCfKps+45sjncYGhZ/ohJBBgRAgAJ
BQJIh+EBAhsMAAoJEJZ2Ekdc7S4U2lkAoIwZLMHVldC0v9wse53xU0NsNIskAKDc
Ft0XWUJ9yajOEUqCVHNs3F99t5kBogRIh+FVEQQAhk/ROtJ5/O+YERl4tZZBEhGH
JendDBDfzmfRO9GIDcZI20nx5KJ1M/zGguqgKiVRlBy32NS/IRqwSI158npWYLfJ
rYCWrC2duMK2i/8prOEfaktnqZXVCHudGtP4mTqNSs+867LnGhQ4w3HmB09zCIpD
eIhhhPOb5H19H8UlojsAoLwsq5BACqUKoiz8lUufpTTFMbaDA/4v1fWmprYAxGq9
cZ9svae772ymN/RRPDb/D+UJoJCCJSjE8m4MukVchyJVT8GmpJM2+dlt62eYwtz8
bGNt+Yzzxr0N8rLutsSks7RaM16MaqiAlM20gAXEovxBiocgP/p5bO3FGKOBbrfd
h47BZDEqLvfJefXjZEsElbZ9oL2zDgP9EsoDS9mbfesHDsagE5jCZRTY1C/FRLBO
zhGgP2IlqBdOX8BYBYZiIlLM+pN5fU0Hcu3VOZY1Hnj6r3VbK1bOScQzqrZ7qgmw
TRgyxUQalaOhMb5rUD0+dUFxa/mhTerx5POrX6zOWmmK0ldYTZO4/+nWr4FwmU8R
41nYYYdi0yS0MURhbm55IERhdmlzIChBIHRlc3QgdXNlcikgPGRhbm55LmRhdmlz
QGRlbHRhLmNvbT6IYAQTEQIAIAUCSIfhVQIbAwYLCQgHAwIEFQIIAwQWAgMBAh4B
AheAAAoJEG7bKmS7rMYAEt8An2jxsmsE1MZVZc4Ev8RB9Gu1zbsCAJ9G5kkYIIf0
OoDqCjkDMDJcpd4MqLkCDQRIh+FVEAgAgHQ+EyseLw6A3BS2EUz6U1ZGzuJ5CXxY
BY8xaQtE+9AJ0WHyzKeptnlnY1x9et3ny1BcVC5aR1OgsDiuVRvSFwpFfVxMKbRT
kvERWADfB0N5EyWwyE0E4BT5hyEhW7fS0bucJL6UK5PKvfE5wexWlUI3yV4K1z6W
2gSNL60o3kmoGn9K5ICWO/jbi6MkPptSoDu/laCJHv/aid6Gf94ckDClQQyLsccj
0ibynm6rI3cIzpPMbimKIsKT1smAqZEBsTucBlOjIuIROANTZUN3reGIRh/kVNyg
YTrkUnIqVS9FnbHa2wxeb6F/cO33fPiVfiCmZuKI1Uh4PMGaaSCh0wADBQf/SaXN
WcuD0mrEnxqgEJRx67ZeFZjZM53Obu3JYQ++lqsthf8MxE7K4J/67xDpOh6waK0G
6GCLwEm3Z7wjCaz1DYg2uJp/3pispWxZio3PLVe7WrMY+oEBHEsiJXicS5dV620a
uoaBnnc0aQWT/DREE5s35IrZCh4WDQgO9rl0i/qcIITm77TmQbq2Xdj5vt6s0cx7
oHKRaFBpQ8DBsCQ+D8Xz7i1oUygNp4Z5xPhItWeCfE9YoCoem4jSB4HGwmMOEicp
VSpY43k01cd0Yfb1OMhA5C8OBwcwn3zvQB7nbxyxyQ9qphfwhMookIL4+tKKBIQL
CnOGhApkAGbjRwuLi4hJBBgRAgAJBQJIh+FVAhsMAAoJEG7bKmS7rMYA+JQAn0E2
WdPQjKEfKnr+bW4yubwMUYKyAJ4uiE8Rv/oEED1oM3xeJqa+MJ9V1w==
=sqld
-----END PGP PUBLIC KEY BLOCK-----"""
def is_list_with_len(o, n):
return isinstance(o, list) and len(o) == n
def compare_keys(k1, k2):
"Compare ASCII keys"
k1 = k1.split('\n')
k2 = k2.split('\n')
del k1[1] # remove version lines
del k2[1]
return k1 != k2
class GPGTestCase(unittest.TestCase):
def setUp(self):
hd = os.path.join(os.getcwd(), 'keys')
if os.path.exists(hd):
self.assertTrue(os.path.isdir(hd),
"Not a directory: %s" % hd)
shutil.rmtree(hd)
self.homedir = hd
self.gpg = gnupg.GPG(gnupghome=hd, gpgbinary='gpg')
def test_environment(self):
"Test the environment by ensuring that setup worked"
hd = self.homedir
self.assertTrue(os.path.exists(hd) and os.path.isdir(hd),
"Not an existing directory: %s" % hd)
def test_list_keys_initial(self):
"Test that initially there are no keys"
logger.debug("test_list_keys_initial begins")
public_keys = self.gpg.list_keys()
self.assertTrue(is_list_with_len(public_keys, 0),
"Empty list expected")
private_keys = self.gpg.list_keys(secret=True)
self.assertTrue(is_list_with_len(private_keys, 0),
"Empty list expected")
logger.debug("test_list_keys_initial ends")
def generate_key(self, first_name, last_name, domain, passphrase=None):
"Generate a key"
params = {
'Key-Type': 'DSA',
'Key-Length': 1024,
'Subkey-Type': 'ELG-E',
'Subkey-Length': 2048,
'Name-Comment': 'A test user',
'Expire-Date': 0,
}
params['Name-Real'] = '%s %s' % (first_name, last_name)
params['Name-Email'] = ("%s.%s@%s" % (first_name, last_name, domain)).lower()
if passphrase is None:
passphrase = ("%s%s" % (first_name[0], last_name)).lower()
params['Passphrase'] = passphrase
cmd = self.gpg.gen_key_input(**params)
return self.gpg.gen_key(cmd)
def do_key_generation(self):
"Test that key generation succeeds"
result = self.generate_key("Barbara", "Brown", "beta.com")
self.assertNotEqual(None, result, "Non-null result")
return result
def test_list_keys_after_generation(self):
"Test that after key generation, the generated key is available"
self.test_list_keys_initial()
self.do_key_generation()
public_keys = self.gpg.list_keys()
self.assertTrue(is_list_with_len(public_keys, 1),
"1-element list expected")
private_keys = self.gpg.list_keys(secret=True)
self.assertTrue(is_list_with_len(private_keys, 1),
"1-element list expected")
def test_encryption_and_decryption(self):
"Test that encryption and decryption works"
logger.debug("test_encryption_and_decryption begins")
key = self.generate_key("Andrew", "Able", "alpha.com",
passphrase="andy")
andrew = key.fingerprint
key = self.generate_key("Barbara", "Brown", "beta.com")
barbara = key.fingerprint
gpg = self.gpg
if gnupg._py3k:
data = 'Hello, André!'
else:
data = unicode('Hello, André', gpg.encoding)
data = data.encode(gpg.encoding)
edata = str(gpg.encrypt(data, barbara))
self.assertNotEqual(data, edata, "Data must have changed")
ddata = gpg.decrypt(edata, passphrase="bbrown")
if data != ddata.data:
logger.debug("was: %r", data)
logger.debug("new: %r", ddata.data)
self.assertEqual(data, ddata.data, "Round-trip must work")
edata = str(gpg.encrypt(data, [andrew, barbara]))
self.assertNotEqual(data, edata, "Data must have changed")
ddata = gpg.decrypt(edata, passphrase="andy")
self.assertEqual(data, ddata.data, "Round-trip must work")
ddata = gpg.decrypt(edata, passphrase="bbrown")
self.assertEqual(data, ddata.data, "Round-trip must work")
logger.debug("test_encryption_and_decryption ends")
# Test symmetric encryption
data = "chippy was here"
edata = str(gpg.encrypt(data, None, passphrase='bbrown', symmetric=True))
ddata = gpg.decrypt(edata, passphrase='bbrown')
self.assertEqual(data, str(ddata))
def test_import_and_export(self):
"Test that key import and export works"
logger.debug("test_import_and_export begins")
self.test_list_keys_initial()
gpg = self.gpg
gpg.import_keys(KEYS_TO_IMPORT)
public_keys = gpg.list_keys()
self.assertTrue(is_list_with_len(public_keys, 2),
"2-element list expected")
private_keys = gpg.list_keys(secret=True)
self.assertTrue(is_list_with_len(private_keys, 0),
"Empty list expected")
ascii = gpg.export_keys([k['keyid'] for k in public_keys])
self.assertTrue(ascii.find("PGP PUBLIC KEY BLOCK") >= 0,
"Exported key should be public")
ascii = ascii.replace("\r", "").strip()
match = compare_keys(ascii, KEYS_TO_IMPORT)
if match:
logger.debug("was: %r", KEYS_TO_IMPORT)
logger.debug("now: %r", ascii)
self.assertEqual(0, match, "Keys must match")
#Generate a key so we can test exporting private keys
key = self.do_key_generation()
ascii = gpg.export_keys(key.fingerprint, True)
self.assertTrue(ascii.find("PGP PRIVATE KEY BLOCK") >= 0,
"Exported key should be private")
logger.debug("test_import_and_export ends")
def test_import_only(self):
"Test that key import works"
logger.debug("test_import_only begins")
self.test_list_keys_initial()
self.gpg.import_keys(KEYS_TO_IMPORT)
public_keys = self.gpg.list_keys()
self.assertTrue(is_list_with_len(public_keys, 2),
"2-element list expected")
private_keys = self.gpg.list_keys(secret=True)
self.assertTrue(is_list_with_len(private_keys, 0),
"Empty list expected")
ascii = self.gpg.export_keys([k['keyid'] for k in public_keys])
self.assertTrue(ascii.find("PGP PUBLIC KEY BLOCK") >= 0,
"Exported key should be public")
ascii = ascii.replace("\r", "").strip()
match = compare_keys(ascii, KEYS_TO_IMPORT)
if match:
logger.debug("was: %r", KEYS_TO_IMPORT)
logger.debug("now: %r", ascii)
self.assertEqual(0, match, "Keys must match")
logger.debug("test_import_only ends")
def test_signature_verification(self):
"Test that signing and verification works"
logger.debug("test_signature_verification begins")
key = self.generate_key("Andrew", "Able", "alpha.com")
if gnupg._py3k:
data = 'Hello, André!'
else:
data = unicode('Hello, André', self.gpg.encoding)
data = data.encode(self.gpg.encoding)
sig = self.gpg.sign(data, keyid=key.fingerprint, passphrase='bbrown')
self.assertFalse(sig, "Bad passphrase should fail")
sig = self.gpg.sign(data, keyid=key.fingerprint, passphrase='aable')
self.assertTrue(sig, "Good passphrase should succeed")
verified = self.gpg.verify(sig.data)
if key.fingerprint != verified.fingerprint:
logger.debug("key: %r", key.fingerprint)
logger.debug("ver: %r", verified.fingerprint)
self.assertEqual(key.fingerprint, verified.fingerprint,
"Fingerprints must match")
if not os.path.exists('random_binary_data'):
data_file = open('random_binary_data', 'wb')
data_file.write(os.urandom(5120 * 1024))
data_file.close()
data_file = open('random_binary_data', 'rb')
sig = self.gpg.sign_file(data_file, keyid=key.fingerprint,
passphrase='aable')
data_file.close()
self.assertTrue(sig, "File signing should succeed")
try:
file = gnupg._make_binary_stream(sig.data, self.gpg.encoding)
verified = self.gpg.verify_file(file)
except UnicodeDecodeError: #happens in Python 2.6
from io import BytesIO
verified = self.gpg.verify_file(BytesIO(sig.data))
if key.fingerprint != verified.fingerprint:
logger.debug("key: %r", key.fingerprint)
logger.debug("ver: %r", verified.fingerprint)
self.assertEqual(key.fingerprint, verified.fingerprint,
"Fingerprints must match")
data_file = open('random_binary_data', 'rb')
sig = self.gpg.sign_file(data_file, keyid=key.fingerprint,
passphrase='aable', detach=True)
data_file.close()
self.assertTrue(sig, "File signing should succeed")
try:
file = gnupg._make_binary_stream(sig.data, self.gpg.encoding)
verified = self.gpg.verify_file(file, 'random_binary_data')
except UnicodeDecodeError: #happens in Python 2.6
from io import BytesIO
verified = self.gpg.verify_file(BytesIO(sig.data))
if key.fingerprint != verified.fingerprint:
logger.debug("key: %r", key.fingerprint)
logger.debug("ver: %r", verified.fingerprint)
self.assertEqual(key.fingerprint, verified.fingerprint,
"Fingerprints must match")
logger.debug("test_signature_verification ends")
def test_deletion(self):
"Test that key deletion works"
logger.debug("test_deletion begins")
self.gpg.import_keys(KEYS_TO_IMPORT)
public_keys = self.gpg.list_keys()
self.assertTrue(is_list_with_len(public_keys, 2),
"2-element list expected")
self.gpg.delete_keys(public_keys[0]['fingerprint'])
public_keys = self.gpg.list_keys()
self.assertTrue(is_list_with_len(public_keys, 1),
"1-element list expected")
logger.debug("test_deletion ends")
def test_nogpg(self):
"Test that absence of gpg is handled correctly"
self.assertRaises(ValueError, gnupg.GPG, gnupghome=self.homedir,
gpgbinary='frob')
def test_file_encryption_and_decryption(self):
"Test that encryption/decryption to/from file works"
logger.debug("test_file_encryption_and_decryption begins")
encfno, encfname = tempfile.mkstemp()
decfno, decfname = tempfile.mkstemp()
# On Windows, if the handles aren't closed, the files can't be deleted
os.close(encfno)
os.close(decfno)
logger.debug('Encrypting to: %r', encfname)
logger.debug('Decrypting to: %r', decfname)
try:
key = self.generate_key("Andrew", "Able", "alpha.com",
passphrase="andy")
andrew = key.fingerprint
key = self.generate_key("Barbara", "Brown", "beta.com")
barbara = key.fingerprint
data = "Hello, world!"
file = gnupg._make_binary_stream(data, self.gpg.encoding)
edata = self.gpg.encrypt_file(file,
barbara,
armor=False, output=encfname)
efile = open(encfname, 'rb')
ddata = self.gpg.decrypt_file(efile, passphrase="bbrown",
output=decfname)
efile.seek(0, 0) # can't use os.SEEK_SET in 2.4
edata = efile.read()
efile.close()
dfile = open(decfname, 'rb')
ddata = dfile.read()
dfile.close()
data = data.encode(self.gpg.encoding)
if ddata != data:
logger.debug("was: %r", data)
logger.debug("new: %r", ddata)
self.assertEqual(data, ddata, "Round-trip must work")
finally:
for fn in (encfname, decfname):
if os.path.exists(fn):
os.remove(fn)
logger.debug("test_file_encryption_and_decryption ends")
TEST_GROUPS = {
'sign' : set(['test_signature_verification']),
'crypt' : set(['test_encryption_and_decryption',
'test_file_encryption_and_decryption']),
'key' : set(['test_deletion', 'test_import_and_export',
'test_list_keys_after_generation']),
'import' : set(['test_import_only']),
'basic' : set(['test_environment', 'test_list_keys_initial',
'test_nogpg']),
}
def suite(args=None):
if args is None:
args = sys.argv[1:]
if not args:
result = unittest.TestLoader().loadTestsFromTestCase(GPGTestCase)
want_doctests = True
else:
tests = set()
want_doctests = False
for arg in args:
if arg in TEST_GROUPS:
tests.update(TEST_GROUPS[arg])
elif arg == "doc":
want_doctests = True
else:
print("Ignoring unknown test group %r" % arg)
result = unittest.TestSuite(list(map(GPGTestCase, tests)))
if want_doctests:
result.addTest(doctest.DocTestSuite(gnupg))
return result
def init_logging():
logging.basicConfig(level=logging.DEBUG, filename="test_gnupg.log",
filemode="w", format="%(asctime)s %(levelname)-5s %(name)-10s %(threadName)-10s %(message)s")
def main():
init_logging()
tests = suite()
unittest.TextTestRunner(verbosity=2).run(tests)
if __name__ == "__main__":
main()