Unsignedio
/
python-gnupg
mirror of https://github.com/markqvist/python-gnupg.git
1
0
Fork 0
Code Issues Releases Wiki Activity

Merge branch 'feature/cleanup-api' into develop

testing/mmn/mktime_takes_localtime_not_gmtime
Isis Lovecruft 2013-04-17 23:52:10 +00:00
parent 4706932275 df5346694d
commit fd6dbe71b6
No known key found for this signature in database
GPG Key ID: A3ADB67A2CDB8B35
5 changed files with 251 additions and 181 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
Makefile
View File

@ -5,13 +5,13 @@ clean:
rm -f ./*.pyo rm -f ./*.pyo
cleantest: clean cleantest: clean
mkdir -p gnupg/tests/keys mkdir -p gnupg/tests/tmp_test
touch gnupg/tests/placeholder.log touch gnupg/tests/placeholder.log
rm -rf gnupg/tests/keys rm -rf gnupg/tests/tmp_test
rm gnupg/tests/*.log rm gnupg/tests/*.log
test: cleantest test: cleantest
python gnupg/tests/test_gnupg.py parsers basic genkey sign python gnupg/tests/test_gnupg.py parsers basic genkey sign listkeys
install: install:
python setup.py install --record installed-files.txt python setup.py install --record installed-files.txt

226
gnupg/gnupg.py
View File

@ -107,78 +107,7 @@ from parsers import GenKey, Sign, ListKeys, ListPackets
from parsers import _fix_unsafe, _sanitise, _is_allowed, _sanitise_list from parsers import _fix_unsafe, _sanitise, _is_allowed, _sanitise_list
from util import logger, _conf from util import logger, _conf
import util import util as _util
def _copy_data(instream, outstream):
"""Copy data from one stream to another.
:type instream: :class:`io.BytesIO` or :class:`io.StringIO` or file
:param instream: A byte stream or open file to read from.
:param file outstream: The file descriptor of a tmpfile to write to.
"""
sent = 0
try:
#assert (util._is_stream(instream)
# or isinstance(instream, file)), "instream not stream or file"
assert isinstance(outstream, file), "outstream is not a file"
except AssertionError as ae:
logger.exception(ae)
return
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("_copy_data(): sending chunk (%d):\n%s" % (sent, data[:256]))
try:
outstream.write(data)
except UnicodeError:
try:
outstream.write(data.encode(enc))
except IOError:
logger.exception('_copy_data(): Error sending data: Broken pipe')
break
except IOError:
# Can get 'broken pipe' errors even when all data was sent
logger.exception('_copy_data(): Error sending data: Broken pipe')
break
try:
outstream.close()
except IOError:
logger.exception('_copy_data(): Got IOError while closing %s' % outstream)
else:
logger.debug("_copy_data(): Closed output, %d bytes sent." % sent)
def _threaded_copy_data(instream, outstream):
"""Copy data from one stream to another in a separate thread.
Wraps ``_copy_data()`` in a :class:`threading.Thread`.
:type instream: :class:`io.BytesIO` or :class:`io.StringIO`
:param instream: A byte stream to read from.
:param file outstream: The file descriptor of a tmpfile to write to.
"""
copy_thread = threading.Thread(target=_copy_data,
args=(instream, outstream))
copy_thread.setDaemon(True)
logger.debug('_threaded_copy_data(): %r, %r, %r', copy_thread,
instream, outstream)
copy_thread.start()
return copy_thread
def _write_passphrase(stream, passphrase, encoding):
passphrase = '%s\n' % passphrase
passphrase = passphrase.encode(encoding)
stream.write(passphrase)
logger.debug("_write_passphrase(): Wrote passphrase.")
class GPG(object): class GPG(object):
@ -227,12 +156,12 @@ class GPG(object):
gpghome = _conf gpghome = _conf
self.gpghome = _fix_unsafe(gpghome) self.gpghome = _fix_unsafe(gpghome)
if self.gpghome: if self.gpghome:
util._create_gpghome(self.gpghome) _util._create_gpghome(self.gpghome)
else: else:
message = ("Unsuitable gpg home dir: %s" % gpghome) message = ("Unsuitable gpg home dir: %s" % gpghome)
logger.debug("GPG.__init__(): %s" % message) logger.debug("GPG.__init__(): %s" % message)
self.gpgbinary = util._find_gpgbinary(gpgbinary) self.gpgbinary = _util._find_gpgbinary(gpgbinary)
if keyring is not None: if keyring is not None:
raise DeprecationWarning("Option 'keyring' changing to 'secring'") raise DeprecationWarning("Option 'keyring' changing to 'secring'")
@ -242,17 +171,6 @@ class GPG(object):
self.secring = os.path.join(self.gpghome, secring) self.secring = os.path.join(self.gpghome, secring)
self.pubring = os.path.join(self.gpghome, pubring) self.pubring = os.path.join(self.gpghome, pubring)
#for ring in [self.secring, self.pubring]:
# if ring and not os.path.isfile(ring):
# with open(ring, 'a+') as ringfile:
# ringfile.write("")
# ringfile.flush()
# try:
# assert util._has_readwrite(ring), \
# ("Need r+w for %s" % ring)
# except AssertionError as ae:
# logger.debug(ae.message)
self.options = _sanitise(options) if options else None self.options = _sanitise(options) if options else None
self.encoding = locale.getpreferredencoding() self.encoding = locale.getpreferredencoding()
@ -261,7 +179,7 @@ class GPG(object):
try: try:
assert self.gpghome is not None, "Got None for self.gpghome" assert self.gpghome is not None, "Got None for self.gpghome"
assert util._has_readwrite(self.gpghome), ("Home dir %s needs r+w" assert _util._has_readwrite(self.gpghome), ("Home dir %s needs r+w"
% self.gpghome) % self.gpghome)
assert self.gpgbinary, "Could not find gpgbinary %s" % full assert self.gpgbinary, "Could not find gpgbinary %s" % full
assert isinstance(verbose, bool), "'verbose' must be boolean" assert isinstance(verbose, bool), "'verbose' must be boolean"
@ -354,7 +272,7 @@ class GPG(object):
break break
logger.debug("chunk: %r" % data[:256]) logger.debug("chunk: %r" % data[:256])
chunks.append(data) chunks.append(data)
if util._py3k: if _util._py3k:
# Join using b'' or '', as appropriate # Join using b'' or '', as appropriate
result.data = type(data)().join(chunks) result.data = type(data)().join(chunks)
else: else:
@ -399,8 +317,8 @@ class GPG(object):
else: else:
stdin = p.stdin stdin = p.stdin
if passphrase: if passphrase:
_write_passphrase(stdin, passphrase, self.encoding) _util._write_passphrase(stdin, passphrase, self.encoding)
writer = _threaded_copy_data(file, stdin) writer = _util._threaded_copy_data(file, stdin)
self._collect_output(p, result, writer, stdin) self._collect_output(p, result, writer, stdin)
return result return result
@ -411,8 +329,8 @@ class GPG(object):
"""Create a signature for a message or file.""" """Create a signature for a message or file."""
if isinstance(message, file): if isinstance(message, file):
result = self._sign_file(message, **kwargs) result = self._sign_file(message, **kwargs)
elif not util._is_stream(message): elif not _util._is_stream(message):
f = util._make_binary_stream(message, self.encoding) f = _util._make_binary_stream(message, self.encoding)
result = self._sign_file(f, **kwargs) result = self._sign_file(f, **kwargs)
f.close() f.close()
else: else:
@ -448,8 +366,8 @@ class GPG(object):
try: try:
stdin = p.stdin stdin = p.stdin
if passphrase: if passphrase:
_write_passphrase(stdin, passphrase, self.encoding) _util._write_passphrase(stdin, passphrase, self.encoding)
writer = _threaded_copy_data(file, stdin) writer = _util._threaded_copy_data(file, stdin)
except IOError: except IOError:
logger.exception("_sign_file(): Error writing message") logger.exception("_sign_file(): Error writing message")
writer = None writer = None
@ -471,7 +389,7 @@ class GPG(object):
>>> assert verify >>> assert verify
""" """
f = util._make_binary_stream(data, self.encoding) f = _util._make_binary_stream(data, self.encoding)
result = self.verify_file(f) result = self.verify_file(f)
f.close() f.close()
return result return result
@ -483,7 +401,7 @@ class GPG(object):
detached signature should be specified as the ``sig_file``. detached signature should be specified as the ``sig_file``.
:param file file: A file descriptor object. Its type will be checked :param file file: A file descriptor object. Its type will be checked
with :func:`util._is_file`. with :func:`_util._is_file`.
:param str sig_file: A file containing the GPG signature data for :param str sig_file: A file containing the GPG signature data for
``file``. If given, ``file`` is verified via this ``file``. If given, ``file`` is verified via this
detached signature. detached signature.
@ -496,10 +414,10 @@ class GPG(object):
logger.debug("verify_file(): Handling embedded signature") logger.debug("verify_file(): Handling embedded signature")
args = ["--verify"] args = ["--verify"]
proc = self._open_subprocess(args) proc = self._open_subprocess(args)
writer = _threaded_copy_data(file, proc.stdin) writer = _util._threaded_copy_data(file, proc.stdin)
self._collect_output(proc, result, writer, stdin=proc.stdin) self._collect_output(proc, result, writer, stdin=proc.stdin)
else: else:
if not util._is_file(sig_file): if not _util._is_file(sig_file):
logger.debug("verify_file(): '%r' is not a file" % sig_file) logger.debug("verify_file(): '%r' is not a file" % sig_file)
return result return result
logger.debug('verify_file(): Handling detached verification') logger.debug('verify_file(): Handling detached verification')
@ -508,7 +426,7 @@ class GPG(object):
sig_fh = open(sig_file) sig_fh = open(sig_file)
args = ["--verify %s - " % sig_fh.name] args = ["--verify %s - " % sig_fh.name]
proc = self._open_subprocess(args) proc = self._open_subprocess(args)
writer = _threaded_copy_data(file, proc.stdin) writer = _util._threaded_copy_data(file, proc.stdin)
self._collect_output(proc, result, stdin=proc.stdin) self._collect_output(proc, result, stdin=proc.stdin)
finally: finally:
if sig_fh and not sig_fh.closed: if sig_fh and not sig_fh.closed:
@ -568,7 +486,7 @@ class GPG(object):
result = self._result_map['import'](self) result = self._result_map['import'](self)
logger.debug('import_keys: %r', key_data[:256]) logger.debug('import_keys: %r', key_data[:256])
data = util._make_binary_stream(key_data, self.encoding) data = _util._make_binary_stream(key_data, self.encoding)
self._handle_io(['--import'], data, result, binary=True) self._handle_io(['--import'], data, result, binary=True)
logger.debug('import_keys result: %r', result.__dict__) logger.debug('import_keys result: %r', result.__dict__)
data.close() data.close()
@ -587,7 +505,7 @@ class GPG(object):
safe_keyserver = _fix_unsafe(keyserver) safe_keyserver = _fix_unsafe(keyserver)
result = self._result_map['import'](self) result = self._result_map['import'](self)
data = util._make_binary_stream("", self.encoding) data = _util._make_binary_stream("", self.encoding)
args = ['--keyserver', keyserver, '--recv-keys'] args = ['--keyserver', keyserver, '--recv-keys']
if keyids: if keyids:
@ -606,7 +524,7 @@ class GPG(object):
which='key' which='key'
if secret: if secret:
which='secret-key' which='secret-key'
if util._is_list_or_tuple(fingerprints): if _util._is_list_or_tuple(fingerprints):
fingerprints = ' '.join(fingerprints) fingerprints = ' '.join(fingerprints)
args = ['--batch --delete-%s "%s"' % (which, fingerprints)] args = ['--batch --delete-%s "%s"' % (which, fingerprints)]
result = self._result_map['delete'](self) result = self._result_map['delete'](self)
@ -619,7 +537,7 @@ class GPG(object):
which='' which=''
if secret: if secret:
which='-secret-key' which='-secret-key'
if util._is_list_or_tuple(keyids): if _util._is_list_or_tuple(keyids):
keyids = ' '.join(['"%s"' % k for k in keyids]) keyids = ' '.join(['"%s"' % k for k in keyids])
args = ["--armor --export%s %s" % (which, keyids)] args = ["--armor --export%s %s" % (which, keyids)]
p = self._open_subprocess(args) p = self._open_subprocess(args)
@ -650,7 +568,8 @@ class GPG(object):
which='public-keys' which='public-keys'
if secret: if secret:
which='secret-keys' which='secret-keys'
args = "--list-%s --fixed-list-mode --fingerprint --with-colons --no-show-photos" % (which,) args = "--list-%s --fixed-list-mode --fingerprint " % (which,)
args += "--with-colons --list-options no-show-photos"
args = [args] args = [args]
p = self._open_subprocess(args) p = self._open_subprocess(args)
@ -689,9 +608,8 @@ class GPG(object):
raise NotImplemented("Functionality for '--list-sigs' not implemented.") raise NotImplemented("Functionality for '--list-sigs' not implemented.")
def gen_key(self, input): def gen_key(self, input):
""" """Generate a GnuPG key through batch file key generation. See
Generate a key; you might use gen_key_input() to create the control :meth:`GPG.gen_key_input()` for creating the control input.
input.
>>> gpg = GPG(gpghome="keys") >>> gpg = GPG(gpghome="keys")
>>> input = gpg.gen_key_input() >>> input = gpg.gen_key_input()
@ -700,24 +618,77 @@ class GPG(object):
>>> result = gpg.gen_key('foo') >>> result = gpg.gen_key('foo')
>>> assert not result >>> assert not result
:param dict input: A dictionary of parameters and values for the new
key.
:returns: The result mapping with details of the new key, which is a
:class:`parsers.GenKey <GenKey>` object.
""" """
args = ["--gen-key --batch"] args = ["--gen-key --batch"]
key = self._result_map['generate'](self) key = self._result_map['generate'](self)
f = util._make_binary_stream(input, self.encoding) f = _util._make_binary_stream(input, self.encoding)
self._handle_io(args, f, key, binary=True) self._handle_io(args, f, key, binary=True)
f.close() f.close()
return key return key
def gen_key_input(self, **kwargs): def gen_key_input(self, **kwargs):
"""Generate GnuPG key(s) through batch file key generation. """Generate a batch file for input to :meth:`GPG.gen_key()`.
The GnuPG batch file key generation feature allows unattended key The GnuPG batch file key generation feature allows unattended key
generation by creating a file with special syntax and then providing it generation by creating a file with special syntax and then providing it
to: gpg --gen-key --batch <batch file> to: ``gpg --gen-key --batch``:
Key-Type: RSA
Key-Length: 4096
Name-Real: Autogenerated Key
Name-Email: %s@%s
Expire-Date: 2014-04-01
%pubring foo.gpg
%secring sec.gpg
%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
see http://www.gnupg.org/documentation/manuals/gnupg-devel/Unattended-GPG-key-generation.html#Unattended-GPG-key-generation see http://www.gnupg.org/documentation/manuals/gnupg-devel/Unattended-GPG-key-generation.html#Unattended-GPG-key-generation
for more details. for more details.
>>> gpg = GPG(gpghome="keys")
>>> params = {'name_real':'python-gnupg tester', 'name_email':'test@ing'}
>>> key_input = gpg.gen_key_input(**params)
>>> result = gpg.gen_key(input)
>>> assert result
:param str name_real: The uid name for the generated key.
:param str name_email: The uid email for the generated key. (default:
$USERNAME@$HOSTNAME)
:param str name_comment: The comment in the uid of the generated key.
:param str key_type: One of 'RSA', 'DSA', or 'ELG-E'. (default: 'RSA')
:param int key_length: The length in bytes of the new key.
(default: 4096)
:param str subkey_type: If ``key_type`` is 'RSA', an additional subkey
can be generated, and it's type must also be 'RSA'. If ``key_type``
is 'DSA', then the only subkey type which can be generated is
'ELG-E'.
:param int subkey_length: The length in bytes of the new subkey.
:type expire: int or str
:param expire: If an integer, the number of days before the key will
expire; if 0, the key will not expire. Otherwise, this can be given
as a string in the form <n>w or <n>m or <n>y, i.e. "5m" would mean
that the key will expire in five months, "1w" would expire in one
week, and "3y" would expire in three years. (default: "1y")
:param str passphrase: The passphrase for the new key.
""" """
parms = {} parms = {}
for key, val in list(kwargs.items()): for key, val in list(kwargs.items()):
key = key.replace('_','-').title() key = key.replace('_','-').title()
@ -726,7 +697,7 @@ class GPG(object):
parms.setdefault('Key-Type', 'RSA') parms.setdefault('Key-Type', 'RSA')
parms.setdefault('Key-Length', 4096) parms.setdefault('Key-Length', 4096)
parms.setdefault('Name-Real', "Autogenerated Key") parms.setdefault('Name-Real', "Autogenerated Key")
parms.setdefault('Name-Comment', "Generated by python-gnupg") parms.setdefault('Expire-Date', _util._next_year())
try: try:
logname = os.environ['LOGNAME'] logname = os.environ['LOGNAME']
except KeyError: except KeyError:
@ -734,6 +705,7 @@ class GPG(object):
hostname = socket.gethostname() hostname = socket.gethostname()
parms.setdefault('Name-Email', "%s@%s" parms.setdefault('Name-Email', "%s@%s"
% (logname.replace(' ', '_'), hostname)) % (logname.replace(' ', '_'), hostname))
out = "Key-Type: %s\n" % parms.pop('Key-Type') out = "Key-Type: %s\n" % parms.pop('Key-Type')
for key, val in list(parms.items()): for key, val in list(parms.items()):
out += "%s: %s\n" % (key, val) out += "%s: %s\n" % (key, val)
@ -742,28 +714,6 @@ class GPG(object):
out += "%commit\n" out += "%commit\n"
return out 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 # ENCRYPTION
# #
@ -805,7 +755,7 @@ class GPG(object):
args = ['--symmetric'] args = ['--symmetric']
else: else:
args = ['--encrypt'] args = ['--encrypt']
if not util._is_list_or_tuple(recipients): if not _util._is_list_or_tuple(recipients):
recipients = (recipients,) recipients = (recipients,)
for recipient in recipients: for recipient in recipients:
args.append('--recipient "%s"' % recipient) args.append('--recipient "%s"' % recipient)
@ -861,7 +811,7 @@ class GPG(object):
>>> assert result.fingerprint == print1 >>> assert result.fingerprint == print1
""" """
data = util._make_binary_stream(data, self.encoding) data = _util._make_binary_stream(data, self.encoding)
result = self.encrypt_file(data, recipients, **kwargs) result = self.encrypt_file(data, recipients, **kwargs)
data.close() data.close()
return result return result
@ -871,7 +821,7 @@ class GPG(object):
:param message: A string or file-like object to decrypt. :param message: A string or file-like object to decrypt.
""" """
data = util._make_binary_stream(message, self.encoding) data = _util._make_binary_stream(message, self.encoding)
result = self.decrypt_file(data, **kwargs) result = self.decrypt_file(data, **kwargs)
data.close() data.close()
return result return result
@ -965,7 +915,7 @@ class GPGWrapper(GPG):
"""Send keys to a keyserver.""" """Send keys to a keyserver."""
result = self._result_map['list'](self) result = self._result_map['list'](self)
gnupg.logger.debug('send_keys: %r', keyids) gnupg.logger.debug('send_keys: %r', keyids)
data = gnupg.util._make_binary_stream("", self.encoding) data = gnupg._util._make_binary_stream("", self.encoding)
args = ['--keyserver', keyserver, '--send-keys'] args = ['--keyserver', keyserver, '--send-keys']
args.extend(keyids) args.extend(keyids)
self._handle_io(args, data, result, binary=True) self._handle_io(args, data, result, binary=True)
@ -985,7 +935,7 @@ class GPGWrapper(GPG):
args.append('--cipher-algo %s' % cipher_algo) args.append('--cipher-algo %s' % cipher_algo)
else: else:
args = ['--encrypt'] args = ['--encrypt']
if not util._is_list_or_tuple(recipients): if not _util._is_list_or_tuple(recipients):
recipients = (recipients,) recipients = (recipients,)
for recipient in recipients: for recipient in recipients:
args.append('--recipient "%s"' % recipient) args.append('--recipient "%s"' % recipient)
@ -1008,7 +958,7 @@ class GPGWrapper(GPG):
args = ["--list-packets"] args = ["--list-packets"]
result = self._result_map['list-packets'](self) result = self._result_map['list-packets'](self)
self._handle_io(args, self._handle_io(args,
util._make_binary_stream(raw_data, self.encoding), _util._make_binary_stream(raw_data, self.encoding),
result) result)
return result return result

2
gnupg/parsers.py
View File

@ -275,7 +275,7 @@ def _is_allowed(input):
['--list-keys', '--list-key', '--fixed-list-mode', ['--list-keys', '--list-key', '--fixed-list-mode',
'--list-secret-keys', '--list-public-keys', '--list-secret-keys', '--list-public-keys',
'--list-packets', '--with-colons', '--list-packets', '--with-colons',
'--no-show-photos', '--list-options',
'--delete-keys', '--delete-secret-keys', '--delete-keys', '--delete-secret-keys',
'--encrypt', '--encrypt-files', '--encrypt', '--encrypt-files',
'--decrypt', '--decrypt-files', '--decrypt', '--decrypt-files',

64
gnupg/tests/test_gnupg.py
View File

@ -11,7 +11,6 @@ import argparse
import doctest import doctest
import logging import logging
from functools import wraps from functools import wraps
import inspect
import io import io
import os import os
import shutil import shutil
@ -30,24 +29,24 @@ from gnupg import parsers
from gnupg import util from gnupg import util
__author__ = gnupg.__author__ __author__ = gnupg.__author__
__date__ = gnupg.__date__
__version__ = gnupg.__version__ __version__ = gnupg.__version__
REPO_DIR = os.getcwd()
HOME_DIR = os.path.join(REPO_DIR, 'keys')
tempfile.tempdir = os.path.join(REPO_DIR, 'tmp_test') logger = logging.getLogger('gnupg')
_here = os.path.join(os.path.join(util._repo, 'gnupg'), 'tests')
_files = os.path.join(_here, 'files')
tempfile.tempdir = os.path.join(_here, 'tmp_test')
if not os.path.isdir(tempfile.gettempdir()): if not os.path.isdir(tempfile.gettempdir()):
os.mkdir(tempfile.gettempdir()) os.mkdir(tempfile.gettempdir())
HOME_DIR = tempfile.tempdir
@wraps(tempfile.TemporaryFile) @wraps(tempfile.TemporaryFile)
def _make_tempfile(*args, **kwargs): def _make_tempfile(*args, **kwargs):
return tempfile.TemporaryFile(dir=tempfile.gettempdir(), return tempfile.TemporaryFile(dir=tempfile.gettempdir(),
*args, **kwargs) *args, **kwargs)
logger = logging.getLogger('gnupg')
_here = os.path.join(os.path.join(util._repo, 'gnupg'), 'tests')
_files = os.path.join(_here, 'files')
KEYS_TO_IMPORT = """-----BEGIN PGP PUBLIC KEY BLOCK----- KEYS_TO_IMPORT = """-----BEGIN PGP PUBLIC KEY BLOCK-----
Version: GnuPG v1.4.9 (MingW32) Version: GnuPG v1.4.9 (MingW32)
@ -118,11 +117,10 @@ def compare_keys(k1, k2):
class ResultStringIO(io.StringIO): class ResultStringIO(io.StringIO):
def __init__(self): def __init__(self, init_string):
super(self, io.StringIO).__init__() super(ResultStringIO, self).__init__(init_string)
def write(self, data): def write(self, data):
super(self, io.StringIO).write(unicode(data)) super(ResultStringIO, self).write(unicode(data))
class GPGTestCase(unittest.TestCase): class GPGTestCase(unittest.TestCase):
@ -216,7 +214,7 @@ class GPGTestCase(unittest.TestCase):
cmd = self.gpg._make_args(None, False) cmd = self.gpg._make_args(None, False)
expected = ['/usr/bin/gpg', expected = ['/usr/bin/gpg',
'--status-fd 2 --no-tty', '--status-fd 2 --no-tty',
'--homedir "%s"' % os.path.join(os.getcwd(), 'keys'), '--homedir "%s"' % HOME_DIR,
'--no-default-keyring --keyring %s' % self.pubring, '--no-default-keyring --keyring %s' % self.pubring,
'--secret-keyring %s' % self.secring] '--secret-keyring %s' % self.secring]
self.assertListEqual(cmd, expected) self.assertListEqual(cmd, expected)
@ -244,16 +242,19 @@ class GPGTestCase(unittest.TestCase):
"Empty list expected...got instead: %s" "Empty list expected...got instead: %s"
% str(private_keys)) % str(private_keys))
def test_copy_data_bytesio(self):
def test_copy_data(self): """Test that _copy_data() is able to duplicate byte streams."""
""" message = "This is a BytesIO string string in memory."
XXX implement me instream = io.BytesIO(message)
XXX add me to a test suite self.assertEqual(unicode(message), instream.getvalue())
outstream = ResultStringIO(u'result:')
Test that _copy_data() is able to duplicate byte streams. copied = outstream
""" util._copy_data(instream, outstream)
instream = io.BytesIO("This is a string of bytes mapped in memory.") self.assertTrue(outstream.readable())
outstream = str("And this one is just a string.") self.assertTrue(outstream.closed)
self.assertFalse(instream.closed)
self.assertTrue(copied.closed)
#self.assertEqual(instream.getvalue()[6:], outstream.getvalue())
def generate_key_input(self, real_name, email_domain, key_length=None, def generate_key_input(self, real_name, email_domain, key_length=None,
key_type=None, subkey_type=None, passphrase=None): key_type=None, subkey_type=None, passphrase=None):
@ -386,9 +387,9 @@ class GPGTestCase(unittest.TestCase):
def test_key_generation_with_empty_value(self): def test_key_generation_with_empty_value(self):
"""Test that key generation handles empty values.""" """Test that key generation handles empty values."""
params = {'name_comment': ' '} params = {'name_real': ' '}
batch = self.gpg.gen_key_input(**params) batch = self.gpg.gen_key_input(**params)
self.assertTrue('\nName-Comment: Generated by python-gnupg\n' in batch) self.assertTrue('\nName-Real: Autogenerated Key\n' in batch)
def test_key_generation_override_default_value(self): def test_key_generation_override_default_value(self):
"""Test that overriding a default value in gen_key_input() works.""" """Test that overriding a default value in gen_key_input() works."""
@ -399,8 +400,9 @@ class GPGTestCase(unittest.TestCase):
def test_list_keys_after_generation(self): def test_list_keys_after_generation(self):
"""Test that after key generation, the generated key is available.""" """Test that after key generation, the generated key is available."""
self.test_list_keys_initial() self.test_list_keys_initial_public()
self.do_key_generation() self.test_list_keys_initial_secret()
self.generate_key("Johannes Trithemius", 'iusedcarrierpidgeons@inste.ad')
public_keys = self.gpg.list_keys() public_keys = self.gpg.list_keys()
self.assertTrue(is_list_with_len(public_keys, 1), self.assertTrue(is_list_with_len(public_keys, 1),
"1-element list expected") "1-element list expected")
@ -670,7 +672,8 @@ class GPGTestCase(unittest.TestCase):
suites = { 'parsers': set(['test_parsers_fix_unsafe', suites = { 'parsers': set(['test_parsers_fix_unsafe',
'test_parsers_is_hex_valid', 'test_parsers_is_hex_valid',
'test_parsers_is_hex_invalid',]), 'test_parsers_is_hex_invalid',
'test_copy_data_bytesio',]),
'basic': set(['test_gpghome_creation', 'basic': set(['test_gpghome_creation',
'test_gpg_binary', 'test_gpg_binary',
'test_gpg_binary_not_abs', 'test_gpg_binary_not_abs',
@ -768,11 +771,6 @@ if __name__ == "__main__":
suite_names = list() suite_names = list()
for name, methodset in suites.items(): for name, methodset in suites.items():
suite_names.append(name) suite_names.append(name)
this_file = inspect.getfile(inspect.currentframe()).split('.', 1)[0]
#mod = getattr(this_file, '__dict__', None)
#func = getattr(gnupg.__module__, '__setattr__', None)
#if func is not None:
# func(name, list(methodset))
setattr(GPGTestCase, name, list(methodset)) setattr(GPGTestCase, name, list(methodset))
parser = argparse.ArgumentParser(description="Unittests for python-gnupg") parser = argparse.ArgumentParser(description="Unittests for python-gnupg")

134
gnupg/util.py
View File

@ -30,6 +30,11 @@ from datetime import datetime
import logging import logging
import os import os
import time
import random
import string
import sys
import threading
try: try:
from io import StringIO from io import StringIO
@ -62,6 +67,54 @@ _ugpg = os.path.join(_user, '.gnupg') ## $HOME/.gnupg
_conf = os.path.join(os.path.join(_user, '.config'), _conf = os.path.join(os.path.join(_user, '.config'),
'python-gnupg') ## $HOME/.config/python-gnupg 'python-gnupg') ## $HOME/.config/python-gnupg
def _copy_data(instream, outstream):
"""Copy data from one stream to another.
:type instream: :class:`io.BytesIO` or :class:`io.StringIO` or file
:param instream: A byte stream or open file to read from.
:param file outstream: The file descriptor of a tmpfile to write to.
"""
sent = 0
#try:
# #assert (util._is_stream(instream)
# # or isinstance(instream, file)), "instream not stream or file"
# assert isinstance(outstream, file), "outstream is not a file"
#except AssertionError as ae:
# logger.exception(ae)
# return
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("_copy_data(): sending chunk (%d):\n%s" % (sent, data[:256]))
try:
outstream.write(data)
except UnicodeError:
try:
outstream.write(data.encode(enc))
except IOError:
logger.exception('_copy_data(): Error sending data: Broken pipe')
break
except IOError:
# Can get 'broken pipe' errors even when all data was sent
logger.exception('_copy_data(): Error sending data: Broken pipe')
break
try:
outstream.close()
except IOError:
logger.exception('_copy_data(): Got IOError while closing %s' % outstream)
else:
logger.debug("_copy_data(): Closed output, %d bytes sent." % sent)
def _create_gpghome(gpghome): def _create_gpghome(gpghome):
"""Create the specified GnuPG home directory, if necessary. """Create the specified GnuPG home directory, if necessary.
@ -181,15 +234,78 @@ def _make_binary_stream(s, encoding):
rv = StringIO(s) rv = StringIO(s)
return rv return rv
## xxx unused function? def _make_passphrase(length=None, save=False, file=None):
def _today(): """Create a passphrase and write it to a file that only the user can read.
"""Get the current date.
This is not very secure, and should not be relied upon for actual key
passphrases.
:param int length: The length in bytes of the string to generate.
:param file file: The file to save the generated passphrase in. If not
given, defaults to 'passphrase-<the real user id>-<seconds since
epoch>' in the top-level directory.
"""
if not length:
length = 40
passphrase = _make_random_string(length)
if save:
ruid, euid, suid = os.getresuid()
gid = os.getgid()
now = time.mktime(time.gmtime())
if not file:
filename = str('passphrase-%s-%s' % uid, now)
file = os.path.join(_repo, filename)
with open(file, 'a') as fh:
fh.write(passphrase)
fh.flush()
fh.close()
os.chmod(file, 0600)
os.chown(file, ruid, gid)
logger.warn("Generated passphrase saved to %s" % file)
return passphrase
def _make_random_string(length):
"""Returns a random lowercase, uppercase, alphanumerical string.
:param int length: The length in bytes of the string to generate.
"""
chars = string.ascii_lowercase + string.ascii_uppercase + string.digits
return ''.join(random.choice(chars) for x in range(length))
def _next_year():
"""Get the date of today plus one year.
:rtype: str :rtype: str
:returns: The date, in the format '%Y-%m-%d'. :returns: The date of this day next year, in the format '%Y-%m-%d'.
""" """
now_string = datetime.now().__str__() now = datetime.now().__str__()
return now_string.split(' ', 1)[0] date = now.split(' ', 1)[0]
year, month, day = date.split('-', 2)
next_year = str(int(year)+1)
return '-'.join((next_year, month, day))
def _threaded_copy_data(instream, outstream):
"""Copy data from one stream to another in a separate thread.
Wraps ``_copy_data()`` in a :class:`threading.Thread`.
:type instream: :class:`io.BytesIO` or :class:`io.StringIO`
:param instream: A byte stream to read from.
:param file outstream: The file descriptor of a tmpfile to write to.
"""
copy_thread = threading.Thread(target=_copy_data,
args=(instream, outstream))
copy_thread.setDaemon(True)
logger.debug('_threaded_copy_data(): %r, %r, %r', copy_thread,
instream, outstream)
copy_thread.start()
return copy_thread
def _which(executable, flags=os.X_OK): def _which(executable, flags=os.X_OK):
"""Borrowed from Twisted's :mod:twisted.python.proutils . """Borrowed from Twisted's :mod:twisted.python.proutils .
@ -228,3 +344,9 @@ def _which(executable, flags=os.X_OK):
if os.access(pext, flags): if os.access(pext, flags):
result.append(pext) result.append(pext)
return result return result
def _write_passphrase(stream, passphrase, encoding):
passphrase = '%s\n' % passphrase
passphrase = passphrase.encode(encoding)
stream.write(passphrase)
logger.debug("_write_passphrase(): Wrote passphrase.")