libervia-backend/sat/memory/crypto.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# SAT: a jabber client
# Copyright (C) 2009-2019 Jérôme Poisson (goffi@goffi.org)
# Copyright (C) 2013-2016 Adrien Cossa (souliane@mailoo.org)

# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.

# This program 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 Affero General Public License for more details.

# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

try:
    from Crypto.Cipher import AES
    from Crypto.Protocol.KDF import PBKDF2
except ImportError:
    raise Exception("PyCrypto is not installed.")

from os import urandom
from base64 import b64encode, b64decode
from twisted.internet.threads import deferToThread
from twisted.internet.defer import succeed


class BlockCipher(object):

    BLOCK_SIZE = AES.block_size  # 16 bits
    MAX_KEY_SIZE = AES.key_size[-1]  # 32 bits = AES-256
    IV_SIZE = BLOCK_SIZE  # initialization vector size, 16 bits

    @classmethod
    def encrypt(cls, key, text, leave_empty=True):
        """Encrypt a message.

        Based on http://stackoverflow.com/a/12525165

        @param key (unicode): the encryption key
        @param text (unicode): the text to encrypt
        @param leave_empty (bool): if True, empty text will be returned "as is"
        @return (D(str)): base-64 encoded encrypted message
        """
        if leave_empty and text == "":
            return succeed(text)
        iv = BlockCipher.getRandomKey()
        key = key.encode("utf-8")
        key = (
            key[: BlockCipher.MAX_KEY_SIZE]
            if len(key) >= BlockCipher.MAX_KEY_SIZE
            else BlockCipher.pad(key)
        )
        cipher = AES.new(key, AES.MODE_CFB, iv)
        d = deferToThread(cipher.encrypt, BlockCipher.pad(text.encode("utf-8")))
        d.addCallback(lambda ciphertext: b64encode(iv + ciphertext))
        d.addCallback(lambda bytes_cypher: bytes_cypher.decode('utf-8'))
        return d

    @classmethod
    def decrypt(cls, key, ciphertext, leave_empty=True):
        """Decrypt a message.

        Based on http://stackoverflow.com/a/12525165

        @param key (unicode): the decryption key
        @param ciphertext (base-64 encoded str): the text to decrypt
        @param leave_empty (bool): if True, empty ciphertext will be returned "as is"
        @return: Deferred: str or None if the password could not be decrypted
        """
        if leave_empty and ciphertext == "":
            return succeed("")
        ciphertext = b64decode(ciphertext)
        iv, ciphertext = (
            ciphertext[: BlockCipher.IV_SIZE],
            ciphertext[BlockCipher.IV_SIZE :],
        )
        key = key.encode("utf-8")
        key = (
            key[: BlockCipher.MAX_KEY_SIZE]
            if len(key) >= BlockCipher.MAX_KEY_SIZE
            else BlockCipher.pad(key)
        )
        cipher = AES.new(key, AES.MODE_CFB, iv)
        d = deferToThread(cipher.decrypt, ciphertext)
        d.addCallback(lambda text: BlockCipher.unpad(text))
        # XXX: cipher.decrypt gives no way to make the distinction between
        # a decrypted empty value and a decryption failure... both return
        # the empty value. Fortunately, we detect empty passwords beforehand
        # thanks to the "leave_empty" parameter which is used by default.
        d.addCallback(lambda text: text if text else None)
        return d

    @classmethod
    def getRandomKey(cls, size=None, base64=False):
        """Return a random key suitable for block cipher encryption.

        Note: a good value for the key length is to make it as long as the block size.

        @param size: key length in bytes, positive or null (default: BlockCipher.IV_SIZE)
        @param base64: if True, encode the result to base-64
        @return: str (eventually base-64 encoded)
        """
        if size is None or size < 0:
            size = BlockCipher.IV_SIZE
        key = urandom(size)
        return b64encode(key) if base64 else key

    @classmethod
    def pad(self, s):
        """Method from http://stackoverflow.com/a/12525165"""
        bs = BlockCipher.BLOCK_SIZE
        return s + (bs - len(s) % bs) * (chr(bs - len(s) % bs)).encode('utf-8')

    @classmethod
    def unpad(self, s):
        """Method from http://stackoverflow.com/a/12525165"""
        s = s.decode('utf-8')
        return s[0 : -ord(s[-1])]


class PasswordHasher(object):

    SALT_LEN = 16  # 128 bits

    @classmethod
    def hash(cls, password, salt=None, leave_empty=True):
        """Hash a password.

        @param password (str): the password to hash
        @param salt (base-64 encoded str): if not None, use the given salt instead of a random value
        @param leave_empty (bool): if True, empty password will be returned "as is"
        @return: Deferred: base-64 encoded str
        """
        if leave_empty and password == "":
            return succeed("")
        salt = (
            b64decode(salt)[: PasswordHasher.SALT_LEN]
            if salt
            else urandom(PasswordHasher.SALT_LEN)
        )
        d = deferToThread(PBKDF2, password, salt)
        d.addCallback(lambda hashed: b64encode(salt + hashed))
        d.addCallback(lambda hashed_bytes: hashed_bytes.decode('utf-8'))
        return d

    @classmethod
    def compare_hash(cls, hashed_attempt, hashed):
        assert isinstance(hashed, str)
        return hashed_attempt == hashed

    @classmethod
    def verify(cls, attempt, hashed):
        """Verify a password attempt.

        @param attempt (str): the attempt to check
        @param hashed (str): the hash of the password
        @return: Deferred: boolean
        """
        assert isinstance(attempt, str)
        assert isinstance(hashed, str)
        leave_empty = hashed == ""
        d = PasswordHasher.hash(attempt, hashed, leave_empty)
        d.addCallback(cls.compare_hash, hashed=hashed)
        return d