[test] add updater tests
[keymanager.git] / src / leap / keymanager / openpgp.py
1 # -*- coding: utf-8 -*-
2 # openpgp.py
3 # Copyright (C) 2013-2015 LEAP
4 #
5 # This program is free software: you can redistribute it and/or modify
6 # it under the terms of the GNU General Public License as published by
7 # the Free Software Foundation, either version 3 of the License, or
8 # (at your option) any later version.
9 #
10 # This program is distributed in the hope that it will be useful,
11 # but WITHOUT ANY WARRANTY; without even the implied warranty of
12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13 # GNU General Public License for more details.
14 #
15 # You should have received a copy of the GNU General Public License
16 # along with this program. If not, see <http://www.gnu.org/licenses/>.
17 """
18 Infrastructure for using OpenPGP keys in Key Manager.
19 """
20 import logging
21 import os
22 import re
23 import shutil
24 import tempfile
25 import io
26
27
28 from datetime import datetime
29 from multiprocessing import cpu_count
30 from gnupg import GPG
31 from gnupg.gnupg import GPGUtilities
32 from twisted.internet import defer
33 from twisted.internet.threads import deferToThread
34
35 from leap.common.check import leap_assert, leap_assert_type, leap_check
36 from leap.keymanager import errors
37 from leap.keymanager.keys import (
38     EncryptionKey,
39     EncryptionScheme,
40     is_address,
41     build_key_from_dict,
42     TYPE_FINGERPRINT_PRIVATE_INDEX,
43     TYPE_ADDRESS_PRIVATE_INDEX,
44     KEY_ADDRESS_KEY,
45     KEY_FINGERPRINT_KEY,
46     KEYMANAGER_ACTIVE_TYPE,
47 )
48
49
50 logger = logging.getLogger(__name__)
51
52
53 #
54 # A temporary GPG keyring wrapped to provide OpenPGP functionality.
55 #
56
57 # This function will be used to call blocking GPG functions outside
58 # of Twisted reactor and match the concurrent calls to the amount of CPU cores
59 cpu_core_semaphore = defer.DeferredSemaphore(cpu_count())
60
61
62 def from_thread(func, *args, **kwargs):
63     call = lambda: deferToThread(func, *args, **kwargs)
64     return cpu_core_semaphore.run(call)
65
66
67 class TempGPGWrapper(object):
68     """
69     A context manager that wraps a temporary GPG keyring which only contains
70     the keys given at object creation.
71     """
72
73     def __init__(self, keys=None, gpgbinary=None):
74         """
75         Create an empty temporary keyring and import any given C{keys} into
76         it.
77
78         :param keys: OpenPGP key, or list of.
79         :type keys: OpenPGPKey or list of OpenPGPKeys
80         :param gpgbinary: Name for GnuPG binary executable.
81         :type gpgbinary: C{str}
82         """
83         self._gpg = None
84         self._gpgbinary = gpgbinary
85         if not keys:
86             keys = list()
87         if not isinstance(keys, list):
88             keys = [keys]
89         self._keys = keys
90         for key in keys:
91             leap_assert_type(key, OpenPGPKey)
92
93     def __enter__(self):
94         """
95         Build and return a GPG keyring containing the keys given on
96         object creation.
97
98         :return: A GPG instance containing the keys given on object creation.
99         :rtype: gnupg.GPG
100         """
101         self._build_keyring()
102         return self._gpg
103
104     def __exit__(self, exc_type, exc_value, traceback):
105         """
106         Ensure the gpg is properly destroyed.
107         """
108         # TODO handle exceptions and log here
109         self._destroy_keyring()
110
111     def _build_keyring(self):
112         """
113         Create a GPG keyring containing the keys given on object creation.
114
115         :return: A GPG instance containing the keys given on object creation.
116         :rtype: gnupg.GPG
117         """
118         privkeys = [key for key in self._keys if key and key.private is True]
119         publkeys = [key for key in self._keys if key and key.private is False]
120         # here we filter out public keys that have a correspondent
121         # private key in the list because the private key_data by
122         # itself is enough to also have the public key in the keyring,
123         # and we want to count the keys afterwards.
124
125         privfps = map(lambda privkey: privkey.fingerprint, privkeys)
126         publkeys = filter(
127             lambda pubkey: pubkey.fingerprint not in privfps, publkeys)
128
129         listkeys = lambda: self._gpg.list_keys()
130         listsecretkeys = lambda: self._gpg.list_keys(secret=True)
131
132         self._gpg = GPG(binary=self._gpgbinary,
133                         homedir=tempfile.mkdtemp())
134         leap_assert(len(listkeys()) is 0, 'Keyring not empty.')
135
136         # import keys into the keyring:
137         # concatenating ascii-armored keys, which is correctly
138         # understood by GPG.
139
140         self._gpg.import_keys("".join(
141             [x.key_data for x in publkeys + privkeys]))
142
143         # assert the number of keys in the keyring
144         leap_assert(
145             len(listkeys()) == len(publkeys) + len(privkeys),
146             'Wrong number of public keys in keyring: %d, should be %d)' %
147             (len(listkeys()), len(publkeys) + len(privkeys)))
148         leap_assert(
149             len(listsecretkeys()) == len(privkeys),
150             'Wrong number of private keys in keyring: %d, should be %d)' %
151             (len(listsecretkeys()), len(privkeys)))
152
153     def _destroy_keyring(self):
154         """
155         Securely erase the keyring.
156         """
157         # TODO: implement some kind of wiping of data or a more
158         # secure way that
159         # does not write to disk.
160
161         try:
162             for secret in [True, False]:
163                 for key in self._gpg.list_keys(secret=secret):
164                     self._gpg.delete_keys(
165                         key['fingerprint'],
166                         secret=secret)
167             leap_assert(len(self._gpg.list_keys()) is 0, 'Keyring not empty!')
168
169         except:
170             raise
171
172         finally:
173             leap_assert(self._gpg.homedir != os.path.expanduser('~/.gnupg'),
174                         "watch out! Tried to remove default gnupg home!")
175             shutil.rmtree(self._gpg.homedir)
176
177
178 def _parse_address(address):
179     """
180     Remove name, '<', '>' and the identity suffix after the '+' until the '@'
181     e.g.: test_user+something@provider.com becomes test_user@provider.com
182     since the key belongs to the identity without the '+' suffix.
183
184     :type address: str
185     :rtype: str
186     """
187     mail_regex = '(.*<)?([\w.-]+)(\+.*)?(@[\w.-]+)(>.*)?'
188     match = re.match(mail_regex, address)
189     if match is None:
190         return None
191     return ''.join(match.group(2, 4))
192
193
194 #
195 # The OpenPGP wrapper
196 #
197
198 class OpenPGPKey(EncryptionKey):
199     """
200     Base class for OpenPGP keys.
201     """
202
203     def __init__(self, address, gpgbinary=None, **kwargs):
204         self._gpgbinary = gpgbinary
205         super(OpenPGPKey, self).__init__(address, **kwargs)
206
207     @property
208     def signatures(self):
209         """
210         Get the key signatures
211
212         :return: the key IDs that have signed the key
213         :rtype: list(str)
214         """
215         with TempGPGWrapper(keys=[self], gpgbinary=self._gpgbinary) as gpg:
216             res = gpg.list_sigs(self.fingerprint)
217             for uid, sigs in res.sigs.iteritems():
218                 if _parse_address(uid) in self.address:
219                     return sigs
220
221         return []
222
223     def merge(self, newkey):
224         if newkey.fingerprint != self.fingerprint:
225             logger.critical(
226                 "Can't put a key whith the same key_id and different "
227                 "fingerprint: %s, %s"
228                 % (newkey.fingerprint, self.fingerprint))
229             raise errors.KeyFingerprintMismatch(newkey.fingerprint)
230
231         with TempGPGWrapper(gpgbinary=self._gpgbinary) as gpg:
232             gpg.import_keys(self.key_data)
233             gpg.import_keys(newkey.key_data)
234             gpgkey = gpg.list_keys(secret=newkey.private).pop()
235
236             if gpgkey['expires']:
237                 self.expiry_date = datetime.fromtimestamp(
238                     int(gpgkey['expires']))
239             else:
240                 self.expiry_date = None
241
242             self.uids = []
243             for uid in gpgkey['uids']:
244                 self.uids.append(_parse_address(uid))
245
246             self.length = int(gpgkey['length'])
247             self.key_data = gpg.export_keys(gpgkey['fingerprint'],
248                                             secret=self.private)
249
250         if newkey.validation > self.validation:
251             self.validation = newkey.validation
252         if newkey.last_audited_at > self.last_audited_at:
253             self.validation = newkey.last_audited_at
254         self.encr_used = newkey.encr_used or self.encr_used
255         self.sign_used = newkey.sign_used or self.sign_used
256         self.refreshed_at = datetime.now()
257
258
259 class OpenPGPScheme(EncryptionScheme):
260     """
261     A wrapper for OpenPGP keys management and use (encryption, decyption,
262     signing and verification).
263     """
264
265     # type used on the soledad documents
266     KEY_TYPE = OpenPGPKey.__name__
267     ACTIVE_TYPE = KEY_TYPE + KEYMANAGER_ACTIVE_TYPE
268
269     def __init__(self, soledad, gpgbinary=None):
270         """
271         Initialize the OpenPGP wrapper.
272
273         :param soledad: A Soledad instance for key storage.
274         :type soledad: leap.soledad.Soledad
275         :param gpgbinary: Name for GnuPG binary executable.
276         :type gpgbinary: C{str}
277         """
278         EncryptionScheme.__init__(self, soledad)
279         self._wait_indexes("get_key", "put_key")
280         self._gpgbinary = gpgbinary
281
282     #
283     # Keys management
284     #
285
286     def gen_key(self, address):
287         """
288         Generate an OpenPGP keypair bound to C{address}.
289
290         :param address: The address bound to the key.
291         :type address: str
292
293         :return: A Deferred which fires with the key bound to address, or fails
294                  with KeyAlreadyExists if key already exists in local database.
295         :rtype: Deferred
296         """
297         # make sure the key does not already exist
298         leap_assert(is_address(address), 'Not an user address: %s' % address)
299
300         @defer.inlineCallbacks
301         def _gen_key(_):
302             with TempGPGWrapper(gpgbinary=self._gpgbinary) as gpg:
303                 # TODO: inspect result, or use decorator
304                 params = gpg.gen_key_input(
305                     key_type='RSA',
306                     key_length=4096,
307                     name_real=address,
308                     name_email=address,
309                     name_comment='')
310                 logger.info("About to generate keys... "
311                             "This might take SOME time.")
312                 yield from_thread(gpg.gen_key, params)
313                 logger.info("Keys for %s have been successfully "
314                             "generated." % (address,))
315                 pubkeys = gpg.list_keys()
316
317                 # assert for new key characteristics
318                 leap_assert(
319                     len(pubkeys) is 1,  # a unitary keyring!
320                     'Keyring has wrong number of keys: %d.' % len(pubkeys))
321                 key = gpg.list_keys(secret=True).pop()
322                 leap_assert(
323                     len(key['uids']) is 1,  # with just one uid!
324                     'Wrong number of uids for key: %d.' % len(key['uids']))
325                 uid_match = False
326                 for uid in key['uids']:
327                     if re.match('.*<%s>$' % address, uid) is not None:
328                         uid_match = True
329                         break
330                 leap_assert(uid_match, 'Key not correctly bound to address.')
331
332                 # insert both public and private keys in storage
333                 deferreds = []
334                 for secret in [True, False]:
335                     key = gpg.list_keys(secret=secret).pop()
336                     openpgp_key = self._build_key_from_gpg(
337                         key,
338                         gpg.export_keys(key['fingerprint'], secret=secret))
339                     d = self.put_key(openpgp_key, address)
340                     deferreds.append(d)
341                 yield defer.gatherResults(deferreds)
342
343         def key_already_exists(_):
344             raise errors.KeyAlreadyExists(address)
345
346         d = self.get_key(address)
347         d.addCallbacks(key_already_exists, _gen_key)
348         d.addCallback(lambda _: self.get_key(address, private=True))
349         return d
350
351     def get_key(self, address, private=False):
352         """
353         Get key bound to C{address} from local storage.
354
355         :param address: The address bound to the key.
356         :type address: str
357         :param private: Look for a private key instead of a public one?
358         :type private: bool
359
360         :return: A Deferred which fires with the OpenPGPKey bound to address,
361                  or which fails with KeyNotFound if the key was not found on
362                  local storage.
363         :rtype: Deferred
364         """
365         address = _parse_address(address)
366
367         def build_key((keydoc, activedoc)):
368             if keydoc is None:
369                 raise errors.KeyNotFound(address)
370             leap_assert(
371                 address in keydoc.content[KEY_ADDRESS_KEY],
372                 'Wrong address in key %s. Expected %s, found %s.'
373                 % (keydoc.content[KEY_FINGERPRINT_KEY], address,
374                    keydoc.content[KEY_ADDRESS_KEY]))
375             key = build_key_from_dict(OpenPGPKey, keydoc.content,
376                                       activedoc.content)
377             key._gpgbinary = self._gpgbinary
378             return key
379
380         d = self._get_key_doc(address, private)
381         d.addCallback(build_key)
382         return d
383
384     def parse_ascii_key(self, key_data):
385         """
386         Parses an ascii armored key (or key pair) data and returns
387         the OpenPGPKey keys.
388
389         :param key_data: the key data to be parsed.
390         :type key_data: str or unicode
391
392         :returns: the public key and private key (if applies) for that data.
393         :rtype: (public, private) -> tuple(OpenPGPKey, OpenPGPKey)
394                 the tuple may have one or both components None
395         """
396         leap_assert_type(key_data, (str, unicode))
397         # TODO: add more checks for correct key data.
398         leap_assert(key_data is not None, 'Data does not represent a key.')
399
400         priv_info, privkey = process_ascii_key(
401             key_data, self._gpgbinary, secret=True)
402         pub_info, pubkey = process_ascii_key(
403             key_data, self._gpgbinary, secret=False)
404
405         if not pubkey:
406             return (None, None)
407
408         openpgp_privkey = None
409         if privkey:
410             # build private key
411             openpgp_privkey = self._build_key_from_gpg(priv_info, privkey)
412             leap_check(pub_info['fingerprint'] == priv_info['fingerprint'],
413                        'Fingerprints for public and private key differ.',
414                        errors.KeyFingerprintMismatch)
415         # build public key
416         openpgp_pubkey = self._build_key_from_gpg(pub_info, pubkey)
417
418         return (openpgp_pubkey, openpgp_privkey)
419
420     def put_ascii_key(self, key_data, address):
421         """
422         Put key contained in ascii-armored C{key_data} in local storage.
423
424         :param key_data: The key data to be stored.
425         :type key_data: str or unicode
426         :param address: address for which this key will be active
427         :type address: str
428
429         :return: A Deferred which fires when the OpenPGPKey is in the storage.
430         :rtype: Deferred
431         """
432         leap_assert_type(key_data, (str, unicode))
433
434         openpgp_privkey = None
435         try:
436             openpgp_pubkey, openpgp_privkey = self.parse_ascii_key(key_data)
437         except (errors.KeyAddressMismatch, errors.KeyFingerprintMismatch) as e:
438             return defer.fail(e)
439
440         def put_key(_, key):
441             return self.put_key(key, address)
442
443         d = defer.succeed(None)
444         if openpgp_pubkey is not None:
445             d.addCallback(put_key, openpgp_pubkey)
446         if openpgp_privkey is not None:
447             d.addCallback(put_key, openpgp_privkey)
448         return d
449
450     def put_key(self, key, address):
451         """
452         Put C{key} in local storage.
453
454         :param key: The key to be stored.
455         :type key: OpenPGPKey
456         :param address: address for which this key will be active.
457         :type address: str
458
459         :return: A Deferred which fires when the key is in the storage.
460         :rtype: Deferred
461         """
462         def merge_and_put((keydoc, activedoc)):
463             if not keydoc:
464                 return put_new_key(activedoc)
465
466             active_content = None
467             if activedoc:
468                 active_content = activedoc.content
469             oldkey = build_key_from_dict(OpenPGPKey, keydoc.content,
470                                          active_content)
471
472             key.merge(oldkey)
473             keydoc.set_json(key.get_json())
474             deferred_key = self._soledad.put_doc(keydoc)
475
476             active_json = key.get_active_json(address)
477             if activedoc:
478                 activedoc.set_json(active_json)
479                 deferred_active = self._soledad.put_doc(activedoc)
480             else:
481                 deferred_active = self._soledad.create_doc_from_json(
482                     active_json)
483
484             return defer.gatherResults([deferred_key, deferred_active])
485
486         def put_new_key(activedoc):
487             deferreds = []
488             if activedoc:
489                 d = self._soledad.delete_doc(activedoc)
490                 deferreds.append(d)
491             for json in [key.get_json(), key.get_active_json(address)]:
492                 d = self._soledad.create_doc_from_json(json)
493                 deferreds.append(d)
494             return defer.gatherResults(deferreds)
495
496         dk = self._get_key_doc_from_fingerprint(key.fingerprint, key.private)
497         da = self._get_active_doc_from_address(address, key.private)
498         d = defer.gatherResults([dk, da])
499         d.addCallback(merge_and_put)
500         return d
501
502     def _get_key_doc(self, address, private=False):
503         """
504         Get the document with a key (public, by default) bound to C{address}.
505
506         If C{private} is True, looks for a private key instead of a public.
507
508         :param address: The address bound to the key.
509         :type address: str
510         :param private: Whether to look for a private key.
511         :type private: bool
512
513         :return: A Deferred which fires with a touple of two SoledadDocument
514                  (keydoc, activedoc) or None if it does not exist.
515         :rtype: Deferred
516         """
517         def get_key_from_active_doc(activedoc):
518             if not activedoc:
519                 return (None, None)
520             fingerprint = activedoc.content[KEY_FINGERPRINT_KEY]
521             d = self._get_key_doc_from_fingerprint(fingerprint, private)
522             d.addCallback(delete_active_if_no_key, activedoc)
523             return d
524
525         def delete_active_if_no_key(keydoc, activedoc):
526             if not keydoc:
527                 d = self._soledad.delete_doc(activedoc)
528                 d.addCallback(lambda _: (None, None))
529                 return d
530             return (keydoc, activedoc)
531
532         d = self._get_active_doc_from_address(address, private)
533         d.addCallback(get_key_from_active_doc)
534         return d
535
536     def _build_key_from_gpg(self, key, key_data):
537         """
538         Build an OpenPGPKey for C{address} based on C{key} from
539         local gpg storage.
540
541         ASCII armored GPG key data has to be queried independently in this
542         wrapper, so we receive it in C{key_data}.
543
544         :param key: Key obtained from GPG storage.
545         :type key: dict
546         :param key_data: Key data obtained from GPG storage.
547         :type key_data: str
548         :return: An instance of the key.
549         :rtype: OpenPGPKey
550         """
551         return build_gpg_key(key, key_data, self._gpgbinary)
552
553     def delete_key(self, key):
554         """
555         Remove C{key} from storage.
556
557         :param key: The key to be removed.
558         :type key: EncryptionKey
559
560         :return: A Deferred which fires when the key is deleted, or which
561                  fails with KeyNotFound if the key was not found on local
562                  storage.
563         :rtype: Deferred
564         """
565         leap_assert_type(key, OpenPGPKey)
566
567         def delete_docs(activedocs):
568             deferreds = []
569             for doc in activedocs:
570                 d = self._soledad.delete_doc(doc)
571                 deferreds.append(d)
572             return defer.gatherResults(deferreds)
573
574         def get_key_docs(_):
575             return self._soledad.get_from_index(
576                 TYPE_FINGERPRINT_PRIVATE_INDEX,
577                 self.KEY_TYPE,
578                 key.fingerprint,
579                 '1' if key.private else '0')
580
581         def delete_key(docs):
582             if len(docs) == 0:
583                 raise errors.KeyNotFound(key)
584             elif len(docs) > 1:
585                 logger.warning("There is more than one key for fingerprint %s"
586                                % key.fingerprint)
587
588             has_deleted = False
589             deferreds = []
590             for doc in docs:
591                 if doc.content['fingerprint'] == key.fingerprint:
592                     d = self._soledad.delete_doc(doc)
593                     deferreds.append(d)
594                     has_deleted = True
595             if not has_deleted:
596                 raise errors.KeyNotFound(key)
597             return defer.gatherResults(deferreds)
598
599         d = self._soledad.get_from_index(
600             TYPE_FINGERPRINT_PRIVATE_INDEX,
601             self.ACTIVE_TYPE,
602             key.fingerprint,
603             '1' if key.private else '0')
604         d.addCallback(delete_docs)
605         d.addCallback(get_key_docs)
606         d.addCallback(delete_key)
607         return d
608
609     #
610     # Data encryption, decryption, signing and verifying
611     #
612
613     @staticmethod
614     def _assert_gpg_result_ok(result):
615         """
616         Check if GPG result is 'ok' and log stderr outputs.
617
618         :param result: GPG results, which have a field calld 'ok' that states
619                        whether the gpg operation was successful or not.
620         :type result: object
621
622         :raise GPGError: Raised when the gpg operation was not successful.
623         """
624         stderr = getattr(result, 'stderr', None)
625         if stderr:
626             logger.debug("%s" % (stderr,))
627         if getattr(result, 'ok', None) is not True:
628             raise errors.GPGError(
629                 'Failed to encrypt/decrypt: %s' % stderr)
630
631     @defer.inlineCallbacks
632     def encrypt(self, data, pubkey, passphrase=None, sign=None,
633                 cipher_algo='AES256'):
634         """
635         Encrypt C{data} using public @{pubkey} and sign with C{sign} key.
636
637         :param data: The data to be encrypted.
638         :type data: str
639         :param pubkey: The key used to encrypt.
640         :type pubkey: OpenPGPKey
641         :param sign: The key used for signing.
642         :type sign: OpenPGPKey
643         :param cipher_algo: The cipher algorithm to use.
644         :type cipher_algo: str
645
646         :return: A Deferred that will be fired with the encrypted data.
647         :rtype: defer.Deferred
648
649         :raise EncryptError: Raised if failed encrypting for some reason.
650         """
651         leap_assert_type(pubkey, OpenPGPKey)
652         leap_assert(pubkey.private is False, 'Key is not public.')
653         keys = [pubkey]
654         if sign is not None:
655             leap_assert_type(sign, OpenPGPKey)
656             leap_assert(sign.private is True)
657             keys.append(sign)
658         with TempGPGWrapper(keys, self._gpgbinary) as gpg:
659             result = yield from_thread(
660                 gpg.encrypt,
661                 data, pubkey.fingerprint,
662                 default_key=sign.fingerprint if sign else None,
663                 passphrase=passphrase, symmetric=False,
664                 cipher_algo=cipher_algo)
665             # Here we cannot assert for correctness of sig because the sig is
666             # in the ciphertext.
667             # result.ok    - (bool) indicates if the operation succeeded
668             # result.data  - (bool) contains the result of the operation
669             try:
670                 self._assert_gpg_result_ok(result)
671                 defer.returnValue(result.data)
672             except errors.GPGError as e:
673                 logger.error('Failed to decrypt: %s.' % str(e))
674                 raise errors.EncryptError()
675
676     @defer.inlineCallbacks
677     def decrypt(self, data, privkey, passphrase=None, verify=None):
678         """
679         Decrypt C{data} using private @{privkey} and verify with C{verify} key.
680
681         :param data: The data to be decrypted.
682         :type data: str
683         :param privkey: The key used to decrypt.
684         :type privkey: OpenPGPKey
685         :param passphrase: The passphrase for the secret key used for
686                            decryption.
687         :type passphrase: str
688         :param verify: The key used to verify a signature.
689         :type verify: OpenPGPKey
690
691         :return: Deferred that will fire with the decrypted data and
692                  if signature verifies (unicode, bool)
693         :rtype: Deferred
694
695         :raise DecryptError: Raised if failed decrypting for some reason.
696         """
697         leap_assert(privkey.private is True, 'Key is not private.')
698         keys = [privkey]
699         if verify is not None:
700             leap_assert_type(verify, OpenPGPKey)
701             leap_assert(verify.private is False)
702             keys.append(verify)
703         with TempGPGWrapper(keys, self._gpgbinary) as gpg:
704             try:
705                 result = yield from_thread(gpg.decrypt,
706                                            data, passphrase=passphrase,
707                                            always_trust=True)
708                 self._assert_gpg_result_ok(result)
709
710                 # verify signature
711                 sign_valid = False
712                 if (verify is not None and
713                         result.valid is True and
714                         verify.fingerprint == result.pubkey_fingerprint):
715                     sign_valid = True
716
717                 defer.returnValue((result.data, sign_valid))
718             except errors.GPGError as e:
719                 logger.error('Failed to decrypt: %s.' % str(e))
720                 raise errors.DecryptError(str(e))
721
722     def is_encrypted(self, data):
723         """
724         Return whether C{data} was asymmetrically encrypted using OpenPGP.
725
726         :param data: The data we want to know about.
727         :type data: str
728
729         :return: Whether C{data} was encrypted using this wrapper.
730         :rtype: bool
731         """
732         with TempGPGWrapper(gpgbinary=self._gpgbinary) as gpg:
733             gpgutil = GPGUtilities(gpg)
734             return gpgutil.is_encrypted_asym(data)
735
736     def sign(self, data, privkey, digest_algo='SHA512', clearsign=False,
737              detach=True, binary=False):
738         """
739         Sign C{data} with C{privkey}.
740
741         :param data: The data to be signed.
742         :type data: str
743
744         :param privkey: The private key to be used to sign.
745         :type privkey: OpenPGPKey
746         :param digest_algo: The hash digest to use.
747         :type digest_algo: str
748         :param clearsign: If True, create a cleartext signature.
749         :type clearsign: bool
750         :param detach: If True, create a detached signature.
751         :type detach: bool
752         :param binary: If True, do not ascii armour the output.
753         :type binary: bool
754
755         :return: The ascii-armored signed data.
756         :rtype: str
757         """
758         leap_assert_type(privkey, OpenPGPKey)
759         leap_assert(privkey.private is True)
760
761         # result.fingerprint - contains the fingerprint of the key used to
762         #                      sign.
763         with TempGPGWrapper(privkey, self._gpgbinary) as gpg:
764             result = gpg.sign(data, default_key=privkey.fingerprint,
765                               digest_algo=digest_algo, clearsign=clearsign,
766                               detach=detach, binary=binary)
767             rfprint = privkey.fingerprint
768             privkey = gpg.list_keys(secret=True).pop()
769             kfprint = privkey['fingerprint']
770             if result.fingerprint is None:
771                 raise errors.SignFailed(
772                     'Failed to sign with key %s: %s' %
773                     (privkey['fingerprint'], result.stderr))
774             leap_assert(
775                 result.fingerprint == kfprint,
776                 'Signature and private key fingerprints mismatch: '
777                 '%s != %s' % (rfprint, kfprint))
778         return result.data
779
780     def verify(self, data, pubkey, detached_sig=None):
781         """
782         Verify signed C{data} with C{pubkey}, eventually using
783         C{detached_sig}.
784
785         :param data: The data to be verified.
786         :type data: str
787         :param pubkey: The public key to be used on verification.
788         :type pubkey: OpenPGPKey
789         :param detached_sig: A detached signature. If given, C{data} is
790                              verified against this detached signature.
791         :type detached_sig: str
792
793         :return: signature matches
794         :rtype: bool
795         """
796         leap_assert_type(pubkey, OpenPGPKey)
797         leap_assert(pubkey.private is False)
798         with TempGPGWrapper(pubkey, self._gpgbinary) as gpg:
799             result = None
800             if detached_sig is None:
801                 result = gpg.verify(data)
802             else:
803                 # to verify using a detached sig we have to use
804                 # gpg.verify_file(), which receives the data as a binary
805                 # stream and the name of a file containing the signature.
806                 sf, sfname = tempfile.mkstemp()
807                 with os.fdopen(sf, 'w') as sfd:
808                     sfd.write(detached_sig)
809                 result = gpg.verify_file(io.BytesIO(data), sig_file=sfname)
810                 os.unlink(sfname)
811             gpgpubkey = gpg.list_keys().pop()
812             valid = result.valid
813             rfprint = result.fingerprint
814             kfprint = gpgpubkey['fingerprint']
815             return valid and rfprint == kfprint
816
817     def _get_active_doc_from_address(self, address, private):
818         d = self._soledad.get_from_index(
819             TYPE_ADDRESS_PRIVATE_INDEX,
820             self.ACTIVE_TYPE,
821             address,
822             '1' if private else '0')
823         d.addCallback(self._repair_and_get_doc, self._repair_active_docs)
824         return d
825
826     def _get_key_doc_from_fingerprint(self, fingerprint, private):
827         d = self._soledad.get_from_index(
828             TYPE_FINGERPRINT_PRIVATE_INDEX,
829             self.KEY_TYPE,
830             fingerprint,
831             '1' if private else '0')
832         d.addCallback(self._repair_and_get_doc, self._repair_key_docs)
833         return d
834
835     def _repair_and_get_doc(self, doclist, repair_func):
836         if len(doclist) is 0:
837             return None
838         elif len(doclist) > 1:
839             return repair_func(doclist)
840         return doclist[0]
841
842
843 def process_ascii_key(key_data, gpgbinary, secret=False):
844     with TempGPGWrapper(gpgbinary=gpgbinary) as gpg:
845         try:
846             gpg.import_keys(key_data)
847             info = gpg.list_keys(secret=secret).pop()
848             key = gpg.export_keys(info['fingerprint'], secret=secret)
849         except IndexError:
850             info = {}
851             key = None
852     return info, key
853
854
855 def build_gpg_key(key_info, key_data, gpgbinary=None):
856     expiry_date = None
857     if key_info['expires']:
858         expiry_date = datetime.fromtimestamp(int(key_info['expires']))
859     address = []
860     for uid in key_info['uids']:
861         address.append(_parse_address(uid))
862
863     return OpenPGPKey(
864         address,
865         gpgbinary=gpgbinary,
866         fingerprint=key_info['fingerprint'],
867         key_data=key_data,
868         private=True if key_info['type'] == 'sec' else False,
869         length=int(key_info['length']),
870         expiry_date=expiry_date,
871         refreshed_at=datetime.now())