summaryrefslogtreecommitdiff
path: root/plugins/gpg/gnupg.py
diff options
context:
space:
mode:
Diffstat (limited to 'plugins/gpg/gnupg.py')
-rw-r--r--plugins/gpg/gnupg.py973
1 files changed, 973 insertions, 0 deletions
diff --git a/plugins/gpg/gnupg.py b/plugins/gpg/gnupg.py
new file mode 100644
index 00000000..e3bed526
--- /dev/null
+++ b/plugins/gpg/gnupg.py
@@ -0,0 +1,973 @@
+""" A wrapper for the 'gpg' command::
+
+Portions of this module are derived from A.M. Kuchling's well-designed
+GPG.py, using Richard Jones' updated version 1.3, which can be found
+in the pycrypto CVS repository on Sourceforge:
+
+http://pycrypto.cvs.sourceforge.net/viewvc/pycrypto/gpg/GPG.py
+
+This module is *not* forward-compatible with amk's; some of the
+old interface has changed. For instance, since I've added decrypt
+functionality, I elected to initialize with a 'gnupghome' argument
+instead of 'keyring', so that gpg can find both the public and secret
+keyrings. I've also altered some of the returned objects in order for
+the caller to not have to know as much about the internals of the
+result classes.
+
+While the rest of ISconf is released under the GPL, I am releasing
+this single file under the same terms that A.M. Kuchling used for
+pycrypto.
+
+Steve Traugott, stevegt@terraluna.org
+Thu Jun 23 21:27:20 PDT 2005
+
+This version of the module has been modified from Steve Traugott's version
+(see http://trac.t7a.org/isconf/browser/trunk/lib/python/isconf/GPG.py) by
+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.
+
+A unittest harness (test_gnupg.py) has also been added.
+"""
+import locale
+
+__author__ = "Vinay Sajip"
+__date__ = "$25-Jan-2011 11:40:48$"
+
+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 locale
+import logging
+import os
+import socket
+from subprocess import Popen
+from subprocess import PIPE
+import sys
+import threading
+
+try:
+ import logging.NullHandler as NullHandler
+except ImportError:
+ class NullHandler(logging.Handler):
+ def handle(self, record):
+ pass
+try:
+ unicode
+ _py3k = False
+except NameError:
+ _py3k = True
+
+logger = logging.getLogger(__name__)
+if not logger.handlers:
+ logger.addHandler(NullHandler())
+
+def _copy_data(instream, outstream):
+ # Copy one stream to another
+ sent = 0
+ if hasattr(sys.stdin, 'encoding'):
+ enc = sys.stdin.encoding
+ else:
+ enc = 'ascii'
+ while True:
+ data = instream.read(1024)
+ if len(data) == 0:
+ break
+ sent += len(data)
+ logger.debug("sending chunk (%d): %r", sent, data[:256])
+ try:
+ outstream.write(data)
+ except UnicodeError:
+ outstream.write(data.encode(enc))
+ except:
+ # Can sometimes get 'broken pipe' errors even when the data has all
+ # been sent
+ logger.exception('Error sending data')
+ break
+ try:
+ outstream.close()
+ except IOError:
+ logger.warning('Exception occurred while closing: ignored', exc_info=1)
+ logger.debug("closed output, %d bytes sent", sent)
+
+def _threaded_copy_data(instream, outstream):
+ wr = threading.Thread(target=_copy_data, args=(instream, outstream))
+ wr.setDaemon(True)
+ logger.debug('data copier: %r, %r, %r', wr, instream, outstream)
+ wr.start()
+ return wr
+
+def _write_passphrase(stream, passphrase, encoding):
+ passphrase = '%s\n' % passphrase
+ passphrase = passphrase.encode(encoding)
+ stream.write(passphrase)
+ logger.debug("Passphrase written")
+
+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:
+ if isinstance(s, str):
+ s = s.encode(encoding)
+ else:
+ if type(s) is not str:
+ s = s.encode(encoding)
+ from io import BytesIO
+ rv = BytesIO(s)
+ except ImportError:
+ 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
+ >>> shutil.rmtree("keys")
+ >>> gpg = GPG(gnupghome="keys")
+ >>> input = gpg.gen_key_input()
+ >>> result = gpg.gen_key(input)
+ >>> print1 = result.fingerprint
+ >>> result = gpg.gen_key(input)
+ >>> print2 = result.fingerprint
+ >>> pubkey1 = gpg.export_keys(print1)
+ >>> seckey1 = gpg.export_keys(print1,secret=True)
+ >>> seckeys = gpg.list_keys(secret=True)
+ >>> pubkeys = gpg.list_keys()
+ >>> assert print1 in seckeys.fingerprints
+ >>> assert print1 in pubkeys.fingerprints
+ >>> str(gpg.delete_keys(print1))
+ 'Must delete secret key first'
+ >>> str(gpg.delete_keys(print1,secret=True))
+ 'ok'
+ >>> str(gpg.delete_keys(print1))
+ 'ok'
+ >>> str(gpg.delete_keys("nosuchkey"))
+ 'No such key'
+ >>> seckeys = gpg.list_keys(secret=True)
+ >>> pubkeys = gpg.list_keys()
+ >>> assert not print1 in seckeys.fingerprints
+ >>> assert not print1 in pubkeys.fingerprints
+ >>> result = gpg.import_keys('foo')
+ >>> assert not result
+ >>> result = gpg.import_keys(pubkey1)
+ >>> pubkeys = gpg.list_keys()
+ >>> seckeys = gpg.list_keys(secret=True)
+ >>> assert not print1 in seckeys.fingerprints
+ >>> assert print1 in pubkeys.fingerprints
+ >>> result = gpg.import_keys(seckey1)
+ >>> assert result
+ >>> seckeys = gpg.list_keys(secret=True)
+ >>> pubkeys = gpg.list_keys()
+ >>> assert print1 in seckeys.fingerprints
+ >>> assert print1 in pubkeys.fingerprints
+ >>> assert print2 in pubkeys.fingerprints
+
+ """
+ result = ImportResult()
+ logger.debug('import_keys: %r', key_data[:256])
+ data = _make_binary_stream(key_data, self.encoding)
+ self._handle_io(['--import'], data, result, binary=True)
+ logger.debug('import_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()
+ p = self._open_subprocess(args)
+ self._collect_output(p, result, stdin=p.stdin)
+ return result
+
+ def export_keys(self, keyids, secret=False):
+ "export the indicated keys. 'keyid' is anything gpg accepts"
+ which=''
+ if secret:
+ which='-secret-key'
+ if _is_sequence(keyids):
+ keyids = ' '.join(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
+ self._collect_output(p, result, stdin=p.stdin)
+ logger.debug('export_keys result: %r', result.data)
+ return result.data.decode(self.encoding, 'replace')
+
+ def list_keys(self, secret=False):
+ """ list the keys currently in the keyring
+
+ >>> import shutil
+ >>> shutil.rmtree("keys")
+ >>> gpg = GPG(gnupghome="keys")
+ >>> input = gpg.gen_key_input()
+ >>> result = gpg.gen_key(input)
+ >>> print1 = result.fingerprint
+ >>> result = gpg.gen_key(input)
+ >>> print2 = result.fingerprint
+ >>> pubkeys = gpg.list_keys()
+ >>> assert print1 in pubkeys.fingerprints
+ >>> assert print2 in pubkeys.fingerprints
+
+ """
+
+ which='keys'
+ if secret:
+ which='secret-keys'
+ args = "--list-%s --fixed-list-mode --fingerprint --with-colons" % (which,)
+ args = [args]
+ p = self._open_subprocess(args)
+
+ # there might be some status thingumy here I should handle... (amk)
+ # ...nope, unless you care about expired sigs or keys (stevegt)
+
+ # Get the response information
+ result = ListKeys()
+ self._collect_output(p, result, stdin=p.stdin)
+ lines = result.data.decode(self.encoding, 'replace').splitlines()
+ valid_keywords = 'pub uid sec fpr sub'.split()
+ for line in lines:
+ if self.verbose:
+ print(line)
+ logger.debug("line: %r", line.rstrip())
+ if not line:
+ break
+ L = line.strip().split(':')
+ if not L:
+ continue
+ keyword = L[0]
+ if keyword in valid_keywords:
+ getattr(result, keyword)(L)
+ return result
+
+ def gen_key(self, input):
+ """Generate a key; you might use gen_key_input() to create the
+ control input.
+
+ >>> gpg = GPG(gnupghome="keys")
+ >>> input = gpg.gen_key_input()
+ >>> result = gpg.gen_key(input)
+ >>> assert result
+ >>> result = gpg.gen_key('foo')
+ >>> assert not result
+
+ """
+ args = ["--gen-key --batch"]
+ result = GenKey()
+ f = _make_file(input)
+ self._handle_io(args, f, result)
+ f.close()
+ return result
+
+ def gen_key_input(self, **kwargs):
+ """
+ Generate --gen-key input per gpg doc/DETAILS
+ """
+ parms = {}
+ for key, val in list(kwargs.items()):
+ key = key.replace('_','-').title()
+ parms[key] = val
+ parms.setdefault('Key-Type','RSA')
+ parms.setdefault('Key-Length',1024)
+ parms.setdefault('Name-Real', "Autogenerated Key")
+ parms.setdefault('Name-Comment', "Generated by gnupg.py")
+ try:
+ logname = os.environ['LOGNAME']
+ except KeyError:
+ logname = os.environ['USERNAME']
+ hostname = socket.gethostname()
+ parms.setdefault('Name-Email', "%s@%s" % (logname.replace(' ', '_'),
+ hostname))
+ out = "Key-Type: %s\n" % parms.pop('Key-Type')
+ for key, val in list(parms.items()):
+ out += "%s: %s\n" % (key, val)
+ out += "%commit\n"
+ return out
+
+ # Key-Type: RSA
+ # Key-Length: 1024
+ # Name-Real: ISdlink Server on %s
+ # Name-Comment: Created by %s
+ # Name-Email: isdlink@%s
+ # Expire-Date: 0
+ # %commit
+ #
+ #
+ # Key-Type: DSA
+ # Key-Length: 1024
+ # Subkey-Type: ELG-E
+ # Subkey-Length: 1024
+ # Name-Real: Joe Tester
+ # Name-Comment: with stupid passphrase
+ # Name-Email: joe@foo.bar
+ # Expire-Date: 0
+ # Passphrase: abc
+ # %pubring foo.pub
+ # %secring foo.sec
+ # %commit
+
+ #
+ # ENCRYPTION
+ #
+ def encrypt_file(self, file, recipients, sign=None,
+ always_trust=False, passphrase=None,
+ armor=True, output=None):
+ "Encrypt the message read from the file-like object 'file'"
+ args = ['--encrypt']
+ 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)
+ if sign:
+ args.append("--sign --default-key %s" % sign)
+ if always_trust:
+ args.append("--always-trust")
+ result = Crypt(self.encoding)
+ self._handle_io(args, file, result, passphrase=passphrase, binary=True)
+ logger.debug('encrypt result: %r', result.data)
+ return result
+
+ def encrypt(self, data, recipients, **kwargs):
+ """Encrypt the message contained in the string 'data'
+
+ >>> import shutil
+ >>> if os.path.exists("keys"):
+ ... shutil.rmtree("keys")
+ >>> gpg = GPG(gnupghome="keys")
+ >>> input = gpg.gen_key_input(passphrase='foo')
+ >>> result = gpg.gen_key(input)
+ >>> print1 = result.fingerprint
+ >>> input = gpg.gen_key_input()
+ >>> result = gpg.gen_key(input)
+ >>> print2 = result.fingerprint
+ >>> result = gpg.encrypt("hello",print2)
+ >>> message = str(result)
+ >>> assert message != 'hello'
+ >>> result = gpg.decrypt(message)
+ >>> assert result
+ >>> str(result)
+ 'hello'
+ >>> result = gpg.encrypt("hello again",print1)
+ >>> message = str(result)
+ >>> result = gpg.decrypt(message)
+ >>> result.status
+ 'need passphrase'
+ >>> result = gpg.decrypt(message,passphrase='bar')
+ >>> result.status
+ 'decryption failed'
+ >>> assert not result
+ >>> result = gpg.decrypt(message,passphrase='foo')
+ >>> result.status
+ 'decryption ok'
+ >>> str(result)
+ 'hello again'
+ >>> result = gpg.encrypt("signed hello",print2,sign=print1)
+ >>> result.status
+ 'need passphrase'
+ >>> result = gpg.encrypt("signed hello",print2,sign=print1,passphrase='foo')
+ >>> result.status
+ 'encryption ok'
+ >>> message = str(result)
+ >>> result = gpg.decrypt(message)
+ >>> result.status
+ 'decryption ok'
+ >>> assert result.fingerprint == print1
+
+ """
+ data = _make_binary_stream(data, self.encoding)
+ result = self.encrypt_file(data, recipients, **kwargs)
+ data.close()
+ return result
+
+ def decrypt(self, message, **kwargs):
+ data = _make_binary_stream(message, self.encoding)
+ result = self.decrypt_file(data, **kwargs)
+ data.close()
+ return result
+
+ def decrypt_file(self, file, always_trust=False, passphrase=None,
+ output=None):
+ args = ["--decrypt"]
+ 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 always_trust:
+ args.append("--always-trust")
+ result = Crypt(self.encoding)
+ 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)