diff --git a/Makefile b/Makefile index c992d03..52cef67 100644 --- a/Makefile +++ b/Makefile @@ -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 diff --git a/gnupg/_meta.py b/gnupg/_meta.py index 5efbd04..ed0babf 100644 --- a/gnupg/_meta.py +++ b/gnupg/_meta.py @@ -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.") diff --git a/gnupg/_parsers.py b/gnupg/_parsers.py index 93da10b..9de57d2 100644 --- a/gnupg/_parsers.py +++ b/gnupg/_parsers.py @@ -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) diff --git a/gnupg/_util.py b/gnupg/_util.py index 8afece5..9a7bc1b 100644 --- a/gnupg/_util.py +++ b/gnupg/_util.py @@ -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')) + # + # + # 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 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 . We hit the + # following error when the `outstream` has type + # . + 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 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 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. diff --git a/gnupg/gnupg.py b/gnupg/gnupg.py index 7168017..ae15293 100644 --- a/gnupg/gnupg.py +++ b/gnupg/gnupg.py @@ -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 diff --git a/gnupg/test/test_gnupg.py b/gnupg/test/test_gnupg.py index 49f5ba5..63ca1f2 100755 --- a/gnupg/test/test_gnupg.py +++ b/gnupg/test/test_gnupg.py @@ -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: