Merge pull request #4838 from SomberNight/keystore_pw_hash2b

keystore: stronger pbkdf for encryption

electrum/crypto.py

@@ -32,6 +32,7 @@ from typing import Union
 import pyaes
 
 from .util import assert_bytes, InvalidPassword, to_bytes, to_string
+from .i18n import _
 
 
 try:
@@ -90,37 +91,103 @@ def aes_decrypt_with_iv(key: bytes, iv: bytes, data: bytes) -> bytes:
         raise InvalidPassword()
 
 
-def EncodeAES(secret: bytes, msg: bytes) -> bytes:
+def EncodeAES_base64(secret: bytes, msg: bytes) -> bytes:
     """Returns base64 encoded ciphertext."""
-    assert_bytes(msg)
-    iv = bytes(os.urandom(16))
-    ct = aes_encrypt_with_iv(secret, iv, msg)
-    e = iv + ct
+    e = EncodeAES_bytes(secret, msg)
     return base64.b64encode(e)
 
 
-def DecodeAES(secret: bytes, ciphertext_b64: Union[bytes, str]) -> bytes:
-    e = bytes(base64.b64decode(ciphertext_b64))
-    iv, e = e[:16], e[16:]
+def EncodeAES_bytes(secret: bytes, msg: bytes) -> bytes:
+    assert_bytes(msg)
+    iv = bytes(os.urandom(16))
+    ct = aes_encrypt_with_iv(secret, iv, msg)
+    return iv + ct
+
+
+def DecodeAES_base64(secret: bytes, ciphertext_b64: Union[bytes, str]) -> bytes:
+    ciphertext = bytes(base64.b64decode(ciphertext_b64))
+    return DecodeAES_bytes(secret, ciphertext)
+
+
+def DecodeAES_bytes(secret: bytes, ciphertext: bytes) -> bytes:
+    assert_bytes(ciphertext)
+    iv, e = ciphertext[:16], ciphertext[16:]
     s = aes_decrypt_with_iv(secret, iv, e)
     return s
 
 
-def pw_encode(data: str, password: Union[bytes, str]) -> str:
+PW_HASH_VERSION_LATEST = 2
+KNOWN_PW_HASH_VERSIONS = (1, 2)
+assert PW_HASH_VERSION_LATEST in KNOWN_PW_HASH_VERSIONS
+
+
+class UnexpectedPasswordHashVersion(InvalidPassword):
+    def __init__(self, version):
+        self.version = version
+
+    def __str__(self):
+        return "{unexpected}: {version}\n{please_update}".format(
+            unexpected=_("Unexpected password hash version"),
+            version=self.version,
+            please_update=_('You are most likely using an outdated version of Electrum. Please update.'))
+
+
+def _hash_password(password: Union[bytes, str], *, version: int, salt: bytes) -> bytes:
+    pw = to_bytes(password, 'utf8')
+    if version == 1:
+        return sha256d(pw)
+    elif version == 2:
+        if not isinstance(salt, bytes) or len(salt) < 16:
+            raise Exception('too weak salt', salt)
+        return hashlib.pbkdf2_hmac(hash_name='sha256',
+                                   password=pw,
+                                   salt=b'ELECTRUM_PW_HASH_V2'+salt,
+                                   iterations=50_000)
+    else:
+        assert version not in KNOWN_PW_HASH_VERSIONS
+        raise UnexpectedPasswordHashVersion(version)
+
+
+def pw_encode(data: str, password: Union[bytes, str, None], *, version: int) -> str:
     if not password:
         return data
-    secret = sha256d(password)
-    return EncodeAES(secret, to_bytes(data, "utf8")).decode('utf8')
+    if version not in KNOWN_PW_HASH_VERSIONS:
+        raise UnexpectedPasswordHashVersion(version)
+    # derive key from password
+    if version == 1:
+        salt = b''
+    elif version == 2:
+        salt = bytes(os.urandom(16))
+    else:
+        assert False, version
+    secret = _hash_password(password, version=version, salt=salt)
+    # encrypt given data
+    e = EncodeAES_bytes(secret, to_bytes(data, "utf8"))
+    # return base64(salt + encrypted data)
+    ciphertext = salt + e
+    ciphertext_b64 = base64.b64encode(ciphertext)
+    return ciphertext_b64.decode('utf8')
 
 
-def pw_decode(data: str, password: Union[bytes, str]) -> str:
+def pw_decode(data: str, password: Union[bytes, str, None], *, version: int) -> str:
     if password is None:
         return data
-    secret = sha256d(password)
+    if version not in KNOWN_PW_HASH_VERSIONS:
+        raise UnexpectedPasswordHashVersion(version)
+    data_bytes = bytes(base64.b64decode(data))
+    # derive key from password
+    if version == 1:
+        salt = b''
+    elif version == 2:
+        salt, data_bytes = data_bytes[:16], data_bytes[16:]
+    else:
+        assert False, version
+    secret = _hash_password(password, version=version, salt=salt)
+    # decrypt given data
     try:
-        d = to_string(DecodeAES(secret, data), "utf8")
-    except Exception:
-        raise InvalidPassword()
+        d = to_string(DecodeAES_bytes(secret, data_bytes), "utf8")
+    except Exception as e:
+        raise InvalidPassword() from e
     return d