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

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