# HG changeset patch # User souliane # Date 1399470403 -7200 # Node ID 127c96020022588eb9eca780975ef0369a691616 # Parent ee46515a12f258fa3f8ca2c3b4862d9308bc8d17 memory, test: added module crypto to hash passwords and encrypt/decrypt passwords or blocks diff -r ee46515a12f2 -r 127c96020022 README4PACKAGERS --- a/README4PACKAGERS Thu May 15 20:25:52 2014 +0200 +++ b/README4PACKAGERS Wed May 07 15:46:43 2014 +0200 @@ -16,6 +16,7 @@ lxml Mutagen PIL +PyCrypto PyFeed Twisted Core Twisted Mail diff -r ee46515a12f2 -r 127c96020022 setup.py --- a/setup.py Thu May 15 20:25:52 2014 +0200 +++ b/setup.py Wed May 07 15:46:43 2014 +0200 @@ -178,6 +178,6 @@ scripts=['frontends/src/jp/jp', 'frontends/src/primitivus/primitivus', 'frontends/src/wix/wix'], zip_safe=False, dependency_links=['http://www.blarg.net/%7Esteveha/pyfeed-0.7.4.tar.gz', 'http://www.blarg.net/%7Esteveha/xe-0.7.4.tar.gz'], - install_requires=['twisted', 'wokkel >= 0.7.1', 'progressbar', 'urwid >= 1.1.0', 'urwid-satext >= 0.3.0', 'pyfeed', 'xe', 'mutagen', 'PIL', 'lxml', 'pyxdg', 'markdown', 'html2text'], + install_requires=['twisted', 'wokkel >= 0.7.1', 'progressbar', 'urwid >= 1.1.0', 'urwid-satext >= 0.3.0', 'pyfeed', 'xe', 'mutagen', 'PIL', 'lxml', 'pyxdg', 'markdown', 'html2text', 'pycrypto'], cmdclass={'install': CustomInstall}, ) # XXX: wxpython doesn't work, it's managed with preinstall_check diff -r ee46515a12f2 -r 127c96020022 src/memory/crypto.py --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/src/memory/crypto.py Wed May 07 15:46:43 2014 +0200 @@ -0,0 +1,143 @@ +#!/usr/bin/python +# -*- coding: utf-8 -*- + +# SAT: a jabber client +# Copyright (C) 2009, 2010, 2011, 2012, 2013, 2014 Jérôme Poisson (goffi@goffi.org) +# Copyright (C) 2013, 2014 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 . + +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 (str): the encryption key + @param text (str): 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[: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)) + 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 (str): 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[: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) + + @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 diff -r ee46515a12f2 -r 127c96020022 src/test/test_memory_crypto.py --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/src/test/test_memory_crypto.py Wed May 07 15:46:43 2014 +0200 @@ -0,0 +1,62 @@ +#!/usr/bin/python +# -*- coding: utf-8 -*- + +# SAT: a jabber client +# Copyright (C) 2009, 2010, 2011, 2012, 2013, 2014 Jérôme Poisson (goffi@goffi.org) +# Copyright (C) 2013, 2014 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 . + + +""" Tests for the plugin radiocol """ + +from sat.test import helpers +from sat.memory.crypto import BlockCipher, PasswordHasher +from os import urandom +from twisted.internet import defer + + +class CryptoTest(helpers.SatTestCase): + + def setUp(self): + self.host = helpers.FakeSAT() + + def test_encrypt_decrypt(self): + d_list = [] + for key_len in (0, 2, 8, 10, 16, 24, 30, 32, 40): + key = urandom(key_len) + for message_len in (0, 2, 16, 24, 32, 100): + message = urandom(message_len) + d = BlockCipher.encrypt(key, message) + d.addCallback(lambda ciphertext: lambda key, cipher: BlockCipher.decrypt(key, ciphertext)) + d.addCallback(lambda decrypted: lambda message, decrypted: self.assertEqual(message, decrypted)) + d_list.append(d) + return defer.DeferredList(d_list) + + def test_hash_verify(self): + d_list = [] + for password in (0, 2, 8, 10, 16, 24, 30, 32, 40): + d = PasswordHasher.hash(password) + + def cb(hashed): + d1 = PasswordHasher.verify(password, hashed) + d1.addCallback(lambda result: self.assertTrue(result)) + d_list.append(d1) + attempt = urandom(10) + d2 = PasswordHasher.verify(attempt, hashed) + d2.addCallback(lambda result: self.assertFalse(result)) + d_list.append(d2) + + d.addCallback(cb) + return defer.DeferredList(d_list)