fix(fhempy): replace cryptography with pycryptodome

FHEM/bindings/python/fhempy/lib/fusionsolar/fusionsolar_api.py

@@ -7,9 +7,9 @@
 from datetime import datetime
 
 import aiohttp
-from Crypto.Cipher import PKCS1_OAEP
-from Crypto.Hash import SHA384
-from Crypto.PublicKey import RSA
+from Cryptodome.Cipher import PKCS1_OAEP
+from Cryptodome.Hash import SHA384
+from Cryptodome.PublicKey import RSA
 
 
 @dataclass

FHEM/bindings/python/fhempy/lib/fusionsolar/manifest.json

@@ -1,3 +1,5 @@
 {
-  "requirements": []
+  "requirements": [
+    "pycryptodome==3.15.0"
+  ]
 }

FHEM/bindings/python/fhempy/lib/utils.py

@@ -1,30 +1,57 @@
 import asyncio
-import base64
+import binascii
 import concurrent.futures
 import json
 import socket
-from codecs import decode, encode
+from base64 import urlsafe_b64decode, urlsafe_b64encode
+from codecs import decode
 from datetime import datetime
 from functools import partial, reduce
 
-from cryptography.fernet import Fernet
+from Cryptodome.Cipher import AES
+from Cryptodome.Hash import HMAC, SHA256
+from Cryptodome.Util.Padding import unpad
 
 from . import fhem
 
 
 def encrypt_string(plain_text, fhem_unique_id):
-    key = base64.b64encode(fhem_unique_id.encode("utf-8"))
-    cipher_suite = Fernet(key)
-    encrypted_text = cipher_suite.encrypt(plain_text.encode("utf-8"))
-    return reduce(encode, ("zlib", "base64"), encrypted_text).decode("utf-8")
+    key = fhem_unique_id.encode("utf-8")
+    b_text = plain_text.encode("utf-8")
+    e_cipher = AES.new(key, AES.MODE_EAX, nonce=key[0:16])
+    e_data = e_cipher.encrypt(b_text)
+    return "crypt-aes:" + urlsafe_b64encode(e_data).decode()
 
 
 def decrypt_string(encrypted_text, fhem_unique_id):
-    key = base64.b64encode(fhem_unique_id.encode("utf-8"))
-    encrypted_text = encrypted_text.encode("utf-8")
-    uncompressed_text = reduce(decode, ("base64", "zlib"), encrypted_text)
-    cipher_suite = Fernet(key)
-    return cipher_suite.decrypt(uncompressed_text).decode("utf-8")
+    if encrypted_text[0:10] != "crypt-aes:":
+        return decrypt_fernet(encrypted_text, fhem_unique_id)
+    key = fhem_unique_id.encode("utf-8")
+    encrypted_data = urlsafe_b64decode(encrypted_text[10:])
+    d_cipher = AES.new(key, AES.MODE_EAX, nonce=key[0:16])
+    return d_cipher.decrypt(encrypted_data).decode()
+
+
+def decrypt_fernet(token_b64, key_str):
+    try:
+        keys = key_str.encode("utf-8")
+        token_z = token_b64.encode("utf-8")
+        token_b64 = reduce(decode, ("base64", "zlib"), token_z).decode()
+        token = urlsafe_b64decode(token_b64)
+
+        if len(keys) != 32 or len(token) < 73 or token[0] != 0x80:
+            raise ValueError("Incorrect format")
+
+        hmac_key, dec_key = keys[:16], keys[16:]
+        HMAC.new(hmac_key, token[:-32], digestmod=SHA256).verify(token[-32:])
+        iv, ct = token[9:25], token[25:-32]
+        # Possibly check time-to-live here
+        message = unpad(AES.new(dec_key, AES.MODE_CBC, iv=iv).decrypt(ct), 16).decode()
+
+    except (ValueError, binascii.Error):
+        message = None
+
+    return message
 
 
 async def run_blocking(function):

FHEM/bindings/python/fhempy/lib/xiaomi_gateway3/manifest.json

@@ -2,6 +2,7 @@
   "requirements": [
     "python-miio==0.5.8",
     "paho-mqtt==1.5.1",
-    "xmodem==0.4.6"
+    "xmodem==0.4.6",
+    "cryptography"
   ]
 }

requirements.txt

@@ -1,6 +1,6 @@
 websockets==10.3
 importlib_metadata==4.8.1
-cryptography==37.0.4
+pycryptodomex==3.16.0
 zeroconf==0.36.12
 aiohttp[speedups]==3.8.1
 markdown2==2.4.2

requirements_dev.txt

@@ -8,7 +8,7 @@ async-upnp-client
 python_semantic_release
 websockets==10.3
 importlib_metadata==4.8.1
-cryptography==37.0.4
+pycryptodomex==3.16.0
 zeroconf==0.36.12
 aiohttp[speedups]==3.8.1
 markdown2==2.4.2

requirements_tests.txt

@@ -7,7 +7,7 @@ requests-mock
 async-upnp-client
 websockets==10.3
 importlib_metadata==4.8.1
-cryptography==37.0.4
+pycryptodomex==3.16.0
 zeroconf==0.36.12
 aiohttp[speedups]==3.8.1
 markdown2==2.4.2