Mercurial > libervia-backend

comparison sat/memory/crypto.py @ 2562:26edcf3a30eb

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
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
comparison
equal deleted inserted replaced
2561:bd30dc3ffe5a 2562:26edcf3a30eb
1 #!/usr/bin/env python2
2 # -*- coding: utf-8 -*-
3
4 # SAT: a jabber client
5 # Copyright (C) 2009-2018 Jérôme Poisson (goffi@goffi.org)
6 # Copyright (C) 2013-2016 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 (unicode): the encryption key
46 @param text (unicode): 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.encode('utf-8')
54 key = key[:BlockCipher.MAX_KEY_SIZE] if len(key) >= BlockCipher.MAX_KEY_SIZE else BlockCipher.pad(key)
55 cipher = AES.new(key, AES.MODE_CFB, iv)
56 d = deferToThread(cipher.encrypt, BlockCipher.pad(text.encode('utf-8')))
57 d.addCallback(lambda ciphertext: b64encode(iv + ciphertext))
58 return d
59
60 @classmethod
61 def decrypt(cls, key, ciphertext, leave_empty=True):
62 """Decrypt a message.
63
64 Based on http://stackoverflow.com/a/12525165
65
66 @param key (unicode): the decryption key
67 @param ciphertext (base-64 encoded str): the text to decrypt
68 @param leave_empty (bool): if True, empty ciphertext will be returned "as is"
69 @return: Deferred: str or None if the password could not be decrypted
70 """
71 if leave_empty and ciphertext == '':
72 return succeed('')
73 ciphertext = b64decode(ciphertext)
74 iv, ciphertext = ciphertext[:BlockCipher.IV_SIZE], ciphertext[BlockCipher.IV_SIZE:]
75 key = key.encode('utf-8')
76 key = key[:BlockCipher.MAX_KEY_SIZE] if len(key) >= BlockCipher.MAX_KEY_SIZE else BlockCipher.pad(key)
77 cipher = AES.new(key, AES.MODE_CFB, iv)
78 d = deferToThread(cipher.decrypt, ciphertext)
79 d.addCallback(lambda text: BlockCipher.unpad(text))
80 # XXX: cipher.decrypt gives no way to make the distinction between
81 # a decrypted empty value and a decryption failure... both return
82 # the empty value. Fortunately, we detect empty passwords beforehand
83 # thanks to the "leave_empty" parameter which is used by default.
84 d.addCallback(lambda text: text.decode('utf-8') if text else None)
85 return d
86
87 @classmethod
88 def getRandomKey(cls, size=None, base64=False):
89 """Return a random key suitable for block cipher encryption.
90
91 Note: a good value for the key length is to make it as long as the block size.
92
93 @param size: key length in bytes, positive or null (default: BlockCipher.IV_SIZE)
94 @param base64: if True, encode the result to base-64
95 @return: str (eventually base-64 encoded)
96 """
97 if size is None or size < 0:
98 size = BlockCipher.IV_SIZE
99 key = urandom(size)
100 return b64encode(key) if base64 else key
101
102 @classmethod
103 def pad(self, s):
104 """Method from http://stackoverflow.com/a/12525165"""
105 bs = BlockCipher.BLOCK_SIZE
106 return s + (bs - len(s) % bs) * chr(bs - len(s) % bs)
107
108 @classmethod
109 def unpad(self, s):
110 """Method from http://stackoverflow.com/a/12525165"""
111 return s[0:-ord(s[-1])]
112
113
114 class PasswordHasher(object):
115
116 SALT_LEN = 16 # 128 bits
117
118 @classmethod
119 def hash(cls, password, salt=None, leave_empty=True):
120 """Hash a password.
121
122 @param password (str): the password to hash
123 @param salt (base-64 encoded str): if not None, use the given salt instead of a random value
124 @param leave_empty (bool): if True, empty password will be returned "as is"
125 @return: Deferred: base-64 encoded str
126 """
127 if leave_empty and password == '':
128 return succeed(password)
129 salt = b64decode(salt)[:PasswordHasher.SALT_LEN] if salt else urandom(PasswordHasher.SALT_LEN)
130 d = deferToThread(PBKDF2, password, salt)
131 d.addCallback(lambda hashed: b64encode(salt + hashed))
132 return d
133
134 @classmethod
135 def verify(cls, attempt, hashed):
136 """Verify a password attempt.
137
138 @param attempt (str): the attempt to check
139 @param hashed (str): the hash of the password
140 @return: Deferred: boolean
141 """
142 leave_empty = hashed == ''
143 d = PasswordHasher.hash(attempt, hashed, leave_empty)
144 d.addCallback(lambda hashed_attempt: hashed_attempt == hashed)
145 return d