Mercurial > libervia-backend
diff sat/memory/crypto.py @ 2562:26edcf3a30eb
core, setup: huge cleaning:
- moved directories from src and frontends/src to sat and sat_frontends, which is the recommanded naming convention
- move twisted directory to root
- removed all hacks from setup.py, and added missing dependencies, it is now clean
- use https URL for website in setup.py
- removed "Environment :: X11 Applications :: GTK", as wix is deprecated and removed
- renamed sat.sh to sat and fixed its installation
- added python_requires to specify Python version needed
- replaced glib2reactor which use deprecated code by gtk3reactor
sat can now be installed directly from virtualenv without using --system-site-packages anymore \o/
author | Goffi <goffi@goffi.org> |
---|---|
date | Mon, 02 Apr 2018 19:44:50 +0200 |
parents | src/memory/crypto.py@0046283a285d |
children | 56f94936df1e |
line wrap: on
line diff
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/sat/memory/crypto.py Mon Apr 02 19:44:50 2018 +0200 @@ -0,0 +1,145 @@ +#!/usr/bin/env python2 +# -*- coding: utf-8 -*- + +# SAT: a jabber client +# Copyright (C) 2009-2018 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: Deferred: base-64 encoded str + """ + 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)) + 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.decode('utf-8') 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) + + @classmethod + def unpad(self, s): + """Method from http://stackoverflow.com/a/12525165""" + 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(password) + 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)) + return d + + @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 + """ + leave_empty = hashed == '' + d = PasswordHasher.hash(attempt, hashed, leave_empty) + d.addCallback(lambda hashed_attempt: hashed_attempt == hashed) + return d