gajim: update to latest gnupg.py

changeset b6b81c4c1f33 in /home/hg/repos/gajim

author: Yann Leboulanger asterix@lagaule.org branches: details:http://hg.gajim.org/gajim?cmd=changeset;node=b6b81c4c1f33 description: update to latest gnupg.py

diffstat:

src/common/gnupg.py | 431 +++++++++++++++++++++++++++++++++++++++------------ 1 files changed, 324 insertions(+), 107 deletions(-)

diffs (truncated from 742 to 300 lines):

diff -r 38cec02c0c91 -r b6b81c4c1f33 src/common/gnupg.py --- a/src/common/gnupg.py Sat Jan 03 20:51:41 2015 +0100 +++ b/src/common/gnupg.py Sat Jan 24 16:09:44 2015 +0100 @@ -31,11 +31,10 @@

A unittest harness (test_gnupg.py) has also been added. """ -import locale

-__version__ = "0.3.4" +__version__ = "0.3.8.dev0" __author__ = "Vinay Sajip" -__date__ = "$05-Jun-2014 09:48:54$" +__date__ = "$07-Dec-2014 18:46:17$"

try: from io import StringIO @@ -53,6 +52,13 @@ import sys import threading

+STARTUPINFO = None +if os.name == 'nt': + try: + from subprocess import STARTUPINFO, STARTF_USESHOWWINDOW, SW_HIDE + except ImportError: + STARTUPINFO = None + try: import logging.NullHandler as NullHandler except ImportError: @@ -73,6 +79,50 @@ if not logger.handlers: logger.addHandler(NullHandler())

+# We use the test below because it works for Jython as well as CPython +if os.path.__name__ == 'ntpath': + # On Windows, we don't need shell quoting, other than worrying about + # paths with spaces in them. + def shell_quote(s): + return '"%s"' % s +else: + # Section copied from sarge + + # This regex determines which shell input needs quoting + # because it may be unsafe + UNSAFE = re.compile(r'[^\w%+,./:=@-]') + + def shell_quote(s): + """ + Quote text so that it is safe for Posix command shells. + + For example, "*.py" would be converted to "'*.py'". If the text is + considered safe it is returned unquoted. + + :param s: The value to quote + :type s: str (or unicode on 2.x) + :return: A safe version of the input, from the point of view of Posix + command shells + :rtype: The passed-in type + """ + if not isinstance(s, string_types): + raise TypeError('Expected string type, got %s' % type(s)) + if not s: + result = "''" + elif not UNSAFE.search(s): + result = s + else: + result = "'%s'" % s.replace("'", r"'''") + return result + + # end of sarge code + +# Now that we use shell=False, we shouldn't need to quote arguments. +# Use no_quote instead of shell_quote to remind us of where quoting +# was needed. +def no_quote(s): + return s + def _copy_data(instream, outstream): # Copy one stream to another sent = 0 @@ -112,11 +162,19 @@ passphrase = '%s\n' % passphrase passphrase = passphrase.encode(encoding) stream.write(passphrase) - logger.debug("Wrote passphrase: %r", passphrase) + logger.debug('Wrote passphrase')

def _is_sequence(instance): return isinstance(instance, (list, tuple, set, frozenset))

+def _make_memory_stream(s): + try: + from io import BytesIO + rv = BytesIO(s) + except ImportError: + rv = StringIO(s) + return rv + def _make_binary_stream(s, encoding): if _py3k: if isinstance(s, str): @@ -124,12 +182,7 @@ else: if type(s) is not str: s = s.encode(encoding) - try: - from io import BytesIO - rv = BytesIO(s) - except ImportError: - rv = StringIO(s) - return rv + return _make_memory_stream(s)

class Verify(object): "Handle status messages for --verify" @@ -175,7 +228,8 @@ "PLAINTEXT_LENGTH", "POLICY_URL", "DECRYPTION_INFO", "DECRYPTION_OKAY", "INV_SGNR", "FILE_START", "FILE_ERROR", "FILE_DONE", "PKA_TRUST_GOOD", "PKA_TRUST_BAD", "BADMDC", - "GOODMDC", "NO_SGNR", "NOTATION_NAME", "NOTATION_DATA"): + "GOODMDC", "NO_SGNR", "NOTATION_NAME", "NOTATION_DATA", + "PROGRESS"): pass elif key == "BADSIG": self.valid = False @@ -230,6 +284,10 @@ else: self.key_status = 'signing key was revoked' self.status = self.key_status + elif key == "UNEXPECTED": + self.valid = False + self.key_id = value + self.status = 'unexpected data' else: raise ValueError("Unknown status message: %r" % key)

@@ -298,8 +356,8 @@ 'problem': reason, 'text': self.problem_reason[reason]}) elif key == "IMPORT_RES": import_res = value.split() - for i in range(len(self.counts)): - setattr(self, self.counts[i], int(import_res[i])) + for i, count in enumerate(self.counts): + setattr(self, count, int(import_res[i])) elif key == "KEYEXPIRED": self.results.append({'fingerprint': None, 'problem': '0', 'text': 'Key expired'}) @@ -326,7 +384,53 @@ r'\0': '\0', }

-class ListKeys(list): +class SendResult(object): + def __init__(self, gpg): + self.gpg = gpg + + def handle_status(self, key, value): + logger.debug('SendResult: %s: %s', key, value) + +class SearchKeys(list): + ''' Handle status messages for --search-keys. + + Handle pub and uid (relating the latter to the former). + + Don't care about the rest + ''' + + UID_INDEX = 1 + FIELDS = 'type keyid algo length date expires'.split() + + def __init__(self, gpg): + self.gpg = gpg + self.curkey = None + self.fingerprints = [] + self.uids = [] + + def get_fields(self, args): + result = {} + for i, var in enumerate(self.FIELDS): + result[var] = args[i] + result['uids'] = [] + return result + + def pub(self, args): + self.curkey = curkey = self.get_fields(args) + self.append(curkey) + + def uid(self, args): + uid = args[self.UID_INDEX] + uid = ESCAPE_PATTERN.sub(lambda m: chr(int(m.group(1), 16)), uid) + for k, v in BASIC_ESCAPES.items(): + uid = uid.replace(k, v) + self.curkey['uids'].append(uid) + self.uids.append(uid) + + def handle_status(self, key, value): + pass + +class ListKeys(SearchKeys): ''' Handle status messages for --list-keys.

Handle pub and uid (relating the latter to the former). @@ -343,25 +447,17 @@ grp = reserved for gpgsm rvk = revocation key ''' - def __init__(self, gpg): - self.gpg = gpg - self.curkey = None - self.fingerprints = [] - self.uids = [] + + UID_INDEX = 9 + FIELDS = 'type trust length algo keyid date expires dummy ownertrust uid'.split()

def key(self, args): - vars = (""" - type trust length algo keyid date expires dummy ownertrust uid - """).split() - self.curkey = {} - for i in range(len(vars)): - self.curkey[vars[i]] = args[i] - self.curkey['uids'] = [] - if self.curkey['uid']: - self.curkey['uids'].append(self.curkey['uid']) - del self.curkey['uid'] - self.curkey['subkeys'] = [] - self.append(self.curkey) + self.curkey = curkey = self.get_fields(args) + if curkey['uid']: + curkey['uids'].append(curkey['uid']) + del curkey['uid'] + curkey['subkeys'] = [] + self.append(curkey)

pub = sec = key

@@ -369,22 +465,34 @@ self.curkey['fingerprint'] = args[9] self.fingerprints.append(args[9])

- def uid(self, args): - uid = args[9] - uid = ESCAPE_PATTERN.sub(lambda m: chr(int(m.group(1), 16)), uid) - for k, v in BASIC_ESCAPES.items(): - uid = uid.replace(k, v) - self.curkey['uids'].append(uid) - self.uids.append(uid) - def sub(self, args): subkey = [args[4], args[11]] self.curkey['subkeys'].append(subkey)

- def handle_status(self, key, value): - pass

-class Crypt(Verify): +class ScanKeys(ListKeys): + ''' Handle status messages for --with-fingerprint.''' + + def sub(self, args): + # --with-fingerprint --with-colons somehow outputs fewer colons, + # use the last value args[-1] instead of args[11] + subkey = [args[4], args[-1]] + self.curkey['subkeys'].append(subkey) + +class TextHandler(object): + def _as_text(self): + return self.data.decode(self.gpg.encoding, self.gpg.decode_errors) + + if _py3k: + __str__ = _as_text + else: + __unicode__ = _as_text + + def __str__(self): + return self.data + + +class Crypt(Verify, TextHandler): "Handle status messages for --encrypt and --decrypt" def __init__(self, gpg): Verify.__init__(self, gpg) @@ -398,19 +506,16 @@

__bool__ = __nonzero__

- def __str__(self): - return self.data.decode(self.gpg.encoding, self.gpg.decode_errors) - def handle_status(self, key, value): if key in ("ENC_TO", "USERID_HINT", "GOODMDC", "END_DECRYPTION", - "BEGIN_SIGNING", "NO_SECKEY", "ERROR", "NODATA", + "BEGIN_SIGNING", "NO_SECKEY", "ERROR", "NODATA", "PROGRESS", "CARDCTRL", "BADMDC", "SC_OP_FAILURE", "SC_OP_SUCCESS"): # in the case of ERROR, this is because a more specific error # message will have come first pass