Unsignedio
/
python-gnupg
mirror of https://github.com/markqvist/python-gnupg.git
1
0
Fork 0
Code Issues Releases Wiki Activity

Merge branch 'release/2.0.1'

master
Isis Lovecruft 2015-03-13 05:06:34 +00:00
parent 939728694c cf4d3efd8d
commit af781626a9
No known key found for this signature in database
GPG Key ID: 18C16EC5F9F1D673
6 changed files with 358 additions and 81 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

71
Makefile
View File

@ -2,6 +2,8 @@ SHELL=/bin/sh
TESTDIR=./gnupg/test
TESTHANDLE=$(TESTDIR)/test_gnupg.py
FILES=$(SHELL find ./gnupg/ -name "*.py" -printf "%p,")
PYTHON=$(SHELL which python)
PYTHON3=$(SHELL which python3)
PKG_NAME=python-gnupg
DOC_DIR=docs
DOC_BUILD_DIR:=$(DOC_DIR)/_build
@ -50,23 +52,70 @@ test-before: cleanup-src cleanup-tests
which python && python --version
-which pip && pip --version && pip list
test: test-before
python $(TESTHANDLE) basic encodings parsers keyrings listkeys genkey \
sign crypt
test-run: test-before
python $(TESTHANDLE) \
basic \
encodings \
parsers \
keyrings \
listkeys \
genkey \
sign \
crypt
py3k-test-run: test-before
python3 $(TESTHANDLE) \
basic \
encodings \
parsers \
keyrings \
listkeys \
genkey \
sign \
crypt
coverage-run: test-before
coverage run --rcfile=".coveragerc" $(PYTHON) $(TESTHANDLE) \
basic \
encodings \
parsers \
keyrings \
listkeys \
genkeys \
sign \
crypt
py3k-coverage-run: test-before
coverage run --rcfile=".coveragerc" $(PYTHON3) $(TESTHANDLE) \
basic \
encodings \
parsers \
keyrings \
listkeys \
genkeys \
sign \
crypt
coverage-report:
coverage report --rcfile=".coveragerc"
coverage-html:
coverage html --rcfile=".coveragerc"
clean-test:
touch gnupg/test/placeholder.log
mv gnupg/test/*.log gnupg/test/logs/
rm gnupg/test/logs/placeholder.log
touch gnupg/test/random_seed_is_sekritly_pi
rm gnupg/test/random_seed*
py3k-test: test-before
python3 $(TESTHANDLE) basic encodings parsers keyrings listkeys genkey \
sign crypt
touch gnupg/test/placeholder.log
mv gnupg/test/*.log gnupg/test/logs/
rm gnupg/test/logs/placeholder.log
touch gnupg/test/random_seed_is_sekritly_pi
rm gnupg/test/random_seed*
test: test-run clean-test
py3k-test: py3k-test-run clean-test
coverage: coverage-run coverage-report coverage-html clean-test
py3k-coverage: py3k-coverage-run coverage-report coverage-html clean-test
install:
python setup.py install --record installed-files.txt

47
gnupg/_meta.py
View File

@ -147,7 +147,7 @@ class GPGBase(object):
def __init__(self, binary=None, home=None, keyring=None, secring=None,
use_agent=False, default_preference_list=None,
verbose=False, options=None):
ignore_homedir_permissions=False, verbose=False, options=None):
"""Create a ``GPGBase``.
This class is used to set up properties for controlling the behaviour
@ -170,13 +170,14 @@ class GPGBase(object):
:ivar str secring: The filename in **homedir** to use as the keyring
file for secret keys.
"""
self.ignore_homedir_permissions = ignore_homedir_permissions
self.binary = _util._find_binary(binary)
self.homedir = os.path.expanduser(home) if home else _util._conf
pub = _parsers._fix_unsafe(keyring) if keyring else 'pubring.gpg'
sec = _parsers._fix_unsafe(secring) if secring else 'secring.gpg'
self.keyring = os.path.join(self._homedir, pub)
self.secring = os.path.join(self._homedir, sec)
self.options = _parsers._sanitise(options) if options else None
self.options = list(_parsers._sanitise_list(options)) if options else None
#: The version string of our GnuPG binary
self.binary_version = '0.0.0'
@ -212,7 +213,7 @@ class GPGBase(object):
"'verbose' must be boolean, string, or 0 <= n <= 9"
assert isinstance(use_agent, bool), "'use_agent' must be boolean"
if self.options is not None:
assert isinstance(self.options, str), "options not string"
assert isinstance(self.options, list), "options not list"
except (AssertionError, AttributeError) as ae:
log.error("GPGBase.__init__(): %s" % str(ae))
raise RuntimeError(str(ae))
@ -413,18 +414,21 @@ class GPGBase(object):
log.debug("GPGBase._homedir_setter(): Check existence of '%s'" % hd)
_util._create_if_necessary(hd)
try:
log.debug("GPGBase._homedir_setter(): checking permissions")
assert _util._has_readwrite(hd), \
"Homedir '%s' needs read/write permissions" % hd
except AssertionError as ae:
msg = ("Unable to set '%s' as GnuPG homedir" % directory)
log.debug("GPGBase.homedir.setter(): %s" % msg)
log.debug(str(ae))
raise RuntimeError(str(ae))
else:
log.info("Setting homedir to '%s'" % hd)
if self.ignore_homedir_permissions:
self._homedir = hd
else:
try:
log.debug("GPGBase._homedir_setter(): checking permissions")
assert _util._has_readwrite(hd), \
"Homedir '%s' needs read/write permissions" % hd
except AssertionError as ae:
msg = ("Unable to set '%s' as GnuPG homedir" % directory)
log.debug("GPGBase.homedir.setter(): %s" % msg)
log.debug(str(ae))
raise RuntimeError(str(ae))
else:
log.info("Setting homedir to '%s'" % hd)
self._homedir = hd
homedir = _util.InheritableProperty(_homedir_getter, _homedir_setter)
@ -533,8 +537,8 @@ class GPGBase(object):
if passphrase: cmd.append('--batch --passphrase-fd 0')
if self.use_agent: cmd.append('--use-agent')
else: cmd.append('--no-use-agent')
if self.use_agent is True: cmd.append('--use-agent')
elif self.use_agent is False: cmd.append('--no-use-agent')
# The arguments for debugging and verbosity should be placed into the
# cmd list before the options/args in order to resolve Issue #76:
@ -990,21 +994,18 @@ class GPGBase(object):
for recp in recipients.split(' '):
self._add_recipient_string(args, hidden_recipients, recp)
## ...and now that we've proven py3k is better...
else:
log.debug("Don't know what to do with recipients: '%s'"
log.debug("Don't know what to do with recipients: %r"
% recipients)
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("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)
if output_filename:
log.info("Writing encrypted output to file: %s" % output_filename)
with open(output_filename, 'w+') as fh:
with open(output_filename, 'wb') as fh:
fh.write(result.data)
fh.flush()
log.info("Encrypted output written successfully.")

27
gnupg/_parsers.py
View File

@ -367,7 +367,7 @@ def _sanitise(*args):
checked += (val + " ")
log.debug("_check_option(): No checks for %s" % val)
return checked
return checked.rstrip(' ')
is_flag = lambda x: x.startswith('--')
@ -516,6 +516,7 @@ def _get_options_group(group=None):
'--import',
'--verify',
'--verify-files',
'--output',
])
#: These options expect a string. see :func:`_check_preferences`.
pref_options = frozenset(['--digest-algo',
@ -557,6 +558,9 @@ def _get_options_group(group=None):
'--list-public-keys',
'--list-secret-keys',
'--list-sigs',
'--lock-multiple',
'--lock-never',
'--lock-once',
'--no-default-keyring',
'--no-default-recipient',
'--no-emit-version',
@ -908,6 +912,7 @@ class Sign(object):
timestamp = None
#: xxx fill me in
what = None
status = None
def __init__(self, gpg):
self._gpg = gpg
@ -930,9 +935,9 @@ class Sign(object):
:raises: :exc:`~exceptions.ValueError` if the status message is unknown.
"""
if key in ("USERID_HINT", "NEED_PASSPHRASE", "BAD_PASSPHRASE",
"GOOD_PASSPHRASE", "BEGIN_SIGNING", "CARDCTRL",
"INV_SGNR", "SIGEXPIRED"):
pass
"GOOD_PASSPHRASE", "MISSING_PASSPHRASE", "PINENTRY_LAUNCHED",
"BEGIN_SIGNING", "CARDCTRL", "INV_SGNR", "SIGEXPIRED"):
self.status = key.replace("_", " ").lower()
elif key == "SIG_CREATED":
(self.sig_type, self.sig_algo, self.sig_hash_algo,
self.what, self.timestamp, self.fingerprint) = value.split()
@ -949,6 +954,7 @@ class Sign(object):
else:
raise ValueError("Unknown status message: %r" % key)
class ListKeys(list):
"""Handle status messages for --list-keys.
@ -1271,7 +1277,8 @@ class Verify(object):
self.trust_level = self.TRUST_LEVELS[key]
elif key in ("RSA_OR_IDEA", "NODATA", "IMPORT_RES", "PLAINTEXT",
"PLAINTEXT_LENGTH", "POLICY_URL", "DECRYPTION_INFO",
"DECRYPTION_OKAY", "INV_SGNR", "PROGRESS"):
"DECRYPTION_OKAY", "INV_SGNR", "PROGRESS",
"PINENTRY_LAUNCHED"):
pass
elif key == "BADSIG":
self.valid = False
@ -1524,21 +1531,21 @@ class ListPackets(object):
:raises: :exc:`~exceptions.ValueError` if the status message is unknown.
"""
if key == 'NODATA':
if key in ('NO_SECKEY', 'BEGIN_DECRYPTION', 'DECRYPTION_FAILED',
'END_DECRYPTION', 'GOOD_PASSPHRASE', 'BAD_PASSPHRASE'):
pass
elif key == 'NODATA':
self.status = nodata(value)
elif key == 'ENC_TO':
key, _, _ = value.split()
if not self.key:
self.key = key
self.encrypted_to.append(key)
elif key == 'NEED_PASSPHRASE':
elif key == ('NEED_PASSPHRASE', 'MISSING_PASSPHRASE'):
self.need_passphrase = True
elif key == 'NEED_PASSPHRASE_SYM':
self.need_passphrase_sym = True
elif key == 'USERID_HINT':
self.userid_hint = value.strip().split()
elif key in ('NO_SECKEY', 'BEGIN_DECRYPTION', 'DECRYPTION_FAILED',
'END_DECRYPTION'):
pass
else:
raise ValueError("Unknown status message: %r" % key)

121
gnupg/_util.py
View File

@ -34,11 +34,52 @@ import re
import string
import sys
# These are all the classes which are stream-like; they are used in
# :func:`_is_stream`.
_STREAMLIKE_TYPES = []
# These StringIO classes are actually utilised.
try:
import io
from io import StringIO
from io import BytesIO
except ImportError:
from cStringIO import StringIO
else:
# The io.IOBase type covers the above example for an open file handle in
# Python3, as well as both io.BytesIO and io.StringIO.
_STREAMLIKE_TYPES.append(io.IOBase)
# 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:
# 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
# name is prefixed with an underscore to distinguish it.
from StringIO import StringIO as _StringIO_StringIO
_STREAMLIKE_TYPES.append(_StringIO_StringIO)
# Import the cStringIO module to test for the cStringIO stream types,
# InputType and OutputType. See
# http://stackoverflow.com/questions/14735295/to-check-an-instance-is-stringio
import cStringIO as _cStringIO
_STREAMLIKE_TYPES.append(_cStringIO.InputType)
_STREAMLIKE_TYPES.append(_cStringIO.OutputType)
# In Python2:
#
# >>> type(open('README.md', 'rb'))
# <open file 'README.md', mode 'rb' at 0x7f9493951d20>
#
# whereas, in Python3, the `file` builtin doesn't exist and instead we get:
#
# >>> type(open('README.md', 'rb'))
# <_io.BufferedReader name='README.md'>
#
# which is covered by the above addition of io.IOBase.
_STREAMLIKE_TYPES.append(file)
from . import _logger
@ -142,7 +183,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:
@ -154,24 +194,73 @@ def _copy_data(instream, outstream):
data = instream.read(1024)
if len(data) == 0:
break
sent += len(data)
log.debug("Sending chunk %d bytes:\n%s"
% (sent, data))
try:
outstream.write(data)
except UnicodeError:
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))
if not _py3k:
try:
outstream.write(coder.encode(data))
except IOError:
log.exception("Error sending data: Broken pipe")
outstream.write(encoded)
except IOError as ioe:
# Can get 'broken pipe' errors even when all data was sent
if 'Broken pipe' in str(ioe):
log.error('Error sending data: Broken pipe')
else:
log.exception(ioe)
break
except IOError as ioe:
# Can get 'broken pipe' errors even when all data was sent
if 'Broken pipe' in str(ioe):
log.error('Error sending data: Broken pipe')
else:
log.exception(ioe)
break
log.debug("Wrote data type <type 'str'> to outstream.")
else:
try:
outstream.write(bytes(encoded))
except TypeError as te:
# XXX FIXME This appears to happen because
# _threaded_copy_data() sometimes passes the `outstream` as an
# object with type <_io.BufferredWriter> and at other times
# with type <encodings.utf_8.StreamWriter>. We hit the
# following error when the `outstream` has type
# <encodings.utf_8.StreamWriter>.
if not "convert 'bytes' object to str implicitly" in str(te):
log.error(str(te))
try:
outstream.write(encoded.decode())
except TypeError as yate:
# We hit the "'str' does not support the buffer interface"
# error in Python3 when the `outstream` is an io.BytesIO and
# we try to write a str to it. We don't care about that
# error, we'll just try again with bytes.
if not "does not support the buffer interface" in str(yate):
log.error(str(yate))
except IOError as ioe:
# Can get 'broken pipe' errors even when all data was sent
if 'Broken pipe' in str(ioe):
log.error('Error sending data: Broken pipe')
else:
log.exception(ioe)
break
else:
log.debug("Wrote data type <class 'str'> outstream.")
except IOError as ioe:
# Can get 'broken pipe' errors even when all data was sent
if 'Broken pipe' in str(ioe):
log.error('Error sending data: Broken pipe')
else:
log.exception(ioe)
break
else:
log.debug("Wrote data type <class 'bytes'> to outstream.")
try:
outstream.close()
except IOError as ioe:
@ -349,7 +438,7 @@ def _is_stream(input):
:rtype: bool
:returns: True if :param:input is a stream, False if otherwise.
"""
return isinstance(input, BytesIO) or isinstance(input, StringIO)
return isinstance(input, tuple(_STREAMLIKE_TYPES))
def _is_list_or_tuple(instance):
"""Check that ``instance`` is a list or tuple.

25
gnupg/gnupg.py
View File

@ -60,7 +60,7 @@ class GPG(GPGBase):
def __init__(self, binary=None, homedir=None, verbose=False,
use_agent=False, keyring=None, secring=None,
options=None):
ignore_homedir_permissions=False, options=None):
"""Initialize a GnuPG process wrapper.
:param str binary: Name for GnuPG binary executable. If the absolute
@ -73,6 +73,10 @@ class GPG(GPGBase):
and private keyrings. Default is whatever GnuPG
defaults to.
:type ignore_homedir_permissions: :obj:`bool`
:param ignore_homedir_permissions: If true, bypass check that homedir
be writable.
:type verbose: :obj:`str` or :obj:`int` or :obj:`bool`
:param verbose: String or numeric value to pass to GnuPG's
``--debug-level`` option. See the GnuPG man page for
@ -117,13 +121,16 @@ class GPG(GPGBase):
secring=secring,
options=options,
verbose=verbose,
use_agent=use_agent,)
use_agent=use_agent,
ignore_homedir_permissions=ignore_homedir_permissions,
)
log.info(textwrap.dedent("""
Initialised settings:
binary: %s
binary version: %s
homedir: %s
ignore_homedir_permissions: %s
keyring: %s
secring: %s
default_preference_list: %s
@ -134,6 +141,7 @@ class GPG(GPGBase):
""" % (self.binary,
self.binary_version,
self.homedir,
self.ignore_homedir_permissions,
self.keyring,
self.secring,
self.default_preference_list,
@ -153,6 +161,12 @@ class GPG(GPGBase):
# fatal error (at least it does with GnuPG>=2.0.0):
self.create_trustdb()
# The --no-use-agent and --use-agent options were deprecated in GnuPG
# 2.x, so we should set use_agent to None here to avoid having
# GPGBase._make_args() add either one.
if self.is_gpg2():
self.use_agent = None
@functools.wraps(_trust._create_trustdb)
def create_trustdb(self):
if self.is_gpg2():
@ -787,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 str(val).strip():
if type(u'')(val).strip():
parms[key] = val
## if Key-Type is 'default', make Subkey-Type also be 'default'
@ -952,7 +966,10 @@ generate keys. Please see
.. seealso:: :meth:`._encrypt`
"""
stream = _make_binary_stream(data, self._encoding)
if _is_stream(data):
stream = data
else:
stream = _make_binary_stream(data, self._encoding)
result = self._encrypt(stream, recipients, **kwargs)
stream.close()
return result

148
gnupg/test/test_gnupg.py
View File

@ -26,6 +26,7 @@ A test harness and unittests for gnupg.py.
from __future__ import absolute_import
from __future__ import print_function
from __future__ import with_statement
from argparse import ArgumentParser
from codecs import open as open
from functools import wraps
@ -288,8 +289,8 @@ class GPGTestCase(unittest.TestCase):
self.assertTrue(os.path.isabs(self.gpg.binary))
def test_make_args_drop_protected_options(self):
"""Test that unsupported gpg options are dropped."""
self.gpg.options = ['--tyrannosaurus-rex', '--stegosaurus']
"""Test that unsupported gpg options are dropped, and supported ones remain."""
self.gpg.options = ['--tyrannosaurus-rex', '--stegosaurus', '--lock-never']
gpg_binary_path = _util._find_binary('gpg')
cmd = self.gpg._make_args(None, False)
expected = [gpg_binary_path,
@ -297,7 +298,8 @@ class GPGTestCase(unittest.TestCase):
'--homedir "%s"' % self.homedir,
'--no-default-keyring --keyring %s' % self.keyring,
'--secret-keyring %s' % self.secring,
'--no-use-agent']
'--no-use-agent',
'--lock-never']
self.assertListEqual(cmd, expected)
def test_make_args(self):
@ -388,7 +390,10 @@ class GPGTestCase(unittest.TestCase):
def test_gen_key_input(self):
"""Test that GnuPG batch file creation is successful."""
key_input = self.generate_key_input("Francisco Ferrer", "an.ok")
self.assertIsInstance(key_input, str)
if _util._py3k:
self.assertIsInstance(key_input, str)
else:
self.assertIsInstance(key_input, basestring)
self.assertGreater(key_input.find('Francisco Ferrer'), 0)
def test_rsa_key_generation(self):
@ -776,6 +781,75 @@ authentication."""
log.debug("Encrypted: %s" % encrypted)
self.assertNotEquals(message, encrypted)
def _encryption_test_setup(self):
passphrase = "craiggentry"
key = self.generate_key("Craig Gentry", "xorr.ox", passphrase=passphrase)
fpr = str(key.fingerprint)
gentry = self.gpg.export_keys(key.fingerprint)
self.gpg.import_keys(gentry)
message = """
In 2010 Riggio and Sicari presented a practical application of homomorphic
encryption to a hybrid wireless sensor/mesh network. The system enables
transparent multi-hop wireless backhauls that are able to perform statistical
analysis of different kinds of data (temperature, humidity, etc.) coming from
a WSN while ensuring both end-to-end encryption and hop-by-hop
authentication."""
return (message, fpr, passphrase)
def _encryption_test(self, stream_type, message, fingerprint, passphrase):
stream = stream_type(message)
encrypted = self.gpg.encrypt(stream, fingerprint).data
decrypted = self.gpg.decrypt(encrypted, passphrase=passphrase).data
if isinstance(decrypted, bytes):
decrypted = decrypted.decode()
if isinstance(message, bytes):
message = message.decode()
self.assertEqual(message, decrypted)
def test_encryption_of_file_like_objects_io_StringIO(self):
"""Test encryption of file-like object io.StringIO."""
message, fpr, passphrase = self._encryption_test_setup()
try:
from io import StringIO
if _util._py3k:
self._encryption_test(StringIO, message, fpr, passphrase)
else:
self._encryption_test(StringIO, unicode(message), fpr, passphrase)
except ImportError:
pass
def test_encryption_of_file_like_objects_io_BytesIO(self):
"""Test encryption of file-like object io.BytesIO."""
message, fpr, passphrase = self._encryption_test_setup()
try:
from io import BytesIO
if _util._py3k:
self._encryption_test(BytesIO, bytes(message, 'utf-8'), fpr, passphrase)
else:
self._encryption_test(BytesIO, message, fpr, passphrase)
except ImportError:
pass
def test_encryption_of_file_like_objects_StringIO_StringIO(self):
"""Test encryption of file-like object StringIO.StringIO (Python2 only)."""
message, fpr, passphrase = self._encryption_test_setup()
if not _util._py3k:
from StringIO import StringIO
self._encryption_test(StringIO, message, fpr, passphrase)
def test_encryption_of_file_like_objects_cStringIO_StringIO(self):
"""Test encryption of file-like object cStringIO.StringIO (Python2 only)."""
message, fpr, passphrase = self._encryption_test_setup()
if not _util._py3k:
from cStringIO import StringIO
self._encryption_test(StringIO, message, fpr, passphrase)
def test_encryption_alt_encoding(self):
"""Test encryption with latin-1 encoding"""
key = self.generate_key("Craig Gentry", "xorr.ox",
@ -784,11 +858,7 @@ authentication."""
key = self.generate_key("Marten van Dijk", "xorr.ox")
dijk = str(key.fingerprint)
self.gpg._encoding = 'latin-1'
if _util._py3k:
data = 'Hello, André!'
else:
data = unicode('Hello, André', self.gpg._encoding)
data = data.encode(self.gpg._encoding)
data = u'Hello, André!'.encode(self.gpg._encoding)
encrypted = self.gpg.encrypt(data, gentry)
edata = str(encrypted.data)
self.assertNotEqual(data, edata)
@ -957,7 +1027,8 @@ boolean circuit causes a considerable overhead."""
## We expect Alice's key to be hidden (returned as zero's) and Bob's
## key to be there.
expected_values = ["0000000000000000", "0000000000000000"]
self.assertEquals(expected_values, self.gpg.list_packets(encrypted).encrypted_to)
packets = self.gpg.list_packets(encrypted)
self.assertEquals(expected_values, packets.encrypted_to)
def test_encryption_decryption_multi_recipient(self):
"""Test decryption of an encrypted string for multiple users"""
@ -1081,9 +1152,9 @@ know, maybe you shouldn't be doing it in the first place.
self.assertTrue(os.path.isfile(output))
# Check the contents:
with open(output) as fh:
with open(output, 'rb') as fh:
encrypted_message = fh.read()
log.debug("Encrypted file contains:\n\n%s\n" % encrypted_message)
self.assertTrue(b"-----BEGIN PGP MESSAGE-----" in encrypted_message)
def test_encryption_to_filehandle(self):
"""Test that ``encrypt(..., output=filelikething)`` is successful."""
@ -1103,9 +1174,45 @@ know, maybe you shouldn't be doing it in the first place.
self.assertTrue(os.path.isfile(output))
# Check the contents:
with open(output) as fh:
with open(output, 'rb') as fh:
encrypted_message = fh.read()
log.debug("Encrypted file contains:\n\n%s\n" % encrypted_message)
self.assertTrue(b"-----BEGIN PGP MESSAGE-----" in encrypted_message)
def test_encryption_from_filehandle(self):
"""Test that ``encrypt(open('foo'), ...)`` is successful."""
message_filename = os.path.join(_files, 'cypherpunk_manifesto')
with open(message_filename, 'rb') as f:
output = os.path.join(self.gpg.homedir, 'test-encryption-from-filehandle.gpg')
kwargs = dict(passphrase='speedtest',
symmetric=True,
cipher_algo='AES256',
encrypt=False,
output=output)
encrypted = self.gpg.encrypt(f, None, **kwargs)
self.assertTrue(encrypted.ok)
self.assertGreater(len(encrypted.data), 0)
def test_encryption_with_output(self):
"""Test that ``encrypt('foo', ..., output='/foo/bar/baz')`` is successful."""
message_filename = os.path.join(_files, 'cypherpunk_manifesto')
with open (message_filename, 'rb') as f:
data = f.read()
output = os.path.join(self.gpg.homedir, 'test-encryption-with-output.gpg')
kwargs = dict(passphrase='speedtest',
symmetric=True,
cipher_algo='AES256',
encrypt=False,
output=output)
encrypted = self.gpg.encrypt(data, None, **kwargs)
self.assertTrue(encrypted.ok)
self.assertGreater(len(encrypted.data), 0)
self.assertTrue(os.path.isfile(output))
# Check the contents:
with open(output, 'rb') as fh:
encrypted_message = fh.read()
self.assertTrue(b"-----BEGIN PGP MESSAGE-----" in encrypted_message)
suites = { 'parsers': set(['test_parsers_fix_unsafe',
@ -1146,6 +1253,10 @@ suites = { 'parsers': set(['test_parsers_fix_unsafe',
'test_signature_string_verification',
'test_signature_string_algorithm_encoding']),
'crypt': set(['test_encryption',
'test_encryption_of_file_like_objects_io_StringIO',
'test_encryption_of_file_like_objects_io_BytesIO',
'test_encryption_of_file_like_objects_StringIO_StringIO',
'test_encryption_of_file_like_objects_cStringIO_StringIO',
'test_encryption_alt_encoding',
'test_encryption_multi_recipient',
'test_encryption_decryption_multi_recipient',
@ -1155,14 +1266,17 @@ suites = { 'parsers': set(['test_parsers_fix_unsafe',
'test_symmetric_encryption_and_decryption',
'test_file_encryption_and_decryption',
'test_encryption_to_filename',
'test_encryption_to_filehandle',]),
'test_encryption_to_filehandle',
'test_encryption_from_filehandle',
'test_encryption_with_output',]),
'listkeys': set(['test_list_keys_after_generation']),
'keyrings': set(['test_public_keyring',
'test_secret_keyring',
'test_import_and_export',
'test_deletion',
'test_import_only',
'test_recv_keys_default',]), }
'test_import_only']),
'recvkeys': set(['test_recv_keys_default']),
}
def main(args):
if not args.quiet: