Mercurial > libervia-backend
comparison src/memory/crypto.py @ 1028:127c96020022
memory, test: added module crypto to hash passwords and encrypt/decrypt passwords or blocks
author | souliane <souliane@mailoo.org> |
---|---|
date | Wed, 07 May 2014 15:46:43 +0200 |
parents | |
children | 77cd312d32c4 |
comparison
equal
deleted
inserted
replaced
1027:ee46515a12f2 | 1028:127c96020022 |
---|---|
1 #!/usr/bin/python | |
2 # -*- coding: utf-8 -*- | |
3 | |
4 # SAT: a jabber client | |
5 # Copyright (C) 2009, 2010, 2011, 2012, 2013, 2014 Jérôme Poisson (goffi@goffi.org) | |
6 # Copyright (C) 2013, 2014 Adrien Cossa (souliane@mailoo.org) | |
7 | |
8 # This program is free software: you can redistribute it and/or modify | |
9 # it under the terms of the GNU Affero General Public License as published by | |
10 # the Free Software Foundation, either version 3 of the License, or | |
11 # (at your option) any later version. | |
12 | |
13 # This program is distributed in the hope that it will be useful, | |
14 # but WITHOUT ANY WARRANTY; without even the implied warranty of | |
15 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the | |
16 # GNU Affero General Public License for more details. | |
17 | |
18 # You should have received a copy of the GNU Affero General Public License | |
19 # along with this program. If not, see <http://www.gnu.org/licenses/>. | |
20 | |
21 try: | |
22 from Crypto.Cipher import AES | |
23 from Crypto.Protocol.KDF import PBKDF2 | |
24 except ImportError: | |
25 raise Exception("PyCrypto is not installed.") | |
26 | |
27 from os import urandom | |
28 from base64 import b64encode, b64decode | |
29 from twisted.internet.threads import deferToThread | |
30 from twisted.internet.defer import succeed | |
31 | |
32 | |
33 class BlockCipher(object): | |
34 | |
35 BLOCK_SIZE = AES.block_size # 16 bits | |
36 MAX_KEY_SIZE = AES.key_size[-1] # 32 bits = AES-256 | |
37 IV_SIZE = BLOCK_SIZE # initialization vector size, 16 bits | |
38 | |
39 @classmethod | |
40 def encrypt(cls, key, text, leave_empty=True): | |
41 """Encrypt a message. | |
42 | |
43 Based on http://stackoverflow.com/a/12525165 | |
44 | |
45 @param key (str): the encryption key | |
46 @param text (str): the text to encrypt | |
47 @param leave_empty (bool): if True, empty text will be returned "as is" | |
48 @return: Deferred: base-64 encoded str | |
49 """ | |
50 if leave_empty and text == '': | |
51 return succeed(text) | |
52 iv = BlockCipher.getRandomKey() | |
53 key = key[:BlockCipher.MAX_KEY_SIZE] if len(key) >= BlockCipher.MAX_KEY_SIZE else BlockCipher.pad(key) | |
54 cipher = AES.new(key, AES.MODE_CFB, iv) | |
55 d = deferToThread(cipher.encrypt, BlockCipher.pad(text)) | |
56 d.addCallback(lambda ciphertext: b64encode(iv + ciphertext)) | |
57 return d | |
58 | |
59 @classmethod | |
60 def decrypt(cls, key, ciphertext, leave_empty=True): | |
61 """Decrypt a message. | |
62 | |
63 Based on http://stackoverflow.com/a/12525165 | |
64 | |
65 @param key (str): the decryption key | |
66 @param ciphertext (base-64 encoded str): the text to decrypt | |
67 @param leave_empty (bool): if True, empty ciphertext will be returned "as is" | |
68 @return: Deferred: str or None if the password could not be decrypted | |
69 """ | |
70 if leave_empty and ciphertext == '': | |
71 return succeed('') | |
72 ciphertext = b64decode(ciphertext) | |
73 iv, ciphertext = ciphertext[:BlockCipher.IV_SIZE], ciphertext[BlockCipher.IV_SIZE:] | |
74 key = key[:BlockCipher.MAX_KEY_SIZE] if len(key) >= BlockCipher.MAX_KEY_SIZE else BlockCipher.pad(key) | |
75 cipher = AES.new(key, AES.MODE_CFB, iv) | |
76 d = deferToThread(cipher.decrypt, ciphertext) | |
77 d.addCallback(lambda text: BlockCipher.unpad(text)) | |
78 # XXX: cipher.decrypt gives no way to make the distinction between | |
79 # a decrypted empty value and a decryption failure... both return | |
80 # the empty value. Fortunately, we detect empty passwords beforehand | |
81 # thanks to the "leave_empty" parameter which is used by default. | |
82 d.addCallback(lambda text: text if text else None) | |
83 return d | |
84 | |
85 @classmethod | |
86 def getRandomKey(cls, size=None, base64=False): | |
87 """Return a random key suitable for block cipher encryption. | |
88 | |
89 Note: a good value for the key length is to make it as long as the block size. | |
90 | |
91 @param size: key length in bytes, positive or null (default: BlockCipher.IV_SIZE) | |
92 @param base64: if True, encode the result to base-64 | |
93 @return: str (eventually base-64 encoded) | |
94 """ | |
95 if size is None or size < 0: | |
96 size = BlockCipher.IV_SIZE | |
97 key = urandom(size) | |
98 return b64encode(key) if base64 else key | |
99 | |
100 @classmethod | |
101 def pad(self, s): | |
102 """Method from http://stackoverflow.com/a/12525165""" | |
103 bs = BlockCipher.BLOCK_SIZE | |
104 return s + (bs - len(s) % bs) * chr(bs - len(s) % bs) | |
105 | |
106 @classmethod | |
107 def unpad(self, s): | |
108 """Method from http://stackoverflow.com/a/12525165""" | |
109 return s[0:-ord(s[-1])] | |
110 | |
111 | |
112 class PasswordHasher(object): | |
113 | |
114 SALT_LEN = 16 # 128 bits | |
115 | |
116 @classmethod | |
117 def hash(cls, password, salt=None, leave_empty=True): | |
118 """Hash a password. | |
119 | |
120 @param password (str): the password to hash | |
121 @param salt (base-64 encoded str): if not None, use the given salt instead of a random value | |
122 @param leave_empty (bool): if True, empty password will be returned "as is" | |
123 @return: Deferred: base-64 encoded str | |
124 """ | |
125 if leave_empty and password == '': | |
126 return succeed(password) | |
127 salt = b64decode(salt)[:PasswordHasher.SALT_LEN] if salt else urandom(PasswordHasher.SALT_LEN) | |
128 d = deferToThread(PBKDF2, password, salt) | |
129 d.addCallback(lambda hashed: b64encode(salt + hashed)) | |
130 return d | |
131 | |
132 @classmethod | |
133 def verify(cls, attempt, hashed): | |
134 """Verify a password attempt. | |
135 | |
136 @param attempt (str): the attempt to check | |
137 @param hashed (str): the hash of the password | |
138 @return: Deferred: boolean | |
139 """ | |
140 leave_empty = hashed == '' | |
141 d = PasswordHasher.hash(attempt, hashed, leave_empty) | |
142 d.addCallback(lambda hashed_attempt: hashed_attempt == hashed) | |
143 return d |