Merge branch 'release/2.0.2'
commit
2beac24161
|
@ -0,0 +1,33 @@
|
|||
[run]
|
||||
source =
|
||||
gnupg
|
||||
branch = True
|
||||
#parallel = True
|
||||
timid = True
|
||||
|
||||
[report]
|
||||
modules = gnupg
|
||||
omit =
|
||||
*/test*
|
||||
*/_version*
|
||||
*/__init__*
|
||||
*/copyleft*
|
||||
*/sitecustomize*
|
||||
# Regexes for lines to exclude from report generation:
|
||||
exclude_lines =
|
||||
pragma: no cover
|
||||
# don't complain if the code doesn't hit unimplemented sections:
|
||||
raise NotImplementedError
|
||||
pass
|
||||
# don't complain if non-runnable or debuging code isn't run:
|
||||
if 0:
|
||||
if False:
|
||||
def __repr__
|
||||
if __name__ == .__main__.:
|
||||
# Ignore source code which cannot be found:
|
||||
ignore_errors = True
|
||||
# Exit with status code 2 if under this percentage is covered:
|
||||
fail_under = 10
|
||||
|
||||
[html]
|
||||
directory = docs/coverage-html
|
|
@ -170,7 +170,7 @@ def displayNewKey(key):
|
|||
# `result` is a `gnupg._parsers.ListKeys`, which is list-like, so iterate
|
||||
# over all the keys and display their info:
|
||||
for gpgkey in keylist:
|
||||
for k, v in gpgkey:
|
||||
for k, v in gpgkey.items():
|
||||
log.info("%s: %s" % (k.capitalize(), v))
|
||||
|
||||
return keylist
|
||||
|
|
|
@ -46,6 +46,8 @@ except ImportError:
|
|||
|
||||
from . import _parsers
|
||||
from . import _util
|
||||
from ._util import b
|
||||
from ._util import s
|
||||
|
||||
from ._parsers import _check_preferences
|
||||
from ._parsers import _sanitise_list
|
||||
|
@ -804,6 +806,19 @@ class GPGBase(object):
|
|||
## We could use _handle_io here except for the fact that if the
|
||||
## passphrase is bad, gpg bails and you can't write the message.
|
||||
result = self._result_map['sign'](self)
|
||||
|
||||
## If the passphrase is an empty string, the message up to and
|
||||
## including its first newline will be cut off before making it to the
|
||||
## GnuPG process. Therefore, if the passphrase='' or passphrase=b'',
|
||||
## we set passphrase=None. See Issue #82:
|
||||
## https://github.com/isislovecruft/python-gnupg/issues/82
|
||||
if _util._is_string(passphrase):
|
||||
passphrase = passphrase if len(passphrase) > 0 else None
|
||||
elif _util._is_bytes(passphrase):
|
||||
passphrase = s(passphrase) if len(passphrase) > 0 else None
|
||||
else:
|
||||
passphrase = None
|
||||
|
||||
proc = self._open_subprocess(args, passphrase is not None)
|
||||
try:
|
||||
if passphrase:
|
||||
|
@ -1001,7 +1016,10 @@ class GPGBase(object):
|
|||
result = self._result_map['crypt'](self)
|
||||
log.debug("Got data '%s' with type '%s'." % (data, type(data)))
|
||||
self._handle_io(args, data, result, passphrase=passphrase, binary=True)
|
||||
log.debug("\n%s" % result.data)
|
||||
# Avoid writing raw encrypted bytes to terminal loggers and breaking
|
||||
# them in that adorable way where they spew hieroglyphics until reset:
|
||||
if armor:
|
||||
log.debug("\n%s" % result.data)
|
||||
|
||||
if output_filename:
|
||||
log.info("Writing encrypted output to file: %s" % output_filename)
|
||||
|
|
119
gnupg/_util.py
119
gnupg/_util.py
|
@ -52,7 +52,7 @@ else:
|
|||
|
||||
# The remaining StringIO classes which are imported are used to determine if a
|
||||
# object is a stream-like in :func:`_is_stream`.
|
||||
if sys.version_info.major == 2:
|
||||
if 2 == sys.version_info[0]:
|
||||
# Import the StringIO class from the StringIO module since it is a
|
||||
# commonly used stream class. It is distinct from either of the
|
||||
# StringIO's that may be loaded in the above try/except clause, so the
|
||||
|
@ -165,6 +165,51 @@ def find_encodings(enc=None, system=False):
|
|||
|
||||
return coder
|
||||
|
||||
|
||||
if _py3k:
|
||||
def b(x):
|
||||
"""See http://python3porting.com/problems.html#nicer-solutions"""
|
||||
coder = find_encodings()
|
||||
if isinstance(x, bytes):
|
||||
return coder.encode(x.decode(coder.name))[0]
|
||||
else:
|
||||
return coder.encode(x)[0]
|
||||
|
||||
def s(x):
|
||||
if isinstance(x, str):
|
||||
return x
|
||||
elif isinstance(x, (bytes, bytearray)):
|
||||
return x.decode(find_encodings().name)
|
||||
else:
|
||||
raise NotImplemented
|
||||
else:
|
||||
def b(x):
|
||||
"""See http://python3porting.com/problems.html#nicer-solutions"""
|
||||
return x
|
||||
|
||||
def s(x):
|
||||
if isinstance(x, basestring):
|
||||
return x
|
||||
elif isinstance(x, (bytes, bytearray)):
|
||||
return x.decode(find_encodings().name)
|
||||
else:
|
||||
raise NotImplemented
|
||||
|
||||
def binary(data):
|
||||
coder = find_encodings()
|
||||
|
||||
if _py3k and isinstance(data, bytes):
|
||||
encoded = coder.encode(data.decode(coder.name))[0]
|
||||
elif _py3k and isinstance(data, str):
|
||||
encoded = coder.encode(data)[0]
|
||||
elif not _py3k and type(data) is not str:
|
||||
encoded = coder.encode(data)[0]
|
||||
else:
|
||||
encoded = data
|
||||
|
||||
return encoded
|
||||
|
||||
|
||||
def author_info(name, contact=None, public_key=None):
|
||||
"""Easy object-oriented representation of contributor info.
|
||||
|
||||
|
@ -183,7 +228,6 @@ def _copy_data(instream, outstream):
|
|||
:param file outstream: The file descriptor of a tmpfile to write to.
|
||||
"""
|
||||
sent = 0
|
||||
coder = find_encodings()
|
||||
|
||||
while True:
|
||||
if ((_py3k and isinstance(instream, str)) or
|
||||
|
@ -196,18 +240,9 @@ def _copy_data(instream, outstream):
|
|||
break
|
||||
|
||||
sent += len(data)
|
||||
log.debug("Sending chunk %d bytes:\n%s" % (sent, data))
|
||||
|
||||
if _py3k and isinstance(data, bytes):
|
||||
encoded = coder.encode(data.decode(coder.name))[0]
|
||||
elif _py3k and isinstance(data, str):
|
||||
encoded = coder.encode(data)[0]
|
||||
elif not _py3k and type(data) is not str:
|
||||
encoded = coder.encode(data)[0]
|
||||
else:
|
||||
encoded = data
|
||||
log.debug("Writing encoded data with type %s to outstream... "
|
||||
% type(encoded))
|
||||
encoded = binary(data)
|
||||
log.debug("Sending %d bytes of data..." % sent)
|
||||
log.debug("Encoded data (type %s):\n%s" % (type(encoded), encoded))
|
||||
|
||||
if not _py3k:
|
||||
try:
|
||||
|
@ -440,6 +475,31 @@ def _is_stream(input):
|
|||
"""
|
||||
return isinstance(input, tuple(_STREAMLIKE_TYPES))
|
||||
|
||||
def _is_string(thing):
|
||||
"""Check that **thing** is a string. The definition of the latter depends
|
||||
upon the Python version.
|
||||
|
||||
:param thing: The thing to check if it's a string.
|
||||
:rtype: bool
|
||||
:returns: ``True`` if **thing** is string (or unicode in Python2).
|
||||
"""
|
||||
if (_py3k and isinstance(thing, str)):
|
||||
return True
|
||||
if (not _py3k and isinstance(thing, basestring)):
|
||||
return True
|
||||
return False
|
||||
|
||||
def _is_bytes(thing):
|
||||
"""Check that **thing** is bytes.
|
||||
|
||||
:param thing: The thing to check if it's bytes.
|
||||
:rtype: bool
|
||||
:returns: ``True`` if **thing** is bytes or a bytearray.
|
||||
"""
|
||||
if isinstance(thing, (bytes, bytearray)):
|
||||
return True
|
||||
return False
|
||||
|
||||
def _is_list_or_tuple(instance):
|
||||
"""Check that ``instance`` is a list or tuple.
|
||||
|
||||
|
@ -471,21 +531,26 @@ def _is_gpg2(version):
|
|||
return True
|
||||
return False
|
||||
|
||||
def _make_binary_stream(s, encoding):
|
||||
"""
|
||||
xxx fill me in
|
||||
def _make_binary_stream(thing, encoding=None, armor=True):
|
||||
"""Encode **thing**, then make it stream/file-like.
|
||||
|
||||
:param thing: The thing to turn into a encoded stream.
|
||||
:rtype: ``io.BytesIO`` or ``io.StringIO``.
|
||||
:returns: The encoded **thing**, wrapped in an ``io.BytesIO`` (if
|
||||
available), otherwise wrapped in a ``io.StringIO``.
|
||||
"""
|
||||
if _py3k:
|
||||
if isinstance(thing, str):
|
||||
thing = thing.encode(encoding)
|
||||
else:
|
||||
if type(thing) is not str:
|
||||
thing = thing.encode(encoding)
|
||||
|
||||
try:
|
||||
if _py3k:
|
||||
if isinstance(s, str):
|
||||
s = s.encode(encoding)
|
||||
else:
|
||||
if type(s) is not str:
|
||||
s = s.encode(encoding)
|
||||
from io import BytesIO
|
||||
rv = BytesIO(s)
|
||||
except ImportError:
|
||||
rv = StringIO(s)
|
||||
rv = BytesIO(thing)
|
||||
except NameError:
|
||||
rv = StringIO(thing)
|
||||
|
||||
return rv
|
||||
|
||||
def _make_passphrase(length=None, save=False, file=None):
|
||||
|
|
|
@ -801,7 +801,7 @@ class GPG(GPGBase):
|
|||
key = key.replace('_','-').title()
|
||||
## to set 'cert', 'Key-Usage' must be blank string
|
||||
if not key in ('Key-Usage', 'Subkey-Usage'):
|
||||
if type(u'')(val).strip():
|
||||
if type('')(val).strip():
|
||||
parms[key] = val
|
||||
|
||||
## if Key-Type is 'default', make Subkey-Type also be 'default'
|
||||
|
|
|
@ -626,6 +626,66 @@ class GPGTestCase(unittest.TestCase):
|
|||
passphrase='wrong horse battery staple')
|
||||
self.assertFalse(sig, "Bad passphrase should fail")
|
||||
|
||||
def test_signature_string_passphrase_empty_string(self):
|
||||
"""Test that a signing attempt with passphrase='' creates a valid
|
||||
signature.
|
||||
|
||||
See Issue #82: https://github.com/isislovecruft/python-gnupg/issues/82
|
||||
"""
|
||||
with open(os.path.join(_files, 'test_key_1.sec')) as fh1:
|
||||
res1 = self.gpg.import_keys(fh1.read())
|
||||
key1 = res1.fingerprints[0]
|
||||
|
||||
message = 'abc\ndef\n'
|
||||
sig = self.gpg.sign(message, default_key=key1, passphrase='')
|
||||
self.assertTrue(sig)
|
||||
self.assertTrue(message in str(sig))
|
||||
|
||||
def test_signature_string_passphrase_empty_bytes_literal(self):
|
||||
"""Test that a signing attempt with passphrase=b'' creates a valid
|
||||
signature.
|
||||
|
||||
See Issue #82: https://github.com/isislovecruft/python-gnupg/issues/82
|
||||
"""
|
||||
with open(os.path.join(_files, 'test_key_1.sec')) as fh1:
|
||||
res1 = self.gpg.import_keys(fh1.read())
|
||||
key1 = res1.fingerprints[0]
|
||||
|
||||
message = 'abc\ndef\n'
|
||||
sig = self.gpg.sign(message, default_key=key1, passphrase=b'')
|
||||
self.assertTrue(sig)
|
||||
print("%r" % str(sig))
|
||||
self.assertTrue(message in str(sig))
|
||||
|
||||
def test_signature_string_passphrase_bytes_literal(self):
|
||||
"""Test that a signing attempt with passphrase=b'overalls' creates a
|
||||
valid signature.
|
||||
"""
|
||||
with open(os.path.join(_files, 'kat.sec')) as fh1:
|
||||
res1 = self.gpg.import_keys(fh1.read())
|
||||
key1 = res1.fingerprints[0]
|
||||
|
||||
message = 'abc\ndef\n'
|
||||
sig = self.gpg.sign(message, default_key=key1, passphrase=b'overalls')
|
||||
self.assertTrue(sig)
|
||||
print("%r" % str(sig))
|
||||
self.assertTrue(message in str(sig))
|
||||
|
||||
def test_signature_string_passphrase_None(self):
|
||||
"""Test that a signing attempt with passphrase=None fails creates a
|
||||
valid signature.
|
||||
|
||||
See Issue #82: https://github.com/isislovecruft/python-gnupg/issues/82
|
||||
"""
|
||||
with open(os.path.join(_files, 'test_key_1.sec')) as fh1:
|
||||
res1 = self.gpg.import_keys(fh1.read())
|
||||
key1 = res1.fingerprints[0]
|
||||
|
||||
message = 'abc\ndef\n'
|
||||
sig = self.gpg.sign(message, default_key=key1, passphrase=None)
|
||||
self.assertTrue(sig)
|
||||
self.assertTrue(message in str(sig))
|
||||
|
||||
def test_signature_file(self):
|
||||
"""Test that signing a message file works."""
|
||||
key = self.generate_key("Leonard Adleman", "rsa.com")
|
||||
|
@ -748,17 +808,84 @@ class GPGTestCase(unittest.TestCase):
|
|||
if os.path.isfile(sigfn):
|
||||
os.unlink(sigfn)
|
||||
|
||||
def test_deletion(self):
|
||||
"""Test that key deletion works."""
|
||||
self.gpg.import_keys(KEYS_TO_IMPORT)
|
||||
def test_deletion_public_key(self):
|
||||
"""Test that key deletion for public keys works, and that it leaves the
|
||||
corresponding secret key intact.
|
||||
"""
|
||||
key1 = None
|
||||
key2 = None
|
||||
|
||||
with open(os.path.join(_files, 'test_key_1.sec')) as fh1:
|
||||
res1 = self.gpg.import_keys(fh1.read())
|
||||
key1 = res1.fingerprints[0]
|
||||
|
||||
with open(os.path.join(_files, 'test_key_2.sec')) as fh2:
|
||||
res2 = self.gpg.import_keys(fh2.read())
|
||||
key2 = res2.fingerprints[0]
|
||||
|
||||
public_keys = self.gpg.list_keys()
|
||||
self.assertTrue(is_list_with_len(public_keys, 2),
|
||||
"2-element list expected, got %d" % len(public_keys))
|
||||
self.gpg.delete_keys(public_keys[0]['fingerprint'])
|
||||
self.assertTrue(len(public_keys), 2)
|
||||
|
||||
self.gpg.delete_keys(key1)
|
||||
|
||||
public_keys = self.gpg.list_keys()
|
||||
self.assertTrue(is_list_with_len(public_keys, 1),
|
||||
"1-element list expected, got %d" % len(public_keys))
|
||||
log.debug("test_deletion ends")
|
||||
secret_keys = self.gpg.list_keys(secret=True)
|
||||
self.assertTrue(len(public_keys), 1)
|
||||
self.assertTrue(len(secret_keys), 2)
|
||||
|
||||
def test_deletion_secret_key(self):
|
||||
"""Test that key deletion for secret keys works, and that it leaves the
|
||||
corresponding public key intact.
|
||||
"""
|
||||
key1 = None
|
||||
key2 = None
|
||||
|
||||
with open(os.path.join(_files, 'test_key_1.sec')) as fh1:
|
||||
res1 = self.gpg.import_keys(fh1.read())
|
||||
key1 = res1.fingerprints[0]
|
||||
|
||||
with open(os.path.join(_files, 'test_key_2.sec')) as fh2:
|
||||
res2 = self.gpg.import_keys(fh2.read())
|
||||
key2 = res2.fingerprints[0]
|
||||
|
||||
public_keys = self.gpg.list_keys()
|
||||
secret_keys = self.gpg.list_keys(secret=True)
|
||||
self.assertEqual(len(public_keys), 2)
|
||||
self.assertEqual(len(secret_keys), 2)
|
||||
|
||||
self.gpg.delete_keys(key1, secret=True)
|
||||
|
||||
public_keys = self.gpg.list_keys()
|
||||
secret_keys = self.gpg.list_keys(secret=True)
|
||||
self.assertEqual(len(public_keys), 2)
|
||||
self.assertEqual(len(secret_keys), 1)
|
||||
|
||||
def test_deletion_subkeys(self):
|
||||
"""Test that key deletion for subkeys deletes both the public and
|
||||
secret portions of the key.
|
||||
"""
|
||||
key1 = None
|
||||
key2 = None
|
||||
|
||||
with open(os.path.join(_files, 'test_key_1.sec')) as fh1:
|
||||
res1 = self.gpg.import_keys(fh1.read())
|
||||
key1 = res1.fingerprints[0]
|
||||
|
||||
with open(os.path.join(_files, 'test_key_2.sec')) as fh2:
|
||||
res2 = self.gpg.import_keys(fh2.read())
|
||||
key2 = res2.fingerprints[0]
|
||||
|
||||
public_keys = self.gpg.list_keys()
|
||||
secret_keys = self.gpg.list_keys(secret=True)
|
||||
self.assertEqual(len(public_keys), 2)
|
||||
self.assertEqual(len(secret_keys), 2)
|
||||
|
||||
self.gpg.delete_keys(key1, subkeys=True)
|
||||
|
||||
public_keys = self.gpg.list_keys()
|
||||
secret_keys = self.gpg.list_keys(secret=True)
|
||||
self.assertEqual(len(public_keys), 1)
|
||||
self.assertEqual(len(secret_keys), 1)
|
||||
|
||||
def test_encryption(self):
|
||||
"""Test encryption of a message string"""
|
||||
|
@ -955,6 +1082,29 @@ authentication."""
|
|||
|
||||
self.assertEqual(message, decrypted)
|
||||
|
||||
def test_decryption_with_bytes_literal(self):
|
||||
"""Test that ``decrypt(encrypt(b'foo'), ...)`` is successful."""
|
||||
with open(os.path.join(_files, 'kat.sec')) as katsec:
|
||||
self.gpg.import_keys(katsec.read())
|
||||
kat = self.gpg.list_keys('kat')[0]['fingerprint']
|
||||
|
||||
message_filename = os.path.join(_files, 'cypherpunk_manifesto')
|
||||
with open(message_filename, 'rb') as f:
|
||||
output = os.path.join(self.gpg.homedir, 'test-decryption-with-bytes-literal.gpg')
|
||||
kwargs = dict(compress_algo='Uncompressed')
|
||||
message = b'Dance like a psycho'
|
||||
encrypted = self.gpg.encrypt(message, kat, **kwargs)
|
||||
self.assertTrue(encrypted.ok)
|
||||
self.assertGreater(len(str(encrypted)), 0)
|
||||
|
||||
decrypted = self.gpg.decrypt(encrypted.data, passphrase='overalls')
|
||||
self.assertTrue(decrypted.ok)
|
||||
self.assertGreater(len(str(decrypted)), 0)
|
||||
# Decode the message so that we can easily compare it with the
|
||||
# decrypted version in both Python2 and Python3:
|
||||
decoded = message.decode(self.gpg._encoding, self.gpg._decode_errors)
|
||||
self.assertEqual(str(decrypted), decoded)
|
||||
|
||||
def test_encryption_one_hidden_recipient_one_not(self):
|
||||
"""Test to ensure hidden recipient isn't detailed in packet info"""
|
||||
|
||||
|
@ -1249,6 +1399,10 @@ suites = { 'parsers': set(['test_parsers_fix_unsafe',
|
|||
'test_signature_verification_detached',
|
||||
'test_signature_verification_detached_binary',
|
||||
'test_signature_file',
|
||||
'test_signature_string_passphrase_empty_string',
|
||||
'test_signature_string_passphrase_empty_bytes_literal',
|
||||
'test_signature_string_passphrase_bytes_literal',
|
||||
'test_signature_string_passphrase_None',
|
||||
'test_signature_string_bad_passphrase',
|
||||
'test_signature_string_verification',
|
||||
'test_signature_string_algorithm_encoding']),
|
||||
|
@ -1263,6 +1417,7 @@ suites = { 'parsers': set(['test_parsers_fix_unsafe',
|
|||
'test_encryption_one_hidden_recipient_one_not',
|
||||
'test_encryption_throw_keyids',
|
||||
'test_decryption',
|
||||
'test_decryption_with_bytes_literal',
|
||||
'test_symmetric_encryption_and_decryption',
|
||||
'test_file_encryption_and_decryption',
|
||||
'test_encryption_to_filename',
|
||||
|
@ -1273,7 +1428,9 @@ suites = { 'parsers': set(['test_parsers_fix_unsafe',
|
|||
'keyrings': set(['test_public_keyring',
|
||||
'test_secret_keyring',
|
||||
'test_import_and_export',
|
||||
'test_deletion',
|
||||
'test_deletion_public_key',
|
||||
'test_deletion_secret_key',
|
||||
'test_deletion_subkeys',
|
||||
'test_import_only']),
|
||||
'recvkeys': set(['test_recv_keys_default']),
|
||||
}
|
||||
|
|
Loading…
Reference in New Issue