src/leap/keymanager/gpg.py

   1 # -*- coding: utf-8 -*-
   2 # gpgwrapper.py
   3 # Copyright (C) 2013 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
  19 """
  20 A GPG wrapper used to handle OpenPGP keys.
  21
  22 This is a temporary class that will be superseded by the a revised version of
  23 python-gnupg.
  24 """
  25
  26
  27 import os
  28 import gnupg
  29 import re
  30 from gnupg import (
  31     logger,
  32     _is_sequence,
  33     _make_binary_stream,
  34 )
  35
  36
  37 class ListPackets():
  38     """
  39     Handle status messages for --list-packets.
  40     """
  41
  42     def __init__(self, gpg):
  43         """
  44         Initialize the packet listing handling class.
  45
  46         :param gpg: GPG object instance.
  47         :type gpg: gnupg.GPG
  48         """
  49         self.gpg = gpg
  50         self.nodata = None
  51         self.key = None
  52         self.need_passphrase = None
  53         self.need_passphrase_sym = None
  54         self.userid_hint = None
  55
  56     def handle_status(self, key, value):
  57         """
  58         Handle one line of the --list-packets status message.
  59
  60         :param key: The status message key.
  61         :type key: str
  62         :param value: The status message value.
  63         :type value: str
  64         """
  65         # TODO: write tests for handle_status
  66         if key == 'NODATA':
  67             self.nodata = True
  68         if key == 'ENC_TO':
  69             # This will only capture keys in our keyring. In the future we
  70             # may want to include multiple unknown keys in this list.
  71             self.key, _, _ = value.split()
  72         if key == 'NEED_PASSPHRASE':
  73             self.need_passphrase = True
  74         if key == 'NEED_PASSPHRASE_SYM':
  75             self.need_passphrase_sym = True
  76         if key == 'USERID_HINT':
  77             self.userid_hint = value.strip().split()
  78
  79
  80 class GPGWrapper(gnupg.GPG):
  81     """
  82     This is a temporary class for handling GPG requests, and should be
  83     replaced by a more general class used throughout the project.
  84     """
  85
  86     GNUPG_HOME = os.environ['HOME'] + "/.config/leap/gnupg"
  87     GNUPG_BINARY = "/usr/bin/gpg"  # this has to be changed based on OS
  88
  89     def __init__(self, gpgbinary=GNUPG_BINARY, gnupghome=GNUPG_HOME,
  90                  verbose=False, use_agent=False, keyring=None, options=None):
  91         """
  92         Initialize a GnuPG process wrapper.
  93
  94         :param gpgbinary: Name for GnuPG binary executable.
  95         :type gpgbinary: C{str}
  96         :param gpghome: Full pathname to directory containing the public and
  97             private keyrings.
  98         :type gpghome: C{str}
  99         :param keyring: Name of alternative keyring file to use. If specified,
 100             the default keyring is not used.
 101         :param verbose: Should some verbose info be output?
 102         :type verbose: bool
 103         :param use_agent: Should pass `--use-agent` to GPG binary?
 104         :type use_agent: bool
 105         :param keyring: Path for the keyring to use.
 106         :type keyring: str
 107         @options: A list of additional options to pass to the GPG binary.
 108         :type options: list
 109
 110         @raise: RuntimeError with explanation message if there is a problem
 111             invoking gpg.
 112         """
 113         # XXX: options isn't always supported, so removing for the time being
 114         gnupg.GPG.__init__(self, gnupghome=gnupghome, gpgbinary=gpgbinary,
 115                            verbose=verbose, use_agent=use_agent,
 116                            keyring=keyring)#, options=options)
 117         self.result_map['list-packets'] = ListPackets
 118
 119     def find_key_by_email(self, email, secret=False):
 120         """
 121         Find user's key based on their email.
 122
 123         :param email: Email address of key being searched for.
 124         :type email: str
 125         :param secret: Should we search for a secret key?
 126         :type secret: bool
 127
 128         :return: The fingerprint of the found key.
 129         :rtype: str
 130         """
 131         for key in self.list_keys(secret=secret):
 132             for uid in key['uids']:
 133                 if re.search(email, uid):
 134                     return key
 135         raise LookupError("GnuPG public key for email %s not found!" % email)
 136
 137     def find_key_by_subkey(self, subkey, secret=False):
 138         """
 139         Find user's key based on a subkey fingerprint.
 140
 141         :param email: Subkey fingerprint of the key being searched for.
 142         :type email: str
 143         :param secret: Should we search for a secret key?
 144         :type secret: bool
 145
 146         :return: The fingerprint of the found key.
 147         :rtype: str
 148         """
 149         for key in self.list_keys(secret=secret):
 150             for sub in key['subkeys']:
 151                 if sub[0] == subkey:
 152                     return key
 153         raise LookupError(
 154             "GnuPG public key for subkey %s not found!" % subkey)
 155
 156     def find_key_by_keyid(self, keyid, secret=False):
 157         """
 158         Find user's key based on the key ID.
 159
 160         :param email: The key ID of the key being searched for.
 161         :type email: str
 162         :param secret: Should we search for a secret key?
 163         :type secret: bool
 164
 165         :return: The fingerprint of the found key.
 166         :rtype: str
 167         """
 168         for key in self.list_keys(secret=secret):
 169             if keyid == key['keyid']:
 170                 return key
 171         raise LookupError(
 172             "GnuPG public key for keyid %s not found!" % keyid)
 173
 174     def find_key_by_fingerprint(self, fingerprint, secret=False):
 175         """
 176         Find user's key based on the key fingerprint.
 177
 178         :param email: The fingerprint of the key being searched for.
 179         :type email: str
 180         :param secret: Should we search for a secret key?
 181         :type secret: bool
 182
 183         :return: The fingerprint of the found key.
 184         :rtype: str
 185         """
 186         for key in self.list_keys(secret=secret):
 187             if fingerprint == key['fingerprint']:
 188                 return key
 189         raise LookupError(
 190             "GnuPG public key for fingerprint %s not found!" % fingerprint)
 191
 192     def encrypt(self, data, recipient, sign=None, always_trust=True,
 193                 passphrase=None, symmetric=False):
 194         """
 195         Encrypt data using GPG.
 196
 197         :param data: The data to be encrypted.
 198         :type data: str
 199         :param recipient: The address of the public key to be used.
 200         :type recipient: str
 201         :param sign: Should the encrypted content be signed?
 202         :type sign: bool
 203         :param always_trust: Skip key validation and assume that used keys
 204             are always fully trusted?
 205         :type always_trust: bool
 206         :param passphrase: The passphrase to be used if symmetric encryption
 207             is desired.
 208         :type passphrase: str
 209         :param symmetric: Should we encrypt to a password?
 210         :type symmetric: bool
 211
 212         :return: An object with encrypted result in the `data` field.
 213         :rtype: gnupg.Crypt
 214         """
 215         # TODO: devise a way so we don't need to "always trust".
 216         return gnupg.GPG.encrypt(self, data, recipient, sign=sign,
 217                                  always_trust=always_trust,
 218                                  passphrase=passphrase,
 219                                  symmetric=symmetric,
 220                                  cipher_algo='AES256')
 221
 222     def decrypt(self, data, always_trust=True, passphrase=None):
 223         """
 224         Decrypt data using GPG.
 225
 226         :param data: The data to be decrypted.
 227         :type data: str
 228         :param always_trust: Skip key validation and assume that used keys
 229             are always fully trusted?
 230         :type always_trust: bool
 231         :param passphrase: The passphrase to be used if symmetric encryption
 232             is desired.
 233         :type passphrase: str
 234
 235         :return: An object with decrypted result in the `data` field.
 236         :rtype: gnupg.Crypt
 237         """
 238         # TODO: devise a way so we don't need to "always trust".
 239         return gnupg.GPG.decrypt(self, data, always_trust=always_trust,
 240                                  passphrase=passphrase)
 241
 242     def send_keys(self, keyserver, *keyids):
 243         """
 244         Send keys to a keyserver
 245
 246         :param keyserver: The keyserver to send the keys to.
 247         :type keyserver: str
 248         :param keyids: The key ids to send.
 249         :type keyids: list
 250
 251         :return: A list of keys sent to server.
 252         :rtype: gnupg.ListKeys
 253         """
 254         # TODO: write tests for this.
 255         # TODO: write a SendKeys class to handle status for this.
 256         result = self.result_map['list'](self)
 257         gnupg.logger.debug('send_keys: %r', keyids)
 258         data = gnupg._make_binary_stream("", self.encoding)
 259         args = ['--keyserver', keyserver, '--send-keys']
 260         args.extend(keyids)
 261         self._handle_io(args, data, result, binary=True)
 262         gnupg.logger.debug('send_keys result: %r', result.__dict__)
 263         data.close()
 264         return result
 265
 266     def encrypt_file(self, file, recipients, sign=None,
 267                      always_trust=False, passphrase=None,
 268                      armor=True, output=None, symmetric=False,
 269                      cipher_algo=None):
 270         """
 271         Encrypt the message read from the file-like object 'file'.
 272
 273         :param file: The file to be encrypted.
 274         :type data: file
 275         :param recipient: The address of the public key to be used.
 276         :type recipient: str
 277         :param sign: Should the encrypted content be signed?
 278         :type sign: bool
 279         :param always_trust: Skip key validation and assume that used keys
 280             are always fully trusted?
 281         :type always_trust: bool
 282         :param passphrase: The passphrase to be used if symmetric encryption
 283             is desired.
 284         :type passphrase: str
 285         :param armor: Create ASCII armored output?
 286         :type armor: bool
 287         :param output: Path of file to write results in.
 288         :type output: str
 289         :param symmetric: Should we encrypt to a password?
 290         :type symmetric: bool
 291         :param cipher_algo: Algorithm to use.
 292         :type cipher_algo: str
 293
 294         :return: An object with encrypted result in the `data` field.
 295         :rtype: gnupg.Crypt
 296         """
 297         args = ['--encrypt']
 298         if symmetric:
 299             args = ['--symmetric']
 300             if cipher_algo:
 301                 args.append('--cipher-algo %s' % cipher_algo)
 302         else:
 303             args = ['--encrypt']
 304             if not _is_sequence(recipients):
 305                 recipients = (recipients,)
 306             for recipient in recipients:
 307                 args.append('--recipient "%s"' % recipient)
 308         if armor:  # create ascii-armored output - set to False for binary
 309             args.append('--armor')
 310         if output:  # write the output to a file with the specified name
 311             if os.path.exists(output):
 312                 os.remove(output)  # to avoid overwrite confirmation message
 313             args.append('--output "%s"' % output)
 314         if sign:
 315             args.append('--sign --default-key "%s"' % sign)
 316         if always_trust:
 317             args.append("--always-trust")
 318         result = self.result_map['crypt'](self)
 319         self._handle_io(args, file, result, passphrase=passphrase, binary=True)
 320         logger.debug('encrypt result: %r', result.data)
 321         return result
 322
 323     def list_packets(self, data):
 324         """
 325         List the sequence of packets.
 326
 327         :param data: The data to extract packets from.
 328         :type data: str
 329
 330         :return: An object with packet info.
 331         :rtype ListPackets
 332         """
 333         args = ["--list-packets"]
 334         result = self.result_map['list-packets'](self)
 335         self._handle_io(
 336             args,
 337             _make_binary_stream(data, self.encoding),
 338             result,
 339         )
 340         return result
 341
 342     def encrypted_to(self, data):
 343         """
 344         Return the key to which data is encrypted to.
 345
 346         :param data: The data to be examined.
 347         :type data: str
 348
 349         :return: The fingerprint of the key to which data is encrypted to.
 350         :rtype: str
 351         """
 352         # TODO: make this support multiple keys.
 353         result = self.list_packets(data)
 354         if not result.key:
 355             raise LookupError(
 356                 "Content is not encrypted to a GnuPG key!")
 357         try:
 358             return self.find_key_by_keyid(result.key)
 359         except:
 360             return self.find_key_by_subkey(result.key)
 361
 362     def is_encrypted_sym(self, data):
 363         """
 364         Say whether some chunk of data is encrypted to a symmetric key.
 365
 366         :param data: The data to be examined.
 367         :type data: str
 368
 369         :return: Whether data is encrypted to a symmetric key.
 370         :rtype: bool
 371         """
 372         result = self.list_packets(data)
 373         return bool(result.need_passphrase_sym)
 374
 375     def is_encrypted_asym(self, data):
 376         """
 377         Say whether some chunk of data is encrypted to a private key.
 378
 379         :param data: The data to be examined.
 380         :type data: str
 381
 382         :return: Whether data is encrypted to a private key.
 383         :rtype: bool
 384         """
 385         result = self.list_packets(data)
 386         return bool(result.key)
 387
 388     def is_encrypted(self, data):
 389         """
 390         Say whether some chunk of data is encrypted to a key.
 391
 392         :param data: The data to be examined.
 393         :type data: str
 394
 395         :return: Whether data is encrypted to a key.
 396         :rtype: bool
 397         """
 398         return self.is_encrypted_asym(data) or self.is_encrypted_sym(data)