Mercurial > libervia-backend

comparison sat/memory/crypto.py @ 3040:fee60f17ebac

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
jp: jp asyncio port: /!\ this commit is huge. Jp is temporarily not working with `dbus` bridge /!\ This patch implements the port of jp to asyncio, so it is now correctly using the bridge asynchronously, and it can be used with bridges like `pb`. This also simplify the code, notably for things which were previously implemented with many callbacks (like pagination with RSM). During the process, some behaviours have been modified/fixed, in jp and backends, check diff for details.
author Goffi <goffi@goffi.org>
date Wed, 25 Sep 2019 08:56:41 +0200
parents ab2696e34d29
children 9d0df638c8b4
comparison
equal deleted inserted replaced
3039:a1bc34f90fa5 3040:fee60f17ebac
43 Based on http://stackoverflow.com/a/12525165 43 Based on http://stackoverflow.com/a/12525165
44 44
45 @param key (unicode): the encryption key 45 @param key (unicode): the encryption key
46 @param text (unicode): the text to encrypt 46 @param text (unicode): the text to encrypt
47 @param leave_empty (bool): if True, empty text will be returned "as is" 47 @param leave_empty (bool): if True, empty text will be returned "as is"
48 @return: Deferred: base-64 encoded str 48 @return (D(str)): base-64 encoded encrypted message
49 """ 49 """
50 if leave_empty and text == "": 50 if leave_empty and text == "":
51 return succeed(text) 51 return succeed(text)
52 iv = BlockCipher.getRandomKey() 52 iv = BlockCipher.getRandomKey()
53 key = key.encode("utf-8") 53 key = key.encode("utf-8")
57 else BlockCipher.pad(key) 57 else BlockCipher.pad(key)
58 ) 58 )
59 cipher = AES.new(key, AES.MODE_CFB, iv) 59 cipher = AES.new(key, AES.MODE_CFB, iv)
60 d = deferToThread(cipher.encrypt, BlockCipher.pad(text.encode("utf-8"))) 60 d = deferToThread(cipher.encrypt, BlockCipher.pad(text.encode("utf-8")))
61 d.addCallback(lambda ciphertext: b64encode(iv + ciphertext)) 61 d.addCallback(lambda ciphertext: b64encode(iv + ciphertext))
62 d.addCallback(lambda bytes_cypher: bytes_cypher.decode('utf-8'))
62 return d 63 return d
63 64
64 @classmethod 65 @classmethod
65 def decrypt(cls, key, ciphertext, leave_empty=True): 66 def decrypt(cls, key, ciphertext, leave_empty=True):
66 """Decrypt a message. 67 """Decrypt a message.
135 @param salt (base-64 encoded str): if not None, use the given salt instead of a random value 136 @param salt (base-64 encoded str): if not None, use the given salt instead of a random value
136 @param leave_empty (bool): if True, empty password will be returned "as is" 137 @param leave_empty (bool): if True, empty password will be returned "as is"
137 @return: Deferred: base-64 encoded str 138 @return: Deferred: base-64 encoded str
138 """ 139 """
139 if leave_empty and password == "": 140 if leave_empty and password == "":
140 return succeed(b"") 141 return succeed("")
141 salt = ( 142 salt = (
142 b64decode(salt)[: PasswordHasher.SALT_LEN] 143 b64decode(salt)[: PasswordHasher.SALT_LEN]
143 if salt 144 if salt
144 else urandom(PasswordHasher.SALT_LEN) 145 else urandom(PasswordHasher.SALT_LEN)
145 ) 146 )
146 d = deferToThread(PBKDF2, password, salt) 147 d = deferToThread(PBKDF2, password, salt)
147 d.addCallback(lambda hashed: b64encode(salt + hashed)) 148 d.addCallback(lambda hashed: b64encode(salt + hashed))
149 d.addCallback(lambda hashed_bytes: hashed_bytes.decode('utf-8'))
148 return d 150 return d
149 151
150 @classmethod 152 @classmethod
151 def compare_hash(cls, hashed_attempt, hashed): 153 def compare_hash(cls, hashed_attempt, hashed):
152 assert isinstance(hashed, bytes) 154 assert isinstance(hashed, str)
153 return hashed_attempt == hashed 155 return hashed_attempt == hashed
154 156
155 @classmethod 157 @classmethod
156 def verify(cls, attempt, hashed): 158 def verify(cls, attempt, hashed):
157 """Verify a password attempt. 159 """Verify a password attempt.
158 160
159 @param attempt (str): the attempt to check 161 @param attempt (str): the attempt to check
160 @param hashed (str): the hash of the password 162 @param hashed (str): the hash of the password
161 @return: Deferred: boolean 163 @return: Deferred: boolean
162 """ 164 """
165 assert isinstance(attempt, str)
166 assert isinstance(hashed, str)
163 leave_empty = hashed == "" 167 leave_empty = hashed == ""
164 d = PasswordHasher.hash(attempt, hashed, leave_empty) 168 d = PasswordHasher.hash(attempt, hashed, leave_empty)
165 d.addCallback(cls.compare_hash, hashed=hashed.encode('utf-8')) 169 d.addCallback(cls.compare_hash, hashed=hashed)
166 return d 170 return d