You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@subversion.apache.org by ph...@apache.org on 2017/07/10 16:55:37 UTC
svn commit: r1801492 - /subversion/trunk/tools/dist/security/_gnupg.py
Author: philip
Date: Mon Jul 10 16:55:37 2017
New Revision: 1801492
URL: http://svn.apache.org/viewvc?rev=1801492&view=rev
Log:
* tools/dist/security/_gnupg.py: Update to the 0.4.1 tag from
https://bitbucket.org/vinay.sajip/python-gnupg to get better
support for gpg2. Add text of the LICENSE file from the tag
to the top of this file.
Modified:
subversion/trunk/tools/dist/security/_gnupg.py
Modified: subversion/trunk/tools/dist/security/_gnupg.py
URL: http://svn.apache.org/viewvc/subversion/trunk/tools/dist/security/_gnupg.py?rev=1801492&r1=1801491&r2=1801492&view=diff
==============================================================================
--- subversion/trunk/tools/dist/security/_gnupg.py (original)
+++ subversion/trunk/tools/dist/security/_gnupg.py Mon Jul 10 16:55:37 2017
@@ -1,3 +1,29 @@
+# Copyright (c) 2008-2014 by Vinay Sajip.
+# All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are met:
+#
+# * Redistributions of source code must retain the above copyright notice,
+# this list of conditions and the following disclaimer.
+# * Redistributions in binary form must reproduce the above copyright notice,
+# this list of conditions and the following disclaimer in the documentation
+# and/or other materials provided with the distribution.
+# * The name(s) of the copyright holder(s) may not be used to endorse or
+# promote products derived from this software without specific prior
+# written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDER(S) "AS IS" AND ANY EXPRESS OR
+# IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
+# MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO
+# EVENT SHALL THE COPYRIGHT HOLDER(S) BE LIABLE FOR ANY DIRECT, INDIRECT,
+# INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
+# PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+# LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE
+# OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF
+# ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
""" A wrapper for the 'gpg' command::
Portions of this module are derived from A.M. Kuchling's well-designed
@@ -27,18 +53,18 @@ Vinay Sajip to make use of the subproces
and so does not work on Windows). Renamed to gnupg.py to avoid confusion with
the previous versions.
-Modifications Copyright (C) 2008-2014 Vinay Sajip. All rights reserved.
+Modifications Copyright (C) 2008-2017 Vinay Sajip. All rights reserved.
A unittest harness (test_gnupg.py) has also been added.
"""
-__version__ = "0.3.8.dev0"
+__version__ = "0.4.1"
__author__ = "Vinay Sajip"
-__date__ = "$07-Dec-2014 18:46:17$"
+__date__ = "$07-Jul-2017 15:09:20$"
try:
from io import StringIO
-except ImportError:
+except ImportError: # pragma: no cover
from cStringIO import StringIO
import codecs
@@ -53,7 +79,7 @@ import sys
import threading
STARTUPINFO = None
-if os.name == 'nt':
+if os.name == 'nt': # pragma: no cover
try:
from subprocess import STARTUPINFO, STARTF_USESHOWWINDOW, SW_HIDE
except ImportError:
@@ -80,7 +106,7 @@ 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':
+if os.path.__name__ == 'ntpath': # pragma: no cover
# On Windows, we don't need shell quoting, other than worrying about
# paths with spaces in them.
def shell_quote(s):
@@ -105,7 +131,7 @@ else:
command shells
:rtype: The passed-in type
"""
- if not isinstance(s, string_types):
+ if not isinstance(s, string_types): # pragma: no cover
raise TypeError('Expected string type, got %s' % type(s))
if not s:
result = "''"
@@ -119,8 +145,18 @@ else:
# 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.
+# was needed. However, note that we still need, on 2.x, to encode any
+# Unicode argument with the file system encoding - see Issue #41 and
+# Python issue #1759845 ("subprocess.call fails with unicode strings in
+# command line").
+
+# Allows the encoding used to be overridden in special cases by setting
+# this module attribute appropriately.
+fsencoding = sys.getfilesystemencoding()
+
def no_quote(s):
+ if not _py3k and isinstance(s, text_type):
+ s = s.encode(fsencoding)
return s
def _copy_data(instream, outstream):
@@ -128,17 +164,23 @@ def _copy_data(instream, outstream):
sent = 0
if hasattr(sys.stdin, 'encoding'):
enc = sys.stdin.encoding
- else:
+ else: # pragma: no cover
enc = 'ascii'
while True:
- data = instream.read(1024)
+ # See issue #39: read can fail when e.g. a text stream is provided
+ # for what is actually a binary file
+ try:
+ data = instream.read(1024)
+ except UnicodeError:
+ logger.warning('Exception occurred while reading', exc_info=1)
+ break
if not data:
break
sent += len(data)
- logger.debug("sending chunk (%d): %r", sent, data[:256])
+ # logger.debug("sending chunk (%d): %r", sent, data[:256])
try:
outstream.write(data)
- except UnicodeError:
+ except UnicodeError: # pragma: no cover
outstream.write(data.encode(enc))
except:
# Can sometimes get 'broken pipe' errors even when the data has all
@@ -147,7 +189,7 @@ def _copy_data(instream, outstream):
break
try:
outstream.close()
- except IOError:
+ except IOError: # pragma: no cover
logger.warning('Exception occurred while closing: ignored', exc_info=1)
logger.debug("closed output, %d bytes sent", sent)
@@ -171,7 +213,7 @@ def _make_memory_stream(s):
try:
from io import BytesIO
rv = BytesIO(s)
- except ImportError:
+ except ImportError: # pragma: no cover
rv = StringIO(s)
return rv
@@ -224,25 +266,20 @@ class Verify(object):
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", "INV_SGNR", "FILE_START", "FILE_ERROR",
- "FILE_DONE", "PKA_TRUST_GOOD", "PKA_TRUST_BAD", "BADMDC",
- "GOODMDC", "NO_SGNR", "NOTATION_NAME", "NOTATION_DATA",
- "PROGRESS", "PINENTRY_LAUNCHED", "NEWSIG"):
- pass
- elif key == "BADSIG":
+ elif key in ("WARNING", "ERROR"):
+ logger.warning('potential problem: %s: %s', key, value)
+ elif key == "BADSIG": # pragma: no cover
self.valid = False
self.status = 'signature bad'
self.key_id, self.username = value.split(None, 1)
- elif key == "ERRSIG":
+ elif key == "ERRSIG": # pragma: no cover
self.valid = False
(self.key_id,
algo, hash_algo,
cls,
self.timestamp) = value.split()[:5]
self.status = 'signature error'
- elif key == "EXPSIG":
+ elif key == "EXPSIG": # pragma: no cover
self.valid = False
self.status = 'signature expired'
self.key_id, self.username = value.split(None, 1)
@@ -261,21 +298,15 @@ class Verify(object):
elif key == "SIG_ID":
(self.signature_id,
self.creation_date, self.timestamp) = value.split()
- elif key == "DECRYPTION_FAILED":
+ elif key == "DECRYPTION_FAILED": # pragma: no cover
self.valid = False
self.key_id = value
self.status = 'decryption failed'
- elif key == "NO_PUBKEY":
+ elif key == "NO_PUBKEY": # pragma: no cover
self.valid = False
self.key_id = value
self.status = 'no public key'
- elif key in ("KEYEXPIRED", "SIGEXPIRED", "KEYREVOKED"):
- # 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 or REVKEYSIG.
- pass
- elif key in ("EXPKEYSIG", "REVKEYSIG"):
+ elif key in ("EXPKEYSIG", "REVKEYSIG"): # pragma: no cover
# signed with expired or revoked key
self.valid = False
self.key_id = value.split()[0]
@@ -284,12 +315,20 @@ class Verify(object):
else:
self.key_status = 'signing key was revoked'
self.status = self.key_status
- elif key == "UNEXPECTED":
+ elif key in ("UNEXPECTED", "FAILURE"): # pragma: no cover
self.valid = False
self.key_id = value
- self.status = 'unexpected data'
- else:
- raise ValueError("Unknown status message: %r" % key)
+ if key == "UNEXPECTED":
+ self.status = 'unexpected data'
+ else:
+ # N.B. there might be other reasons
+ if not self.status:
+ self.status = 'incorrect passphrase'
+ elif key in ("DECRYPTION_INFO", "PLAINTEXT", "PLAINTEXT_LENGTH",
+ "NO_SECKEY", "BEGIN_SIGNING"):
+ pass
+ else: # pragma: no cover
+ logger.debug('message ignored: %s, %s', key, value)
class ImportResult(object):
"Handle status messages for --import"
@@ -330,10 +369,12 @@ class ImportResult(object):
}
def handle_status(self, key, value):
- if key == "IMPORTED":
+ if key in ("WARNING", "ERROR"):
+ logger.warning('potential problem: %s: %s', key, value)
+ elif key in ("IMPORTED", "KEY_CONSIDERED"):
# this duplicates info we already see in import_ok & import_problem
pass
- elif key == "NODATA":
+ elif key == "NODATA": # pragma: no cover
self.results.append({'fingerprint': None,
'problem': '0', 'text': 'No valid data found'})
elif key == "IMPORT_OK":
@@ -346,7 +387,7 @@ class ImportResult(object):
self.results.append({'fingerprint': fingerprint,
'ok': reason, 'text': reasontext})
self.fingerprints.append(fingerprint)
- elif key == "IMPORT_PROBLEM":
+ elif key == "IMPORT_PROBLEM": # pragma: no cover
try:
reason, fingerprint = value.split()
except:
@@ -358,19 +399,22 @@ class ImportResult(object):
import_res = value.split()
for i, count in enumerate(self.counts):
setattr(self, count, int(import_res[i]))
- elif key == "KEYEXPIRED":
+ elif key == "KEYEXPIRED": # pragma: no cover
self.results.append({'fingerprint': None,
'problem': '0', 'text': 'Key expired'})
- elif key == "SIGEXPIRED":
+ elif key == "SIGEXPIRED": # pragma: no cover
self.results.append({'fingerprint': None,
'problem': '0', 'text': 'Signature expired'})
- else:
- raise ValueError("Unknown status message: %r" % key)
+ elif key == "FAILURE": # pragma: no cover
+ self.results.append({'fingerprint': None,
+ 'problem': '0', 'text': 'Other failure'})
+ else: # pragma: no cover
+ logger.debug('message ignored: %s, %s', key, value)
def summary(self):
l = []
l.append('%d imported' % self.imported)
- if self.not_imported:
+ if self.not_imported: # pragma: no cover
l.append('%d not imported' % self.not_imported)
return ', '.join(l)
@@ -411,8 +455,12 @@ class SearchKeys(list):
def get_fields(self, args):
result = {}
for i, var in enumerate(self.FIELDS):
- result[var] = args[i]
+ if i < len(args):
+ result[var] = args[i]
+ else:
+ result[var] = 'unavailable'
result['uids'] = []
+ result['sigs'] = []
return result
def pub(self, args):
@@ -427,11 +475,11 @@ class SearchKeys(list):
self.curkey['uids'].append(uid)
self.uids.append(uid)
- def handle_status(self, key, value):
+ def handle_status(self, key, value): # pragma: no cover
pass
class ListKeys(SearchKeys):
- ''' Handle status messages for --list-keys.
+ ''' Handle status messages for --list-keys, --list-sigs.
Handle pub and uid (relating the latter to the former).
@@ -439,7 +487,6 @@ class ListKeys(SearchKeys):
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
@@ -449,7 +496,12 @@ class ListKeys(SearchKeys):
'''
UID_INDEX = 9
- FIELDS = 'type trust length algo keyid date expires dummy ownertrust uid'.split()
+ FIELDS = 'type trust length algo keyid date expires dummy ownertrust uid sig'.split()
+
+ def __init__(self, gpg):
+ super(ListKeys, self).__init__(gpg)
+ self.in_subkey = False
+ self.key_map = {}
def key(self, args):
self.curkey = curkey = self.get_fields(args)
@@ -458,17 +510,35 @@ class ListKeys(SearchKeys):
del curkey['uid']
curkey['subkeys'] = []
self.append(curkey)
+ self.in_subkey = False
pub = sec = key
def fpr(self, args):
- self.curkey['fingerprint'] = args[9]
- self.fingerprints.append(args[9])
+ fp = args[9]
+ if fp in self.key_map: # pragma: no cover
+ raise ValueError('Unexpected fingerprint collision: %s' % fp)
+ if not self.in_subkey:
+ self.curkey['fingerprint'] = fp
+ self.fingerprints.append(fp)
+ self.key_map[fp] = self.curkey
+ else:
+ self.curkey['subkeys'][-1].append(fp)
+ self.key_map[fp] = self.curkey
def sub(self, args):
- subkey = [args[4], args[11]]
+ subkey = [args[4], args[11]] # keyid, type
+ self.curkey['subkeys'].append(subkey)
+ self.in_subkey = True
+
+ def ssb(self, args):
+ subkey = [args[4], None] # keyid, type
self.curkey['subkeys'].append(subkey)
+ self.in_subkey = True
+ def sig(self, args):
+ # keyid, uid, sigclass
+ self.curkey['sigs'].append((args[4], args[9], args[10]))
class ScanKeys(ListKeys):
''' Handle status messages for --with-fingerprint.'''
@@ -478,6 +548,7 @@ class ScanKeys(ListKeys):
# use the last value args[-1] instead of args[11]
subkey = [args[4], args[-1]]
self.curkey['subkeys'].append(subkey)
+ self.in_subkey = True
class TextHandler(object):
def _as_text(self):
@@ -507,13 +578,10 @@ class Crypt(Verify, TextHandler):
__bool__ = __nonzero__
def handle_status(self, key, value):
- if key in ("ENC_TO", "USERID_HINT", "GOODMDC", "END_DECRYPTION",
- "BEGIN_SIGNING", "NO_SECKEY", "ERROR", "NODATA", "PROGRESS",
- "CARDCTRL", "BADMDC", "SC_OP_FAILURE", "SC_OP_SUCCESS",
- "PINENTRY_LAUNCHED"):
- # in the case of ERROR, this is because a more specific error
- # message will have come first
- pass
+ if key in ("WARNING", "ERROR"):
+ logger.warning('potential problem: %s: %s', key, value)
+ elif key == "NODATA":
+ self.status = "no data was provided"
elif key in ("NEED_PASSPHRASE", "BAD_PASSPHRASE", "GOOD_PASSPHRASE",
"MISSING_PASSPHRASE", "DECRYPTION_FAILED",
"KEY_NOT_CREATED", "NEED_PASSPHRASE_PIN"):
@@ -530,14 +598,19 @@ class Crypt(Verify, TextHandler):
elif key == "END_ENCRYPTION":
self.status = 'encryption ok'
self.ok = True
- elif key == "INV_RECP":
+ elif key == "INV_RECP": # pragma: no cover
self.status = 'invalid recipient'
- elif key == "KEYEXPIRED":
+ elif key == "KEYEXPIRED": # pragma: no cover
self.status = 'key expired'
- elif key == "SIG_CREATED":
+ elif key == "SIG_CREATED": # pragma: no cover
self.status = 'sig created'
- elif key == "SIGEXPIRED":
+ elif key == "SIGEXPIRED": # pragma: no cover
self.status = 'sig expired'
+ elif key in ("ENC_TO", "USERID_HINT", "GOODMDC",
+ "END_DECRYPTION", "CARDCTRL", "BADMDC",
+ "SC_OP_FAILURE", "SC_OP_SUCCESS",
+ "PINENTRY_LAUNCHED", "KEY_CONSIDERED"):
+ pass
else:
Verify.handle_status(self, key, value)
@@ -558,13 +631,26 @@ class GenKey(object):
return self.fingerprint or ''
def handle_status(self, key, value):
- if key in ("PROGRESS", "GOOD_PASSPHRASE", "NODATA", "KEY_NOT_CREATED",
- "PINENTRY_LAUNCHED"):
- pass
+ if key in ("WARNING", "ERROR"): # pragma: no cover
+ logger.warning('potential problem: %s: %s', key, value)
elif key == "KEY_CREATED":
(self.type,self.fingerprint) = value.split()
+ elif key in ("PROGRESS", "GOOD_PASSPHRASE", "KEY_NOT_CREATED"):
+ pass
+ else: # pragma: no cover
+ logger.debug('message ignored: %s, %s', key, value)
+
+class ExportResult(GenKey):
+ """Handle status messages for --export[-secret-key].
+
+ For now, just use an existing class to base it on - if needed, we
+ can override handle_status for more specific message handling.
+ """
+ def handle_status(self, key, value):
+ if key in ("EXPORTED", "EXPORT_RES"):
+ pass
else:
- raise ValueError("Unknown status message: %r" % key)
+ super(ExportResult, self).handle_status(key, value)
class DeleteResult(object):
"Handle status messages for --delete-key and --delete-secret-key"
@@ -582,11 +668,11 @@ class DeleteResult(object):
}
def handle_status(self, key, value):
- if key == "DELETE_PROBLEM":
+ if key == "DELETE_PROBLEM": # pragma: no cover
self.status = self.problem_reason.get(value,
"Unknown error: %r" % value)
- else:
- raise ValueError("Unknown status message: %r" % key)
+ else: # pragma: no cover
+ logger.debug('message ignored: %s, %s', key, value)
def __nonzero__(self):
return self.status == 'ok'
@@ -601,6 +687,7 @@ class Sign(TextHandler):
self.type = None
self.hash_algo = None
self.fingerprint = None
+ self.status = None
def __nonzero__(self):
return self.fingerprint is not None
@@ -608,22 +695,23 @@ class Sign(TextHandler):
__bool__ = __nonzero__
def handle_status(self, key, value):
- if key in ("USERID_HINT", "NEED_PASSPHRASE", "BAD_PASSPHRASE",
- "GOOD_PASSPHRASE", "BEGIN_SIGNING", "CARDCTRL", "INV_SGNR",
- "NO_SGNR", "MISSING_PASSPHRASE", "NEED_PASSPHRASE_PIN",
- "SC_OP_FAILURE", "SC_OP_SUCCESS", "PROGRESS"):
- pass
- elif key in ("KEYEXPIRED", "SIGEXPIRED"):
+ if key in ("WARNING", "ERROR", "FAILURE"): # pragma: no cover
+ logger.warning('potential problem: %s: %s', key, value)
+ elif key in ("KEYEXPIRED", "SIGEXPIRED"): # pragma: no cover
self.status = 'key expired'
- elif key == "KEYREVOKED":
+ elif key == "KEYREVOKED": # pragma: no cover
self.status = 'key revoked'
elif key == "SIG_CREATED":
(self.type,
algo, self.hash_algo, cls,
self.timestamp, self.fingerprint
) = value.split()
- else:
- raise ValueError("Unknown status message: %r" % key)
+ self.status = 'signature created'
+ elif key in ("USERID_HINT", "NEED_PASSPHRASE", "GOOD_PASSPHRASE",
+ "BAD_PASSPHRASE", "BEGIN_SIGNING"):
+ pass
+ else: # pragma: no cover
+ logger.debug('message ignored: %s, %s', key, value)
VERSION_RE = re.compile(r'gpg \(GnuPG\) (\d+(\.\d+)*)'.encode('ascii'), re.I)
HEX_DIGITS_RE = re.compile(r'[0-9a-f]+$', re.I)
@@ -643,6 +731,7 @@ class GPG(object):
'search': SearchKeys,
'sign': Sign,
'verify': Verify,
+ 'export': ExportResult,
}
"Encapsulate access to the gpg executable"
@@ -677,9 +766,10 @@ class GPG(object):
self.secret_keyring = secret_keyring
self.verbose = verbose
self.use_agent = use_agent
- if isinstance(options, str):
+ if isinstance(options, str): # pragma: no cover
options = [options]
self.options = options
+ self.on_data = None # or a callable - will be called with data chunks
# Changed in 0.3.7 to use Latin-1 encoding rather than
# locale.getpreferredencoding falling back to sys.stdin.encoding
# falling back to utf-8, because gpg itself uses latin-1 as the default
@@ -687,14 +777,19 @@ class GPG(object):
self.encoding = 'latin-1'
if gnupghome and not os.path.isdir(self.gnupghome):
os.makedirs(self.gnupghome,0x1C0)
- p = self._open_subprocess(["--version"])
+ try:
+ p = self._open_subprocess(["--version"])
+ except OSError:
+ msg = 'Unable to run gpg - it may not be available.'
+ logger.exception(msg)
+ raise OSError(msg)
result = self.result_map['verify'](self) # any result will do for this
self._collect_output(p, result, stdin=p.stdin)
- if p.returncode != 0:
+ if p.returncode != 0: # pragma: no cover
raise ValueError("Error invoking gpg: %s: %s" % (p.returncode,
result.stderr))
m = VERSION_RE.match(result.data)
- if not m:
+ if not m: # pragma: no cover
self.version = None
else:
dot = '.'.encode('ascii')
@@ -707,6 +802,11 @@ class GPG(object):
a passphrase will be sent to GPG, else False.
"""
cmd = [self.gpgbinary, '--status-fd', '2', '--no-tty']
+ cmd.extend(['--debug', 'ipc'])
+ if passphrase and hasattr(self, 'version'):
+ if self.version >= (2, 1):
+ cmd[1:1] = ['--pinentry-mode', 'loopback']
+ cmd.extend(['--fixed-list-mode', '--batch', '--with-colons'])
if self.gnupghome:
cmd.extend(['--homedir', no_quote(self.gnupghome)])
if self.keyring:
@@ -717,8 +817,8 @@ class GPG(object):
for fn in self.secret_keyring:
cmd.extend(['--secret-keyring', no_quote(fn)])
if passphrase:
- cmd.extend(['--batch', '--passphrase-fd', '0'])
- if self.use_agent:
+ cmd.extend(['--passphrase-fd', '0'])
+ if self.use_agent: # pragma: no cover
cmd.append('--use-agent')
if self.options:
cmd.extend(self.options)
@@ -728,19 +828,35 @@ class GPG(object):
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 debug_print(cmd):
+ # result = []
+ # for c in cmd:
+ # if ' ' not in c:
+ # result.append(c)
+ # else:
+ # if '"' not in c:
+ # result.append('"%s"' % c)
+ # elif "'" not in c:
+ # result.append("'%s'" % c)
+ # else:
+ # result.append(c) # give up
+ # return ' '.join(cmd)
+ from subprocess import list2cmdline as debug_print
+
cmd = self.make_args(args, passphrase)
- if self.verbose:
- pcmd = ' '.join(cmd)
- print(pcmd)
- logger.debug("%s", cmd)
+ if self.verbose: # pragma: no cover
+ print(debug_print(cmd))
if not STARTUPINFO:
si = None
- else:
+ else: # pragma: no cover
si = STARTUPINFO()
si.dwFlags = STARTF_USESHOWWINDOW
si.wShowWindow = SW_HIDE
- return Popen(cmd, shell=False, stdin=PIPE, stdout=PIPE, stderr=PIPE,
- startupinfo=si)
+ result = Popen(cmd, shell=False, stdin=PIPE, stdout=PIPE, stderr=PIPE,
+ startupinfo=si)
+ logger.debug("%s: %s", result.pid, debug_print(cmd))
+ return result
def _read_response(self, stream, result):
# Internal method: reads all the stderr output from GPG, taking notice
@@ -755,7 +871,7 @@ class GPG(object):
break
lines.append(line)
line = line.rstrip()
- if self.verbose:
+ if self.verbose: # pragma: no cover
print(line)
logger.debug("%s", line)
if line[0:9] == '[GNUPG:] ':
@@ -770,7 +886,7 @@ class GPG(object):
result.handle_status(keyword, value)
result.stderr = ''.join(lines)
- def _read_data(self, stream, result):
+ def _read_data(self, stream, result, on_data=None):
# Read the contents of the file from GPG's stdout
chunks = []
while True:
@@ -779,6 +895,8 @@ class GPG(object):
break
logger.debug("chunk: %r" % data[:256])
chunks.append(data)
+ if on_data:
+ on_data(data)
if _py3k:
# Join using b'' or '', as appropriate
result.data = type(data)().join(chunks)
@@ -799,7 +917,7 @@ class GPG(object):
rr.start()
stdout = process.stdout
- dr = threading.Thread(target=self._read_data, args=(stdout, result))
+ dr = threading.Thread(target=self._read_data, args=(stdout, result, self.on_data))
dr.setDaemon(True)
logger.debug('stdout reader: %r', dr)
dr.start()
@@ -812,7 +930,7 @@ class GPG(object):
if stdin is not None:
try:
stdin.close()
- except IOError:
+ except IOError: # pragma: no cover
pass
stderr.close()
stdout.close()
@@ -822,7 +940,7 @@ 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:
+ if not binary: # pragma: no cover
stdin = codecs.getwriter(self.encoding)(p.stdin)
else:
stdin = p.stdin
@@ -846,14 +964,14 @@ class GPG(object):
"If writing to a file which exists, avoid a confirmation message."
if os.path.exists(output):
# We need to avoid an overwrite confirmation message
- args.extend(['--batch', '--yes'])
- args.extend(['--output', output])
+ args.extend(['--yes'])
+ args.extend(['--output', no_quote(output)])
def sign_file(self, file, keyid=None, passphrase=None, clearsign=True,
- detach=False, binary=False, output=None):
+ detach=False, binary=False, output=None, extra_args=None):
"""sign file"""
logger.debug("sign_file: %s", file)
- if binary:
+ if binary: # pragma: no cover
args = ['-s']
else:
args = ['-sa']
@@ -868,6 +986,8 @@ class GPG(object):
if output: # write the output to a file with the specified name
self.set_output_without_confirmation(args, output)
+ if extra_args:
+ args.extend(extra_args)
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.
@@ -877,17 +997,18 @@ class GPG(object):
if passphrase:
_write_passphrase(stdin, passphrase, self.encoding)
writer = _threaded_copy_data(file, stdin)
- except IOError:
+ except IOError: # pragma: no cover
logging.exception("error writing message")
writer = None
self._collect_output(p, result, writer, stdin)
return result
- def verify(self, data):
+ def verify(self, data, **kwargs):
"""Verify the signature on the contents of the string 'data'
- >>> gpg = GPG(gnupghome="keys")
- >>> input = gpg.gen_key_input(Passphrase='foo')
+ >>> GPGBINARY = os.environ.get('GPGBINARY', 'gpg')
+ >>> gpg = GPG(gpgbinary=GPGBINARY, 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')
@@ -899,15 +1020,17 @@ class GPG(object):
"""
f = _make_binary_stream(data, self.encoding)
- result = self.verify_file(f)
+ result = self.verify_file(f, **kwargs)
f.close()
return result
- def verify_file(self, file, data_filename=None):
+ def verify_file(self, file, data_filename=None, close_file=True, extra_args=None):
"Verify the signature on the contents of the file-like object 'file'"
logger.debug('verify_file: %r, %r', file, data_filename)
result = self.result_map['verify'](self)
args = ['--verify']
+ if extra_args:
+ args.extend(extra_args)
if data_filename is None:
self._handle_io(args, file, result, binary=True)
else:
@@ -915,7 +1038,8 @@ class GPG(object):
import tempfile
fd, fn = tempfile.mkstemp(prefix='pygpg')
s = file.read()
- file.close()
+ if close_file:
+ file.close()
logger.debug('Wrote to temp file: %r', s)
os.write(fd, s)
os.close(fd)
@@ -928,11 +1052,14 @@ class GPG(object):
os.unlink(fn)
return result
- def verify_data(self, sig_filename, data):
+ def verify_data(self, sig_filename, data, extra_args=None):
"Verify the signature in sig_filename against data in memory"
logger.debug('verify_data: %r, %r ...', sig_filename, data[:16])
result = self.result_map['verify'](self)
- args = ['--verify', no_quote(sig_filename), '-']
+ args = ['--verify']
+ if extra_args:
+ args.extend(extra_args)
+ args.extend([no_quote(sig_filename), '-'])
stream = _make_memory_stream(data)
self._handle_io(args, stream, result, binary=True)
return result
@@ -942,49 +1069,8 @@ class GPG(object):
#
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
-
+ """
+ Import the key_data into our keyring.
"""
result = self.result_map['import'](self)
logger.debug('import_keys: %r', key_data[:256])
@@ -998,10 +1084,11 @@ class GPG(object):
"""Import a key from a keyserver
>>> import shutil
- >>> shutil.rmtree("keys")
- >>> gpg = GPG(gnupghome="keys")
+ >>> shutil.rmtree("keys", ignore_errors=True)
+ >>> GPGBINARY = os.environ.get('GPGBINARY', 'gpg')
+ >>> gpg = GPG(gpgbinary=GPGBINARY, gnupghome="keys")
>>> os.chmod('keys', 0x1C0)
- >>> result = gpg.recv_keys('keyserver.ubuntu.com', '92905378')
+ >>> result = gpg.recv_keys('pgp.mit.edu', '92905378')
>>> assert result
"""
@@ -1033,26 +1120,48 @@ class GPG(object):
data.close()
return result
- def delete_keys(self, fingerprints, secret=False):
+ def delete_keys(self, fingerprints, secret=False, passphrase=None):
which='key'
- if secret:
+ if secret: # pragma: no cover
+ if self.version >= (2, 1) and passphrase is None:
+ raise ValueError('For GnuPG >= 2.1, deleting secret keys '
+ 'needs a passphrase to be provided')
which='secret-key'
- if _is_sequence(fingerprints):
+ if _is_sequence(fingerprints): # pragma: no cover
fingerprints = [no_quote(s) for s in fingerprints]
else:
fingerprints = [no_quote(fingerprints)]
- args = ['--batch', '--delete-%s' % which]
+ args = ['--delete-%s' % which]
args.extend(fingerprints)
result = self.result_map['delete'](self)
- p = self._open_subprocess(args)
- self._collect_output(p, result, stdin=p.stdin)
+ if not secret or self.version < (2, 1):
+ p = self._open_subprocess(args)
+ self._collect_output(p, result, stdin=p.stdin)
+ else:
+ # Need to send in a passphrase.
+ f = _make_binary_stream('', self.encoding)
+ try:
+ self._handle_io(args, f, result, passphrase=passphrase,
+ binary=True)
+ finally:
+ f.close()
return result
- def export_keys(self, keyids, secret=False, armor=True, minimal=False):
- "export the indicated keys. 'keyid' is anything gpg accepts"
+ def export_keys(self, keyids, secret=False, armor=True, minimal=False,
+ passphrase=None):
+ """
+ Export the indicated keys. A 'keyid' is anything gpg accepts.
+
+ Since GnuPG 2.1, you can't export secret keys without providing a
+ passphrase.
+ """
+
which=''
if secret:
which='-secret-key'
+ if self.version >= (2, 1) and passphrase is None:
+ raise ValueError('For GnuPG >= 2.1, exporting secret keys '
+ 'needs a passphrase to be provided')
if _is_sequence(keyids):
keyids = [no_quote(k) for k in keyids]
else:
@@ -1060,17 +1169,30 @@ class GPG(object):
args = ['--export%s' % which]
if armor:
args.append('--armor')
- if minimal:
+ if minimal: # pragma: no cover
args.extend(['--export-options','export-minimal'])
args.extend(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 = self.result_map['delete'](self) # any result will do
- self._collect_output(p, result, stdin=p.stdin)
+ result = self.result_map['export'](self)
+ if not secret or self.version < (2, 1):
+ p = self._open_subprocess(args)
+ self._collect_output(p, result, stdin=p.stdin)
+ else:
+ # Need to send in a passphrase.
+ f = _make_binary_stream('', self.encoding)
+ try:
+ self._handle_io(args, f, result, passphrase=passphrase,
+ binary=True)
+ finally:
+ f.close()
logger.debug('export_keys result: %r', result.data)
- return result.data.decode(self.encoding, self.decode_errors)
+ # Issue #49: Return bytes if armor not specified, else text
+ result = result.data
+ if armor:
+ result = result.decode(self.encoding, self.decode_errors)
+ return result
def _get_list_output(self, p, kind):
# Get the response information
@@ -1078,43 +1200,50 @@ class GPG(object):
self._collect_output(p, result, stdin=p.stdin)
lines = result.data.decode(self.encoding,
self.decode_errors).splitlines()
- valid_keywords = 'pub uid sec fpr sub'.split()
+ valid_keywords = 'pub uid sec fpr sub ssb sig'.split()
for line in lines:
- if self.verbose:
+ if self.verbose: # pragma: no cover
print(line)
logger.debug("line: %r", line.rstrip())
- if not line:
+ if not line: # pragma: no cover
break
L = line.strip().split(':')
- if not L:
+ if not L: # pragma: no cover
continue
keyword = L[0]
if keyword in valid_keywords:
getattr(result, keyword)(L)
return result
- def list_keys(self, secret=False):
+ def list_keys(self, secret=False, keys=None, sigs=False):
""" list the keys currently in the keyring
>>> import shutil
- >>> shutil.rmtree("keys")
- >>> gpg = GPG(gnupghome="keys")
- >>> input = gpg.gen_key_input()
+ >>> shutil.rmtree("keys", ignore_errors=True)
+ >>> GPGBINARY = os.environ.get('GPGBINARY', 'gpg')
+ >>> gpg = GPG(gpgbinary=GPGBINARY, gnupghome="keys")
+ >>> input = gpg.gen_key_input(passphrase='foo')
>>> result = gpg.gen_key(input)
- >>> print1 = result.fingerprint
+ >>> fp1 = result.fingerprint
>>> result = gpg.gen_key(input)
- >>> print2 = result.fingerprint
+ >>> fp2 = result.fingerprint
>>> pubkeys = gpg.list_keys()
- >>> assert print1 in pubkeys.fingerprints
- >>> assert print2 in pubkeys.fingerprints
+ >>> assert fp1 in pubkeys.fingerprints
+ >>> assert fp2 in pubkeys.fingerprints
"""
- which='keys'
+ if sigs:
+ which = 'sigs'
+ else: which='keys'
if secret:
which='secret-keys'
- args = ["--list-%s" % which, "--fixed-list-mode", "--fingerprint",
- "--with-colons"]
+ args = ['--list-%s' % which,
+ '--fingerprint', '--fingerprint'] # get subkey FPs, too
+ if keys:
+ if isinstance(keys, string_types):
+ keys = [keys]
+ args.extend(keys)
p = self._open_subprocess(args)
return self._get_list_output(p, 'list')
@@ -1123,10 +1252,20 @@ class GPG(object):
List details of an ascii armored or binary key file
without first importing it to the local keyring.
- The function achieves this by running:
+ The function achieves this on modern GnuPG by running:
+
+ $ gpg --dry-run --import-options import-show --import
+
+ On older versions, it does the *much* riskier:
+
$ gpg --with-fingerprint --with-colons filename
"""
- args = ['--with-fingerprint', '--with-colons']
+ if self.version >= (2, 1):
+ args = ['--dry-run', '--import-options', 'import-show', '--import']
+ else:
+ logger.warning('Trying to list packets, but if the file is not a '
+ 'keyring, might accidentally decrypt')
+ args = ['--with-fingerprint', '--with-colons', '--fixed-list-mode']
args.append(no_quote(filename))
p = self._open_subprocess(args)
return self._get_list_output(p, 'scan')
@@ -1135,20 +1274,21 @@ class GPG(object):
""" search keyserver by query (using --search-keys option)
>>> import shutil
- >>> shutil.rmtree('keys')
- >>> gpg = GPG(gnupghome='keys')
+ >>> shutil.rmtree('keys', ignore_errors=True)
+ >>> GPGBINARY = os.environ.get('GPGBINARY', 'gpg')
+ >>> gpg = GPG(gpgbinary=GPGBINARY, gnupghome='keys')
>>> os.chmod('keys', 0x1C0)
>>> result = gpg.search_keys('<vi...@hotmail.com>')
>>> assert result, 'Failed using default keyserver'
- >>> keyserver = 'keyserver.ubuntu.com'
- >>> result = gpg.search_keys('<vi...@hotmail.com>', keyserver)
- >>> assert result, 'Failed using keyserver.ubuntu.com'
+ >>> #keyserver = 'keyserver.ubuntu.com'
+ >>> #result = gpg.search_keys('<vi...@hotmail.com>', keyserver)
+ >>> #assert result, 'Failed using keyserver.ubuntu.com'
"""
query = query.strip()
if HEX_DIGITS_RE.match(query):
query = '0x' + query
- args = ['--fixed-list-mode', '--fingerprint', '--with-colons',
+ args = ['--fingerprint',
'--keyserver', no_quote(keyserver), '--search-keys',
no_quote(query)]
p = self._open_subprocess(args)
@@ -1160,13 +1300,13 @@ class GPG(object):
self.decode_errors).splitlines()
valid_keywords = ['pub', 'uid']
for line in lines:
- if self.verbose:
+ if self.verbose: # pragma: no cover
print(line)
logger.debug('line: %r', line.rstrip())
if not line: # sometimes get blank lines on Windows
continue
L = line.strip().split(':')
- if not L:
+ if not L: # pragma: no cover
continue
keyword = L[0]
if keyword in valid_keywords:
@@ -1177,15 +1317,16 @@ class GPG(object):
"""Generate a key; you might use gen_key_input() to create the
control input.
- >>> gpg = GPG(gnupghome="keys")
- >>> input = gpg.gen_key_input()
+ >>> GPGBINARY = os.environ.get('GPGBINARY', 'gpg')
+ >>> gpg = GPG(gpgbinary=GPGBINARY, gnupghome="keys")
+ >>> input = gpg.gen_key_input(passphrase='foo')
>>> result = gpg.gen_key(input)
>>> assert result
>>> result = gpg.gen_key('foo')
>>> assert not result
"""
- args = ["--gen-key", "--batch"]
+ args = ["--gen-key"]
result = self.result_map['generate'](self)
f = _make_binary_stream(input, self.encoding)
self._handle_io(args, f, result, binary=True)
@@ -1242,7 +1383,7 @@ class GPG(object):
#
def encrypt_file(self, file, recipients, sign=None,
always_trust=False, passphrase=None,
- armor=True, output=None, symmetric=False):
+ armor=True, output=None, symmetric=False, extra_args=None):
"Encrypt the message read from the file-like object 'file'"
args = ['--encrypt']
if symmetric:
@@ -1264,12 +1405,14 @@ class GPG(object):
args.append('--armor')
if output: # write the output to a file with the specified name
self.set_output_without_confirmation(args, output)
- if sign is True:
+ if sign is True: # pragma: no cover
args.append('--sign')
- elif sign:
+ elif sign: # pragma: no cover
args.extend(['--sign', '--default-key', no_quote(sign)])
- if always_trust:
+ if always_trust: # pragma: no cover
args.append('--always-trust')
+ if extra_args:
+ args.extend(extra_args)
result = self.result_map['crypt'](self)
self._handle_io(args, file, result, passphrase=passphrase, binary=True)
logger.debug('encrypt result: %r', result.data)
@@ -1280,40 +1423,41 @@ class GPG(object):
>>> import shutil
>>> if os.path.exists("keys"):
- ... shutil.rmtree("keys")
- >>> gpg = GPG(gnupghome="keys")
- >>> input = gpg.gen_key_input(passphrase='foo')
+ ... shutil.rmtree("keys", ignore_errors=True)
+ >>> GPGBINARY = os.environ.get('GPGBINARY', 'gpg')
+ >>> gpg = GPG(gpgbinary=GPGBINARY, gnupghome="keys")
+ >>> input = gpg.gen_key_input(name_email='user1@test', passphrase='pp1')
>>> result = gpg.gen_key(input)
- >>> print1 = result.fingerprint
- >>> input = gpg.gen_key_input()
+ >>> fp1 = result.fingerprint
+ >>> input = gpg.gen_key_input(name_email='user2@test', passphrase='pp2')
>>> result = gpg.gen_key(input)
- >>> print2 = result.fingerprint
- >>> result = gpg.encrypt("hello",print2)
+ >>> fp2 = result.fingerprint
+ >>> result = gpg.encrypt("hello",fp2)
>>> message = str(result)
>>> assert message != 'hello'
- >>> result = gpg.decrypt(message)
+ >>> result = gpg.decrypt(message, passphrase='pp2')
>>> assert result
>>> str(result)
'hello'
- >>> result = gpg.encrypt("hello again",print1)
+ >>> result = gpg.encrypt("hello again", fp1)
>>> message = str(result)
- >>> result = gpg.decrypt(message,passphrase='bar')
+ >>> result = gpg.decrypt(message, passphrase='bar')
>>> result.status in ('decryption failed', 'bad passphrase')
True
>>> assert not result
- >>> result = gpg.decrypt(message,passphrase='foo')
+ >>> result = gpg.decrypt(message, passphrase='pp1')
>>> result.status == 'decryption ok'
True
>>> str(result)
'hello again'
- >>> result = gpg.encrypt("signed hello",print2,sign=print1,passphrase='foo')
+ >>> result = gpg.encrypt("signed hello", fp2, sign=fp1, passphrase='pp1')
>>> result.status == 'encryption ok'
True
>>> message = str(result)
- >>> result = gpg.decrypt(message)
+ >>> result = gpg.decrypt(message, passphrase='pp2')
>>> result.status == 'decryption ok'
True
- >>> assert result.fingerprint == print1
+ >>> assert result.fingerprint == fp1
"""
data = _make_binary_stream(data, self.encoding)
@@ -1328,14 +1472,15 @@ class GPG(object):
return result
def decrypt_file(self, file, always_trust=False, passphrase=None,
- output=None):
+ output=None, extra_args=None):
args = ["--decrypt"]
if output: # write the output to a file with the specified name
self.set_output_without_confirmation(args, output)
- if always_trust:
+ if always_trust: # pragma: no cover
args.append("--always-trust")
+ if extra_args:
+ args.extend(extra_args)
result = self.result_map['crypt'](self)
self._handle_io(args, file, result, passphrase, binary=True)
logger.debug('decrypt result: %r', result.data)
return result
-