python_gnupg.py 5.03 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107
# Copyright (C) 2019 Philipp Hörist <philipp AT hoerist.com>
# Copyright (C) 2003-2014 Yann Leboulanger <asterix AT lagaule.org>
# Copyright (C) 2005 Alex Mauer <hawke AT hawkesnest.net>
# Copyright (C) 2005-2006 Nikos Kouremenos <kourem AT gmail.com>
# Copyright (C) 2007 Stephan Erb <steve-e AT h3c.de>
# Copyright (C) 2008 Jean-Marie Traissard <jim AT lapin.org>
#                    Jonathan Schleifer <js-gajim AT webkeks.org>
#
# This file is part of PGP Gajim Plugin.
#
# PGP Gajim Plugin is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published
# by the Free Software Foundation; version 3 only.
#
# PGP Gajim Plugin is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with PGP Gajim Plugin. If not, see <http://www.gnu.org/licenses/>.

import os
import logging
from functools import lru_cache

import gnupg

from gajim.common.helpers import Singleton

from pgp.exceptions import SignError


logger = logging.getLogger('gajim.p.pgplegacy')
if logger.getEffectiveLevel() == logging.DEBUG:
    logger = logging.getLogger('gnupg')
    logger.addHandler(logging.StreamHandler())
    logger.setLevel(logging.DEBUG)


class PGP(gnupg.GPG, metaclass=Singleton):
    def __init__(self, binary, encoding=None):
        super().__init__(gpgbinary=binary,
                         use_agent=True)

        if encoding is not None:
            self.encoding = encoding
        self.decode_errors = 'replace'

    def encrypt(self, payload, recipients, always_trust=False):
        if not always_trust:
            # check that we'll be able to encrypt
            result = self.get_key(recipients[0])
            for key in result:
                if key['trust'] not in ('f', 'u'):
                    return '', 'NOT_TRUSTED ' + key['keyid'][-8:]

        result = super().encrypt(
            payload.encode('utf8'),
            recipients,
            always_trust=always_trust)

        if result.ok:
            error = ''
        else:
            error = result.status

        return self._strip_header_footer(str(result)), error

    def decrypt(self, payload):
        data = self._add_header_footer(payload, 'MESSAGE')
        result = super().decrypt(data.encode('utf8'))

        return result.data.decode('utf8')

    @lru_cache(maxsize=8)
    def sign(self, payload, key_id):
        if payload is None:
            payload = ''
        result = super().sign(payload.encode('utf8'),
                              keyid=key_id,
                              detach=True)

        if result.fingerprint:
            return self._strip_header_footer(str(result))
        raise SignError(result.status)

    def verify(self, payload, signed):
        # Hash algorithm is not transfered in the signed
        # presence stanza so try all algorithms.
        # Text name for hash algorithms from RFC 4880 - section 9.4

        if payload is None:
            payload = ''

        hash_algorithms = ['SHA512', 'SHA384', 'SHA256',
                           'SHA224', 'SHA1', 'RIPEMD160']
        for algo in hash_algorithms:
            data = os.linesep.join(
                ['-----BEGIN PGP SIGNED MESSAGE-----',
                 'Hash: ' + algo,
                 '',
                 payload,
                 self._add_header_footer(signed, 'SIGNATURE')]
                )
            result = super().verify(data.encode('utf8'))
            if result.valid:
108
                return result.fingerprint
109 110 111 112 113 114 115 116 117 118

    def get_key(self, key_id):
        return super().list_keys(keys=[key_id])

    def get_keys(self, secret=False):
        keys = {}
        result = super().list_keys(secret=secret)

        for key in result:
            # Take first not empty uid
119 120 121 122 123
            try:
                keys[key['fingerprint']] = next(uid for uid in key['uids'] if uid)
            except StopIteration:
                # No key non-empty uids present
                keys[key['fingerprint']] = ''
124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157
        return keys

    @staticmethod
    def _strip_header_footer(data):
        """
        Remove header and footer from data
        """
        if not data:
            return ''
        lines = data.splitlines()
        while lines[0] != '':
            lines.remove(lines[0])
        while lines[0] == '':
            lines.remove(lines[0])
        i = 0
        for line in lines:
            if line:
                if line[0] == '-':
                    break
            i = i+1
        line = '\n'.join(lines[0:i])
        return line

    @staticmethod
    def _add_header_footer(data, type_):
        """
        Add header and footer from data
        """
        out = "-----BEGIN PGP %s-----" % type_ + os.linesep
        out = out + "Version: PGP" + os.linesep
        out = out + os.linesep
        out = out + data + os.linesep
        out = out + "-----END PGP %s-----" % type_ + os.linesep
        return out