gajim: update python-gnupg to 0.3.0 + add KEYEXPIRED support. Fi... - Commits

19 Jun 2012

changeset 4a0e9afc56f3 in /home/hg/repos/gajim
details:http://hg.gajim.org/gajim?cmd=changeset;node=4a0e9afc56f3
description: update python-gnupg to 0.3.0 + add KEYEXPIRED support. Fixes #7151
diffstat:
src/common/gnupg.py |  1190 ++++++++++++++++++++++++++------------------------
 src/common/gpg.py   |     4 +-
 2 files changed, 622 insertions(+), 572 deletions(-)
diffs (truncated from 1364 to 300 lines):
diff -r 3997ab204178 -r 4a0e9afc56f3 src/common/gnupg.py

--- a/src/common/gnupg.py	Wed May 09 12:16:48 2012 +0400
+++ b/src/common/gnupg.py	Fri May 18 16:58:51 2012 +0200
@@ -27,25 +27,22 @@
 and so does not work on Windows). Renamed to gnupg.py to avoid confusion with
 the previous versions.
-Modifications Copyright (C) 2008-2011 Vinay Sajip. All rights reserved.
+Modifications Copyright (C) 2008-2012 Vinay Sajip. All rights reserved.
A unittest harness (test_gnupg.py) has also been added.
 """
 import locale
+__version__ = "0.3.0"
 __author__ = "Vinay Sajip"
-__date__  = "$25-Jan-2011 11:40:48$"
+__date__  = "$12-May-2012 10:49:10$"
try:
     from io import StringIO
-    from io import TextIOWrapper
-    from io import BufferedReader
-    from io import BufferedWriter
 except ImportError:
     from cStringIO import StringIO
-    class BufferedReader: pass
-    class BufferedWriter: pass
+import codecs
 import locale
 import logging
 import os
@@ -110,34 +107,11 @@
     passphrase = '%s\n' % passphrase
     passphrase = passphrase.encode(encoding)
     stream.write(passphrase)
-    logger.debug("Passphrase written")
+    logger.debug("Wrote passphrase: %r", passphrase)
def _is_sequence(instance):
     return isinstance(instance,list) or isinstance(instance,tuple)
-def _wrap_input(inp):
-    if isinstance(inp, BufferedWriter):
-        oldinp = inp
-        inp = TextIOWrapper(inp)
-        logger.debug('wrapped input: %r -> %r', oldinp, inp)
-    return inp
-
-def _wrap_output(outp):
-    if isinstance(outp, BufferedReader):
-        oldoutp = outp
-        outp = TextIOWrapper(outp)
-        logger.debug('wrapped output: %r -> %r', oldoutp, outp)
-    return outp
-
-#The following is needed for Python2.7 :-(
-def _make_file(s):
-    try:
-        rv = StringIO(s)
-    except (TypeError, UnicodeError):
-        from io import BytesIO
-        rv = BytesIO(s)
-    return rv
-
 def _make_binary_stream(s, encoding):
     try:
         if _py3k:
@@ -152,526 +126,11 @@
         rv = StringIO(s)
     return rv
-class GPG(object):
-    "Encapsulate access to the gpg executable"
-    def __init__(self, gpgbinary='gpg', gnupghome=None, verbose=False, use_agent=False):
-        """Initialize a GPG process wrapper.  Options are:
-
-        gpgbinary -- full pathname for GPG binary.
-
-        gnupghome -- full pathname to where we can find the public and
-        private keyrings.  Default is whatever gpg defaults to.
-        """
-        self.gpgbinary = gpgbinary
-        self.gnupghome = gnupghome
-        self.verbose = verbose
-        self.use_agent = use_agent
-        self.encoding = locale.getpreferredencoding()
-        if self.encoding is None: # This happens on Jython!
-            self.encoding = sys.stdin.encoding
-        if gnupghome and not os.path.isdir(self.gnupghome):
-            os.makedirs(self.gnupghome,0x1C0)
-        p = self._open_subprocess(["--version"])
-        result = Verify() # any result will do for this
-        self._collect_output(p, result, stdin=p.stdin)
-        if p.returncode != 0:
-            raise ValueError("Error invoking gpg: %s: %s" % (p.returncode,
-                                                             result.stderr))
-
-    def _open_subprocess(self, args, passphrase=False):
-        # Internal method: open a pipe to a GPG subprocess and return
-        # the file objects for communicating with it.
-        cmd = [self.gpgbinary, '--status-fd 2 --no-tty']
-        if self.gnupghome:
-            cmd.append('--homedir "%s" ' % self.gnupghome)
-        if passphrase:
-            cmd.append('--batch --passphrase-fd 0')
-        if self.use_agent:
-            cmd.append('--use-agent')
-        cmd.extend(args)
-        cmd = ' '.join(cmd)
-        if self.verbose:
-            print(cmd)
-        logger.debug("%s", cmd)
-        return Popen(cmd, shell=True, stdin=PIPE, stdout=PIPE, stderr=PIPE)
-
-    def _read_response(self, stream, result):
-        # Internal method: reads all the output from GPG, taking notice
-        # only of lines that begin with the magic [GNUPG:] prefix.
-        #
-        # Calls methods on the response object for each valid token found,
-        # with the arg being the remainder of the status line.
-        lines = []
-        while True:
-            line = stream.readline()
-            lines.append(line)
-            if self.verbose:
-                print(line)
-            logger.debug("%s", line.rstrip())
-            if line == "": break
-            line = line.rstrip()
-            if line[0:9] == '[GNUPG:] ':
-                # Chop off the prefix
-                line = line[9:]
-                L = line.split(None, 1)
-                keyword = L[0]
-                if len(L) > 1:
-                    value = L[1]
-                else:
-                    value = ""
-                result.handle_status(keyword, value)
-        result.stderr = ''.join(lines)
-
-    def _read_data(self, stream, result):
-        # Read the contents of the file from GPG's stdout
-        chunks = []
-        while True:
-            data = stream.read(1024)
-            if len(data) == 0:
-                break
-            logger.debug("chunk: %r" % data[:256])
-            chunks.append(data)
-        if _py3k:
-            # Join using b'' or '', as appropriate
-            result.data = type(data)().join(chunks)
-        else:
-            result.data = ''.join(chunks)
-
-    def _collect_output(self, process, result, writer=None, stdin=None):
-        """
-        Drain the subprocesses output streams, writing the collected output
-        to the result. If a writer thread (writing to the subprocess) is given,
-        make sure it's joined before returning. If a stdin stream is given,
-        close it before returning.
-        """
-        stderr = _wrap_output(process.stderr)
-        rr = threading.Thread(target=self._read_response, args=(stderr, result))
-        rr.setDaemon(True)
-        logger.debug('stderr reader: %r', rr)
-        rr.start()
-
-        stdout = process.stdout # _wrap_output(process.stdout)
-        dr = threading.Thread(target=self._read_data, args=(stdout, result))
-        dr.setDaemon(True)
-        logger.debug('stdout reader: %r', dr)
-        dr.start()
-
-        dr.join()
-        rr.join()
-        if writer is not None:
-            writer.join()
-        process.wait()
-        if stdin is not None:
-            try:
-                stdin.close()
-            except IOError:
-                pass
-        stderr.close()
-        stdout.close()
-
-    def _handle_io(self, args, file, result, passphrase=None, binary=False):
-        "Handle a call to GPG - pass input data, collect output data"
-        # Handle a basic data call - pass data to GPG, handle the output
-        # including status information. Garbage In, Garbage Out :)
-        p = self._open_subprocess(args, passphrase is not None)
-        if not binary and not isinstance(file, BufferedReader):
-            stdin = _wrap_input(p.stdin)
-        else:
-            stdin = p.stdin
-        if passphrase:
-            _write_passphrase(stdin, passphrase, self.encoding)
-        writer = _threaded_copy_data(file, stdin)
-        self._collect_output(p, result, writer, stdin)
-        return result
-
-    #
-    # SIGNATURE METHODS
-    #
-    def sign(self, message, **kwargs):
-        """sign message"""
-        f = _make_binary_stream(message, self.encoding)
-        result = self.sign_file(f, **kwargs)
-        f.close()
-        return result
-
-    def sign_file(self, file, keyid=None, passphrase=None, clearsign=True,
-                  detach=False, binary=False):
-        """sign file"""
-        logger.debug("sign_file: %s", file)
-        if binary:
-            args = ['-s']
-        else:
-            args = ['-sa']
-        # You can't specify detach-sign and clearsign together: gpg ignores
-        # the detach-sign in that case.
-        if detach:
-            args.append("--detach-sign")
-        elif clearsign:
-            args.append("--clearsign")
-        if keyid:
-            args.append("--default-key %s" % keyid)
-        result = Sign(self.encoding)
-        #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.
-        #self._handle_io(args, _make_file(message), result, passphrase=passphrase)
-        p = self._open_subprocess(args, passphrase is not None)
-        try:
-            stdin = p.stdin
-            if passphrase:
-                _write_passphrase(stdin, passphrase, self.encoding)
-            writer = _threaded_copy_data(file, stdin)
-        except IOError:
-            logging.exception("error writing message")
-            writer = None
-        self._collect_output(p, result, writer, stdin)
-        return result
-
-    def verify(self, data):
-        """Verify the signature on the contents of the string 'data'
-
-        >>> gpg = GPG(gnupghome="keys")
-        >>> input = gpg.gen_key_input(Passphrase='foo')
-        >>> key = gpg.gen_key(input)
-        >>> assert key
-        >>> sig = gpg.sign('hello',keyid=key.fingerprint,passphrase='bar')
-        >>> assert not sig
-        >>> sig = gpg.sign('hello',keyid=key.fingerprint,passphrase='foo')
-        >>> assert sig
-        >>> verify = gpg.verify(sig.data)
-        >>> assert verify
-
-        """
-        f = _make_binary_stream(data, self.encoding)
-        result = self.verify_file(f)
-        f.close()
-        return result
-
-    def verify_file(self, file, data_filename=None):
-        "Verify the signature on the contents of the file-like object 'file'"
-        logger.debug('verify_file: %r, %r', file, data_filename)
-        result = Verify()
-        args = ['--verify']
-        if data_filename is None:
-            self._handle_io(args, file, result, binary=True)
-        else:
-            logger.debug('Handling detached verification')
-            import tempfile
-            fd, fn = tempfile.mkstemp(prefix='pygpg')
-            s = file.read()
-            file.close()
-            logger.debug('Wrote to temp file: %r', s)
-            os.write(fd, s)
-            os.close(fd)
-            args.append(fn)
-            args.append(data_filename)
-            try:
-                p = self._open_subprocess(args)
-                self._collect_output(p, result, stdin=p.stdin)
-            finally:
-                os.unlink(fn)
-        return result
-
-    #
-    # KEY MANAGEMENT
-    #
-
-    def import_keys(self, key_data):
-        """ import the key_data into our keyring
-
-        >>> import shutil

    