diff --git a/README4PACKAGERS b/README4PACKAGERS
index 4285f2da..be55dde6 100644
--- a/README4PACKAGERS
+++ b/README4PACKAGERS
@@ -16,6 +16,7 @@ dbus
lxml
Mutagen
PIL
+PyCrypto
PyFeed
Twisted Core
Twisted Mail
diff --git a/setup.py b/setup.py
index ab6a97dd..b0b2970a 100644
--- a/setup.py
+++ b/setup.py
@@ -178,6 +178,6 @@ setup(name=NAME,
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 --git a/src/memory/crypto.py b/src/memory/crypto.py
new file mode 100644
index 00000000..3ff39631
--- /dev/null
+++ b/src/memory/crypto.py
@@ -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 --git a/src/test/test_memory_crypto.py b/src/test/test_memory_crypto.py
new file mode 100644
index 00000000..b52b4daa
--- /dev/null
+++ b/src/test/test_memory_crypto.py
@@ -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)