Improve handling of exceptions in background threads.

Author: vsajip

gnupg.py

@@ -39,6 +39,10 @@
 from io import StringIO
 import logging
 import os
+try:
+    from queue import Queue, Empty
+except ImportError:
+    from Queue import Queue, Empty
 import re
 import socket
 from subprocess import Popen, PIPE
@@ -137,7 +141,7 @@ def no_quote(s):
     return s
 
 
-def _copy_data(instream, outstream, buffer_size):
+def _copy_data(instream, outstream, buffer_size, error_queue):
     # Copy one stream to another
     assert buffer_size > 0
     sent = 0
@@ -150,8 +154,10 @@ def _copy_data(instream, outstream, buffer_size):
         # for what is actually a binary file
         try:
             data = instream.read(buffer_size)
-        except Exception:  # pragma: no cover
+        except Exception as e:  # pragma: no cover
             logger.warning('Exception occurred while reading', exc_info=1)
+            error_queue.put_nowait(e)
+            logger.debug('queued exception: %s', e)
             break
         if not data:
             break
@@ -161,10 +167,12 @@ def _copy_data(instream, outstream, buffer_size):
             outstream.write(data)
         except UnicodeError:  # pragma: no cover
             outstream.write(data.encode(enc))
-        except Exception:  # pragma: no cover
+        except Exception as e:  # pragma: no cover
             # Can sometimes get 'broken pipe' errors even when the data has all
             # been sent
             logger.exception('Error sending data')
+            error_queue.put_nowait(e)
+            logger.debug('queued exception: %s', e)
             break
     try:
         outstream.close()
@@ -173,9 +181,9 @@ def _copy_data(instream, outstream, buffer_size):
     logger.debug('closed output, %d bytes sent', sent)
 
 
-def _threaded_copy_data(instream, outstream, buffer_size):
+def _threaded_copy_data(instream, outstream, buffer_size, error_queue):
     assert buffer_size > 0
-    wr = threading.Thread(target=_copy_data, args=(instream, outstream, buffer_size))
+    wr = threading.Thread(target=_copy_data, args=(instream, outstream, buffer_size, error_queue))
     wr.daemon = True
     logger.debug('data copier: %r, %r, %r', wr, instream, outstream)
     wr.start()
@@ -1358,8 +1366,15 @@ def _handle_io(self, args, fileobj_or_path, result, passphrase=None, binary=Fals
                 stdin = p.stdin
             if passphrase:
                 _write_passphrase(stdin, passphrase, self.encoding)
-            writer = _threaded_copy_data(fileobj, stdin, self.buffer_size)
+            error_queue = Queue()
+            writer = _threaded_copy_data(fileobj, stdin, self.buffer_size, error_queue)
             self._collect_output(p, result, writer, stdin)
+            try:
+                exc = error_queue.get_nowait()
+                # if we get here, that means an error occurred in the copying thread
+                raise exc
+            except Empty:
+                pass
             return result
         finally:
             if writer:
@@ -1481,14 +1496,21 @@ def sign_file(self,
         # passphrase is bad, gpg bails and you can't write the message.
         fileobj = self._get_fileobj(fileobj_or_path)
         p = self._open_subprocess(args, passphrase is not None)
+        writer = None
         try:
             stdin = p.stdin
             if passphrase:
                 _write_passphrase(stdin, passphrase, self.encoding)
-            writer = _threaded_copy_data(fileobj, stdin, self.buffer_size)
+            error_queue = Queue()
+            writer = _threaded_copy_data(fileobj, stdin, self.buffer_size, error_queue)
+            try:
+                exc = error_queue.get_nowait()
+                # if we get here, that means an error occurred in the copying thread
+                raise exc
+            except Empty:
+                pass
         except IOError:  # pragma: no cover
             logging.exception('error writing message')
-            writer = None
         finally:
             if writer:
                 writer.join(0.01)

test_gnupg.py

@@ -5,6 +5,7 @@
 Copyright (C) 2008-2024 Vinay Sajip. All rights reserved.
 """
 import argparse
+import io
 import json
 import logging
 import os.path
@@ -1167,12 +1168,12 @@ def do_file_encryption_and_decryption(self, encfname, decfname):
             # Try opening the encrypted file in text mode (Issue #39)
             # this doesn't fail in 2.x
             if gnupg._py3k:
-                efile = open(encfname, 'r')
-                ddata = self.gpg.decrypt_file(efile, passphrase='bbrown', output=decfname)
-                self.assertEqual(2, ddata.returncode, 'Unexpected return code')
-                self.assertFalse(ddata)
-                self.assertEqual(ddata.status, 'no data was provided')
-                efile.close()
+                logger.debug('about to pass text stream to decrypt_file')
+                with open(encfname, 'r') as efile:
+                    self.assertRaises(UnicodeDecodeError, self.gpg.decrypt_file, efile, passphrase='bbrown', output=decfname)
+                    # self.assertEqual(2, ddata.returncode, 'Unexpected return code')
+                    # self.assertFalse(ddata)
+                    # self.assertEqual(ddata.status, 'no data was provided')
         finally:
             for fn in (encfname, decfname):
                 if os.name == 'posix' and mode is not None:
@@ -1587,6 +1588,15 @@ def test_configured_group(self):
         gpg = gnupg.GPG(gnupghome=self.homedir, gpgbinary=GPGBINARY)
         self.assertEqual(gpg.version, self.gpg.version)
 
+    def test_exception_propagation(self):
+        if sys.version_info[0] < 3:
+            raise unittest.SkipTest('python 2 is too loose with Unicode')
+        key = self.generate_key('Andrew', 'Able', 'alpha.com', passphrase='andy')
+        self.assertEqual(0, key.returncode, 'Non-zero return code')
+        andrew = key.fingerprint
+        stream = io.StringIO(u'Hello, world!')  # make the wrong type of stream
+        self.assertRaises(TypeError, self.gpg.encrypt_file, stream, [andrew], armor=False)
+
 
 TEST_GROUPS = {
     'sign':
@@ -1609,7 +1619,7 @@ def test_configured_group(self):
     'basic':
     set(['test_environment', 'test_list_keys_initial', 'test_nogpg', 'test_make_args', 'test_quote_with_shell']),
     'test':
-    set(['test_configured_group']),
+    set(['test_filenames_with_spaces']),
 }
 
 
@@ -1630,11 +1640,19 @@ def suite(args=None):
 
 
 def init_logging():
+    class PrimegenFilter(logging.Filter):
+        def filter(self, record):
+            arg = record.args
+            if isinstance(arg, (list, tuple)) and len(arg) > 0:
+                arg = arg[0]
+            return not arg or not isinstance(arg, unicode) or '[GNUPG:] PROGRESS primegen' not in arg
+
     logging.basicConfig(level=logging.DEBUG,
                         filename='test_gnupg.log',
                         filemode='w',
                         format='%(asctime)s %(levelname)-5s %(name)-10s '
                         '%(threadName)-10s %(lineno)4d %(message)s')
+    logging.root.handlers[0].addFilter(PrimegenFilter())
 
 
 def main():