diff options
Diffstat (limited to 'plugins/gpg')
| -rw-r--r-- | plugins/gpg/gnupg.py | 857 |
1 files changed, 469 insertions, 388 deletions
diff --git a/plugins/gpg/gnupg.py b/plugins/gpg/gnupg.py index e9e45711..9109451a 100644 --- a/plugins/gpg/gnupg.py +++ b/plugins/gpg/gnupg.py @@ -27,25 +27,22 @@ Vinay Sajip to make use of the subprocess module (Steve's version uses os.fork() 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.1" __author__ = "Vinay Sajip" -__date__ = "$25-Jan-2011 11:40:48$" +__date__ = "$01-Sep-2012 20:02:51$" 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 @@ def _write_passphrase(stream, passphrase, encoding): 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,51 +126,437 @@ def _make_binary_stream(s, encoding): rv = StringIO(s) return rv +class Verify(object): + "Handle status messages for --verify" + + TRUST_UNDEFINED = 0 + TRUST_NEVER = 1 + TRUST_MARGINAL = 2 + TRUST_FULLY = 3 + TRUST_ULTIMATE = 4 + + TRUST_LEVELS = { + "TRUST_UNDEFINED" : TRUST_UNDEFINED, + "TRUST_NEVER" : TRUST_NEVER, + "TRUST_MARGINAL" : TRUST_MARGINAL, + "TRUST_FULLY" : TRUST_FULLY, + "TRUST_ULTIMATE" : TRUST_ULTIMATE, + } + + def __init__(self, gpg): + self.gpg = gpg + self.valid = False + self.fingerprint = self.creation_date = self.timestamp = None + self.signature_id = self.key_id = None + self.username = None + self.status = None + self.pubkey_fingerprint = None + self.expire_timestamp = None + self.sig_timestamp = None + self.trust_text = None + self.trust_level = None + + def __nonzero__(self): + return self.valid + + __bool__ = __nonzero__ + + def handle_status(self, key, value): + if key in self.TRUST_LEVELS: + self.trust_text = key + 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"): + pass + elif key == "BADSIG": + self.valid = False + self.status = 'signature bad' + self.key_id, self.username = value.split(None, 1) + elif key == "GOODSIG": + self.valid = True + self.status = 'signature good' + self.key_id, self.username = value.split(None, 1) + elif key == "VALIDSIG": + (self.fingerprint, + self.creation_date, + self.sig_timestamp, + self.expire_timestamp) = value.split()[:4] + # may be different if signature is made with a subkey + self.pubkey_fingerprint = value.split()[-1] + self.status = 'signature valid' + elif key == "SIG_ID": + (self.signature_id, + self.creation_date, self.timestamp) = value.split() + elif key == "ERRSIG": + self.valid = False + (self.key_id, + algo, hash_algo, + cls, + self.timestamp) = value.split()[:5] + self.status = 'signature error' + elif key == "DECRYPTION_FAILED": + self.valid = False + self.key_id = value + self.status = 'decryption failed' + elif key == "NO_PUBKEY": + self.valid = False + self.key_id = value + self.status = 'no public key' + elif key in ("KEYEXPIRED", "SIGEXPIRED"): + # these are useless in verify, since they are spit out for any + # pub/subkeys on the key, not just the one doing the signing. + # if we want to check for signatures with expired key, + # the relevant flag is EXPKEYSIG. + pass + elif key in ("EXPKEYSIG", "REVKEYSIG"): + # signed with expired or revoked key + self.valid = False + self.key_id = value.split()[0] + self.status = (('%s %s') % (key[:3], key[3:])).lower() + else: + raise ValueError("Unknown status message: %r" % key) + +class ImportResult(object): + "Handle status messages for --import" + + counts = '''count no_user_id imported imported_rsa unchanged + n_uids n_subk n_sigs n_revoc sec_read sec_imported + sec_dups not_imported'''.split() + def __init__(self, gpg): + self.gpg = gpg + self.imported = [] + self.results = [] + self.fingerprints = [] + for result in self.counts: + setattr(self, result, None) + + def __nonzero__(self): + if self.not_imported: return False + if not self.fingerprints: return False + return True + + __bool__ = __nonzero__ + + ok_reason = { + '0': 'Not actually changed', + '1': 'Entirely new key', + '2': 'New user IDs', + '4': 'New signatures', + '8': 'New subkeys', + '16': 'Contains private key', + } + + problem_reason = { + '0': 'No specific reason given', + '1': 'Invalid Certificate', + '2': 'Issuer Certificate missing', + '3': 'Certificate Chain too long', + '4': 'Error storing certificate', + } + + def handle_status(self, key, value): + if key == "IMPORTED": + # this duplicates info we already see in import_ok & import_problem + pass + elif key == "NODATA": + self.results.append({'fingerprint': None, + 'problem': '0', 'text': 'No valid data found'}) + elif key == "IMPORT_OK": + reason, fingerprint = value.split() + reasons = [] + for code, text in list(self.ok_reason.items()): + if int(reason) | int(code) == int(reason): + reasons.append(text) + reasontext = '\n'.join(reasons) + "\n" + self.results.append({'fingerprint': fingerprint, + 'ok': reason, 'text': reasontext}) + self.fingerprints.append(fingerprint) + elif key == "IMPORT_PROBLEM": + try: + reason, fingerprint = value.split() + except: + reason = value + fingerprint = '<unknown>' + self.results.append({'fingerprint': fingerprint, + '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])) + elif key == "KEYEXPIRED": + self.results.append({'fingerprint': None, + 'problem': '0', 'text': 'Key expired'}) + elif key == "SIGEXPIRED": + self.results.append({'fingerprint': None, + 'problem': '0', 'text': 'Signature expired'}) + else: + raise ValueError("Unknown status message: %r" % key) + + def summary(self): + l = [] + l.append('%d imported' % self.imported) + if self.not_imported: + l.append('%d not imported' % self.not_imported) + return ', '.join(l) + +class ListKeys(list): + ''' Handle status messages for --list-keys. + + Handle pub and uid (relating the latter to the former). + + Don't care about (info from src/DETAILS): + + crt = X.509 certificate + crs = X.509 certificate and private key available + ssb = secret subkey (secondary key) + uat = user attribute (same as user id except for field 10). + sig = signature + rev = revocation signature + pkd = public key data (special field format, see below) + grp = reserved for gpgsm + rvk = revocation key + ''' + def __init__(self, gpg): + self.gpg = gpg + self.curkey = None + self.fingerprints = [] + self.uids = [] + + 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) + + pub = sec = key + + def fpr(self, args): + self.curkey['fingerprint'] = args[9] + self.fingerprints.append(args[9]) + + def uid(self, args): + self.curkey['uids'].append(args[9]) + self.uids.append(args[9]) + + def sub(self, args): + subkey = [args[4], args[11]] + self.curkey['subkeys'].append(subkey) + + def handle_status(self, key, value): + pass + +class Crypt(Verify): + "Handle status messages for --encrypt and --decrypt" + def __init__(self, gpg): + Verify.__init__(self, gpg) + self.data = '' + self.ok = False + self.status = '' + + def __nonzero__(self): + if self.ok: return True + return False + + __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", + "CARDCTRL"): + # in the case of ERROR, this is because a more specific error + # message will have come first + pass + elif key in ("NEED_PASSPHRASE", "BAD_PASSPHRASE", "GOOD_PASSPHRASE", + "MISSING_PASSPHRASE", "DECRYPTION_FAILED", + "KEY_NOT_CREATED"): + self.status = key.replace("_", " ").lower() + elif key == "NEED_PASSPHRASE_SYM": + self.status = 'need symmetric passphrase' + elif key == "BEGIN_DECRYPTION": + self.status = 'decryption incomplete' + elif key == "BEGIN_ENCRYPTION": + self.status = 'encryption incomplete' + elif key == "DECRYPTION_OKAY": + self.status = 'decryption ok' + self.ok = True + elif key == "END_ENCRYPTION": + self.status = 'encryption ok' + self.ok = True + elif key == "INV_RECP": + self.status = 'invalid recipient' + elif key == "KEYEXPIRED": + self.status = 'key expired' + elif key == "SIG_CREATED": + self.status = 'sig created' + elif key == "SIGEXPIRED": + self.status = 'sig expired' + else: + Verify.handle_status(self, key, value) + +class GenKey(object): + "Handle status messages for --gen-key" + def __init__(self, gpg): + self.gpg = gpg + self.type = None + self.fingerprint = None + + def __nonzero__(self): + if self.fingerprint: return True + return False + + __bool__ = __nonzero__ + + def __str__(self): + return self.fingerprint or '' + + def handle_status(self, key, value): + if key in ("PROGRESS", "GOOD_PASSPHRASE", "NODATA", "KEY_NOT_CREATED"): + pass + elif key == "KEY_CREATED": + (self.type,self.fingerprint) = value.split() + else: + raise ValueError("Unknown status message: %r" % key) + +class DeleteResult(object): + "Handle status messages for --delete-key and --delete-secret-key" + def __init__(self, gpg): + self.gpg = gpg + self.status = 'ok' + + def __str__(self): + return self.status + + problem_reason = { + '1': 'No such key', + '2': 'Must delete secret key first', + '3': 'Ambigious specification', + } + + def handle_status(self, key, value): + if key == "DELETE_PROBLEM": + self.status = self.problem_reason.get(value, + "Unknown error: %r" % value) + else: + raise ValueError("Unknown status message: %r" % key) + +class Sign(object): + "Handle status messages for --sign" + def __init__(self, gpg): + self.gpg = gpg + self.type = None + self.fingerprint = None + + def __nonzero__(self): + return self.fingerprint is not None + + __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 ("USERID_HINT", "NEED_PASSPHRASE", "BAD_PASSPHRASE", + "GOOD_PASSPHRASE", "BEGIN_SIGNING", "CARDCTRL", "INV_SGNR"): + pass + elif key == "SIG_CREATED": + (self.type, + algo, hashalgo, cls, + self.timestamp, self.fingerprint + ) = value.split() + else: + raise ValueError("Unknown status message: %r" % key) + + class GPG(object): + + decode_errors = 'strict' + + result_map = { + 'crypt': Crypt, + 'delete': DeleteResult, + 'generate': GenKey, + 'import': ImportResult, + 'list': ListKeys, + 'sign': Sign, + 'verify': Verify, + } + "Encapsulate access to the gpg executable" - def __init__(self, gpgbinary='gpg', gnupghome=None, verbose=False, use_agent=False): + def __init__(self, gpgbinary='gpg', gnupghome=None, verbose=False, + use_agent=False, keyring=None, options=None): """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. + keyring -- name of alternative keyring file to use. If specified, + the default keyring is not used. + options =-- a list of additional options to pass to the GPG binary. """ self.gpgbinary = gpgbinary self.gnupghome = gnupghome + self.keyring = keyring self.verbose = verbose self.use_agent = use_agent + if isinstance(options, str): + options = [options] + self.options = options 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 + result = self.result_map['verify'](self) # 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. + def make_args(self, args, passphrase): + """ + Make a list of command line elements for GPG. The value of ``args`` + will be appended. The ``passphrase`` argument needs to be True if + a passphrase will be sent to GPG, else False. + """ cmd = [self.gpgbinary, '--status-fd 2 --no-tty'] if self.gnupghome: cmd.append('--homedir "%s" ' % self.gnupghome) + if self.keyring: + cmd.append('--no-default-keyring --keyring "%s" ' % self.keyring) if passphrase: cmd.append('--batch --passphrase-fd 0') if self.use_agent: cmd.append('--use-agent') + if self.options: + cmd.extend(self.options) cmd.extend(args) - cmd = ' '.join(cmd) + return cmd + + 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 = ' '.join(self.make_args(args, passphrase)) 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 + # Internal method: reads all the stderr 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, @@ -204,14 +564,13 @@ class GPG(object): lines = [] while True: line = stream.readline() - if not isinstance(line, str): - line = line.decode('utf-8') + if len(line) == 0: + break lines.append(line) + line = line.rstrip() if self.verbose: print(line) - logger.debug("%s", line.rstrip()) - if line == "": break - line = line.rstrip() + logger.debug("%s", line) if line[0:9] == '[GNUPG:] ': # Chop off the prefix line = line[9:] @@ -246,13 +605,13 @@ class GPG(object): make sure it's joined before returning. If a stdin stream is given, close it before returning. """ - stderr = _wrap_output(process.stderr) + stderr = codecs.getreader(self.encoding)(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) + stdout = process.stdout dr = threading.Thread(target=self._read_data, args=(stdout, result)) dr.setDaemon(True) logger.debug('stdout reader: %r', dr) @@ -276,8 +635,8 @@ class GPG(object): # 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) + if not binary: + stdin = codecs.getwriter(self.encoding)(p.stdin) else: stdin = p.stdin if passphrase: @@ -311,11 +670,10 @@ class GPG(object): elif clearsign: args.append("--clearsign") if keyid: - args.append("--default-key %s" % keyid) - result = Sign(self.encoding) + args.append('--default-key "%s"' % keyid) + result = self.result_map['sign'](self) #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 @@ -351,7 +709,7 @@ class GPG(object): 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() + result = self.result_map['verify'](self) args = ['--verify'] if data_filename is None: self._handle_io(args, file, result, binary=True) @@ -365,7 +723,7 @@ class GPG(object): os.write(fd, s) os.close(fd) args.append(fn) - args.append(data_filename) + args.append('"%s"' % data_filename) try: p = self._open_subprocess(args) self._collect_output(p, result, stdin=p.stdin) @@ -422,7 +780,7 @@ class GPG(object): >>> assert print2 in pubkeys.fingerprints """ - result = ImportResult() + result = self.result_map['import'](self) logger.debug('import_keys: %r', key_data[:256]) data = _make_binary_stream(key_data, self.encoding) self._handle_io(['--import'], data, result, binary=True) @@ -430,14 +788,35 @@ class GPG(object): data.close() return result + def recv_keys(self, keyserver, *keyids): + """Import a key from a keyserver + + >>> import shutil + >>> shutil.rmtree("keys") + >>> gpg = GPG(gnupghome="keys") + >>> result = gpg.recv_keys('pgp.mit.edu', '3FF0DB166A7476EA') + >>> assert result + + """ + result = self.result_map['import'](self) + logger.debug('recv_keys: %r', keyids) + data = _make_binary_stream("", self.encoding) + #data = "" + args = ['--keyserver', keyserver, '--recv-keys'] + args.extend(keyids) + self._handle_io(args, data, result, binary=True) + logger.debug('recv_keys result: %r', result.__dict__) + data.close() + return result + def delete_keys(self, fingerprints, secret=False): which='key' if secret: which='secret-key' if _is_sequence(fingerprints): fingerprints = ' '.join(fingerprints) - args = ["--batch --delete-%s %s" % (which, fingerprints)] - result = DeleteResult() + args = ['--batch --delete-%s "%s"' % (which, fingerprints)] + result = self.result_map['delete'](self) p = self._open_subprocess(args) self._collect_output(p, result, stdin=p.stdin) return result @@ -448,16 +827,16 @@ class GPG(object): if secret: which='-secret-key' if _is_sequence(keyids): - keyids = ' '.join(keyids) + keyids = ' '.join(['"%s"' % k for k in keyids]) args = ["--armor --export%s %s" % (which, keyids)] p = self._open_subprocess(args) # gpg --export produces no status-fd output; stdout will be # empty in case of failure #stdout, stderr = p.communicate() - result = DeleteResult() # any result will do + result = self.result_map['delete'](self) # any result will do self._collect_output(p, result, stdin=p.stdin) logger.debug('export_keys result: %r', result.data) - return result.data.decode(self.encoding, 'replace') + return result.data.decode(self.encoding, self.decode_errors) def list_keys(self, secret=False): """ list the keys currently in the keyring @@ -487,9 +866,10 @@ class GPG(object): # ...nope, unless you care about expired sigs or keys (stevegt) # Get the response information - result = ListKeys() + result = self.result_map['list'](self) self._collect_output(p, result, stdin=p.stdin) - lines = result.data.decode(self.encoding, 'replace').splitlines() + lines = result.data.decode(self.encoding, + self.decode_errors).splitlines() valid_keywords = 'pub uid sec fpr sub'.split() for line in lines: if self.verbose: @@ -518,9 +898,9 @@ class GPG(object): """ args = ["--gen-key --batch"] - result = GenKey() - f = _make_file(input) - self._handle_io(args, f, result) + result = self.result_map['generate'](self) + f = _make_binary_stream(input, self.encoding) + self._handle_io(args, f, result, binary=True) f.close() return result @@ -576,24 +956,28 @@ class GPG(object): # def encrypt_file(self, file, recipients, sign=None, always_trust=False, passphrase=None, - armor=True, output=None): + armor=True, output=None, symmetric=False): "Encrypt the message read from the file-like object 'file'" args = ['--encrypt'] + if symmetric: + args = ['--symmetric'] + else: + args = ['--encrypt'] + if not _is_sequence(recipients): + recipients = (recipients,) + for recipient in recipients: + args.append('--recipient "%s"' % recipient) if armor: # create ascii-armored output - set to False for binary output args.append('--armor') if output: # write the output to a file with the specified name if os.path.exists(output): os.remove(output) # to avoid overwrite confirmation message - args.append('--output %s' % output) - if not _is_sequence(recipients): - recipients = (recipients,) - for recipient in recipients: - args.append('--recipient %s' % recipient) + args.append('--output "%s"' % output) if sign: - args.append("--sign --default-key %s" % sign) + args.append('--sign --default-key "%s"' % sign) if always_trust: args.append("--always-trust") - result = Crypt(self.encoding) + result = self.result_map['crypt'](self) self._handle_io(args, file, result, passphrase=passphrase, binary=True) logger.debug('encrypt result: %r', result.data) return result @@ -621,27 +1005,27 @@ class GPG(object): >>> result = gpg.encrypt("hello again",print1) >>> message = str(result) >>> result = gpg.decrypt(message) - >>> result.status - 'need passphrase' + >>> result.status == 'need passphrase' + True >>> result = gpg.decrypt(message,passphrase='bar') - >>> result.status - 'decryption failed' + >>> result.status in ('decryption failed', 'bad passphrase') + True >>> assert not result >>> result = gpg.decrypt(message,passphrase='foo') - >>> result.status - 'decryption ok' + >>> result.status == 'decryption ok' + True >>> str(result) 'hello again' >>> result = gpg.encrypt("signed hello",print2,sign=print1) - >>> result.status - 'need passphrase' + >>> result.status == 'need passphrase' + True >>> result = gpg.encrypt("signed hello",print2,sign=print1,passphrase='foo') - >>> result.status - 'encryption ok' + >>> result.status == 'encryption ok' + True >>> message = str(result) >>> result = gpg.decrypt(message) - >>> result.status - 'decryption ok' + >>> result.status == 'decryption ok' + True >>> assert result.fingerprint == print1 """ @@ -662,314 +1046,11 @@ class GPG(object): if output: # write the output to a file with the specified name if os.path.exists(output): os.remove(output) # to avoid overwrite confirmation message - args.append('--output %s' % output) + args.append('--output "%s"' % output) if always_trust: args.append("--always-trust") - result = Crypt(self.encoding) + result = self.result_map['crypt'](self) self._handle_io(args, file, result, passphrase, binary=True) logger.debug('decrypt result: %r', result.data) return result -class Verify(object): - "Handle status messages for --verify" - - def __init__(self): - self.valid = False - self.fingerprint = self.creation_date = self.timestamp = None - self.signature_id = self.key_id = None - self.username = None - - def __nonzero__(self): - return self.valid - - __bool__ = __nonzero__ - - def handle_status(self, key, value): - if key in ("TRUST_UNDEFINED", "TRUST_NEVER", "TRUST_MARGINAL", - "TRUST_FULLY", "TRUST_ULTIMATE", "RSA_OR_IDEA"): - pass - elif key in ("PLAINTEXT", "PLAINTEXT_LENGTH"): - pass - elif key == "IMPORT_RES": - # If auto-key-retrieve option is enabled, this can happen - pass - elif key == "BADSIG": - self.valid = False - self.key_id, self.username = value.split(None, 1) - elif key == "GOODSIG": - self.valid = True - self.key_id, self.username = value.split(None, 1) - elif key == "VALIDSIG": - (self.fingerprint, - self.creation_date, - self.sig_timestamp, - self.expire_timestamp) = value.split()[:4] - elif key == "SIG_ID": - (self.signature_id, - self.creation_date, self.timestamp) = value.split() - elif key == "ERRSIG": - self.valid = False - (self.key_id, - algo, hash_algo, - cls, - self.timestamp) = value.split()[:5] - elif key == "NO_PUBKEY": - self.valid = False - self.key_id = value - else: - raise ValueError("Unknown status message: %r" % key) - -class ImportResult(object): - "Handle status messages for --import" - - counts = '''count no_user_id imported imported_rsa unchanged - n_uids n_subk n_sigs n_revoc sec_read sec_imported - sec_dups not_imported'''.split() - def __init__(self): - self.imported = [] - self.results = [] - self.fingerprints = [] - for result in self.counts: - setattr(self, result, None) - - def __nonzero__(self): - if self.not_imported: return False - if not self.fingerprints: return False - return True - - __bool__ = __nonzero__ - - ok_reason = { - '0': 'Not actually changed', - '1': 'Entirely new key', - '2': 'New user IDs', - '4': 'New signatures', - '8': 'New subkeys', - '16': 'Contains private key', - } - - problem_reason = { - '0': 'No specific reason given', - '1': 'Invalid Certificate', - '2': 'Issuer Certificate missing', - '3': 'Certificate Chain too long', - '4': 'Error storing certificate', - } - - def handle_status(self, key, value): - if key == "IMPORTED": - # this duplicates info we already see in import_ok & import_problem - pass - elif key == "NODATA": - self.results.append({'fingerprint': None, - 'problem': '0', 'text': 'No valid data found'}) - elif key == "IMPORT_OK": - reason, fingerprint = value.split() - reasons = [] - for code, text in list(self.ok_reason.items()): - if int(reason) | int(code) == int(reason): - reasons.append(text) - reasontext = '\n'.join(reasons) + "\n" - self.results.append({'fingerprint': fingerprint, - 'ok': reason, 'text': reasontext}) - self.fingerprints.append(fingerprint) - elif key == "IMPORT_PROBLEM": - try: - reason, fingerprint = value.split() - except: - reason = value - fingerprint = '<unknown>' - self.results.append({'fingerprint': fingerprint, - '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])) - elif key == "KEYEXPIRED": - self.results.append({'fingerprint': None, - 'problem': '0', 'text': 'Key expired'}) - elif key == "SIGEXPIRED": - self.results.append({'fingerprint': None, - 'problem': '0', 'text': 'Signature expired'}) - else: - raise ValueError("Unknown status message: %r" % key) - - def summary(self): - l = [] - l.append('%d imported'%self.imported) - if self.not_imported: - l.append('%d not imported'%self.not_imported) - return ', '.join(l) - -class ListKeys(list): - ''' Handle status messages for --list-keys. - - Handle pub and uid (relating the latter to the former). - - Don't care about (info from src/DETAILS): - - crt = X.509 certificate - crs = X.509 certificate and private key available - sub = subkey (secondary key) - ssb = secret subkey (secondary key) - uat = user attribute (same as user id except for field 10). - sig = signature - rev = revocation signature - pkd = public key data (special field format, see below) - grp = reserved for gpgsm - rvk = revocation key - ''' - def __init__(self): - self.curkey = None - self.fingerprints = [] - self.uids = [] - - 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) - - pub = sec = key - - def fpr(self, args): - self.curkey['fingerprint'] = args[9] - self.fingerprints.append(args[9]) - - def uid(self, args): - self.curkey['uids'].append(args[9]) - self.uids.append(args[9]) - - def sub(self, args): - subkey = [args[4],args[11]] - self.curkey['subkeys'].append(subkey) - - def handle_status(self, key, value): - pass - -class Crypt(Verify): - "Handle status messages for --encrypt and --decrypt" - def __init__(self, encoding): - Verify.__init__(self) - self.data = '' - self.ok = False - self.status = '' - self.encoding = encoding - - def __nonzero__(self): - if self.ok: return True - return False - - __bool__ = __nonzero__ - - def __str__(self): - return self.data.decode(self.encoding, 'replace') - - def handle_status(self, key, value): - if key in ("ENC_TO", "USERID_HINT", "GOODMDC", "END_DECRYPTION", - "BEGIN_SIGNING", "NO_SECKEY"): - pass - elif key in ("NEED_PASSPHRASE", "BAD_PASSPHRASE", "GOOD_PASSPHRASE", - "MISSING_PASSPHRASE", "DECRYPTION_FAILED"): - self.status = key.replace("_", " ").lower() - elif key == "NEED_PASSPHRASE_SYM": - self.status = 'need symmetric passphrase' - elif key == "BEGIN_DECRYPTION": - self.status = 'decryption incomplete' - elif key == "BEGIN_ENCRYPTION": - self.status = 'encryption incomplete' - elif key == "DECRYPTION_OKAY": - self.status = 'decryption ok' - self.ok = True - elif key == "END_ENCRYPTION": - self.status = 'encryption ok' - self.ok = True - elif key == "INV_RECP": - self.status = 'invalid recipient' - elif key == "KEYEXPIRED": - self.status = 'key expired' - elif key == "SIG_CREATED": - self.status = 'sig created' - elif key == "SIGEXPIRED": - self.status = 'sig expired' - else: - Verify.handle_status(self, key, value) - -class GenKey(object): - "Handle status messages for --gen-key" - def __init__(self): - self.type = None - self.fingerprint = None - - def __nonzero__(self): - if self.fingerprint: return True - return False - - __bool__ = __nonzero__ - - def __str__(self): - return self.fingerprint or '' - - def handle_status(self, key, value): - if key in ("PROGRESS", "GOOD_PASSPHRASE", "NODATA"): - pass - elif key == "KEY_CREATED": - (self.type,self.fingerprint) = value.split() - else: - raise ValueError("Unknown status message: %r" % key) - -class DeleteResult(object): - "Handle status messages for --delete-key and --delete-secret-key" - def __init__(self): - self.status = 'ok' - - def __str__(self): - return self.status - - problem_reason = { - '1': 'No such key', - '2': 'Must delete secret key first', - '3': 'Ambigious specification', - } - - def handle_status(self, key, value): - if key == "DELETE_PROBLEM": - self.status = self.problem_reason.get(value, - "Unknown error: %r" % value) - else: - raise ValueError("Unknown status message: %r" % key) - -class Sign(object): - "Handle status messages for --sign" - def __init__(self, encoding): - self.type = None - self.fingerprint = None - self.encoding = encoding - - def __nonzero__(self): - return self.fingerprint is not None - - __bool__ = __nonzero__ - - def __str__(self): - return self.data.decode(self.encoding, 'replace') - - def handle_status(self, key, value): - if key in ("USERID_HINT", "NEED_PASSPHRASE", "BAD_PASSPHRASE", - "GOOD_PASSPHRASE", "BEGIN_SIGNING"): - pass - elif key == "SIG_CREATED": - (self.type, - algo, hashalgo, cls, - self.timestamp, self.fingerprint - ) = value.split() - else: - raise ValueError("Unknown status message: %r" % key) |