Merge pull request #16 from tendai-zw/master

Replace MaltegoOAuth's pycrptodome dependency with cryptography lib

maltego_trx/oauth.py

@@ -1,9 +1,10 @@
 """Maltego OAuth Crypto Helper"""
 import base64
-from Crypto import Random
-from Crypto.Cipher import PKCS1_v1_5, AES
-from Crypto.Hash import SHA
-from Crypto.PublicKey import RSA
+
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives import serialization, padding as primitives_padding
+from cryptography.hazmat.primitives.asymmetric import padding as asymmetric_padding
+from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
 from requests.auth import AuthBase
 
 
@@ -16,33 +17,37 @@ class MaltegoOauth:
     3. Refresh Token
     4. Expires In
 
-    Contains 1 Methods:
+    Contains Methods:
         1. decrypt_secrets(private_key_path="pem file", ciphertext="request.getTransformSetting('name from TDS')")
     """
 
     @staticmethod
-    def _rsa_decrypt(private_key_path=None, ciphertext=None):
+    def _rsa_decrypt(private_key_path=None, ciphertext=None, password=None):
         """
         RSA Decryption function, returns decrypted plaintext in b64 encoding
         """
-        dsize = SHA.digest_size
-        sentinel = Random.new().read(20 + dsize)
         ciphertext = base64.b64decode(ciphertext)
-        private_key = RSA.import_key(open(private_key_path).read())
-        cipher = PKCS1_v1_5.new(private_key)
-        plaintext = cipher.decrypt(ciphertext, sentinel).decode('utf8')
+
+        with open(private_key_path, "rb") as key_file:
+            private_key = serialization.load_pem_private_key(key_file.read(),
+                                                             password,
+                                                             backend=None)
+            plaintext = private_key.decrypt(ciphertext, asymmetric_padding.PKCS1v15())
+
         return plaintext
 
     @staticmethod
     def _aes_decrypt(key=None, ciphertext=None):
         """
         AES Decryption function, returns decrypted plaintext value
         """
-        unpad = lambda s: s[:-ord(s[len(s) - 1:])]
         key = base64.b64decode(key)
+        cipher = Cipher(algorithms.AES(key), modes.ECB(), backend=default_backend())
+        decryptor = cipher.decryptor();
         ciphertext = base64.b64decode(ciphertext)
-        cipher = AES.new(key, AES.MODE_ECB)
-        plaintext = unpad(cipher.decrypt(ciphertext)).decode('utf8')
+        padded_b64_plaintext = decryptor.update(ciphertext) + decryptor.finalize()
+        unpadder = primitives_padding.PKCS7(128).unpadder()
+        plaintext = (unpadder.update(padded_b64_plaintext) + unpadder.finalize()).decode('utf8')
         return plaintext
 
     @classmethod

setup.py

@@ -11,7 +11,7 @@
       install_requires=[
        'flask>=1',
        'six>=1',
-       'pycryptodome>=3.9.7'
+       'cryptography>=3.3.1'
       ],
       packages=[
           'maltego_trx',

tests/test.py

@@ -1,4 +1,5 @@
 from maltego_trx.oauth import MaltegoOauth
+import os
 
 okta_ciphertoken = "pZ598ZEZ7EwpBQSOJSvCZJKkcWhtbX95K7Q0f0hwbk93O+xaUB4/NegK3r54PH1NReis/Jgt4UbGc5oCuU+R7UDM1icoDUmyCmV1U78iqjHElPdDBHlxPrl2zoXBwcU1iFbukDNy49Xghy3cwwqhmEXg/FKnYUlBQl6jdf06kE1pwfHiRgF5NrJISAD7eCmUybqiAVRDW4lbLeMGk/uSrrCpDQBxd2Em/sYKBzqO87pQRPLll23G9KNxyVinHQksGgc+dI0OPYv9EQcUs7g1JFraUKUeFjHAf/OyGGrp6WH84nGqG35afkH9xsDmySBb5DZWRjef8DzSd7oRagoGq3wKBHDh700mMKf/YkR8L9cK4l1w4yNh7EfIVlCWD5yWJUx4a1lN9CvJmFs7N9A903spVVGeP8avz50raWFb5g/4XhtIpei4ylVFi49dDeVjgb0BR0I4vuU7VDmcwFweAtSfclPb5Xfp98zI9yXnr8DB31gmzf3DIkVFIIFLxZe0gG+BSIEmC7/z0L7J+YvIqoGDq8jC0Ehe9IVXYfK1yluWAsbwj62rNAzkPVSe5l+FMQFrQRXpxZFBMXzphqRAhyYKfGNX/tocoMJoLFp8O3PBRC0kIHjTNaSX8KpftNYWreEdZx6wjk2n2eeOXM5nVeSOx4uoIifRe4NUV7VYzbGm3FSRc0k1HZIGHwn1WTvn95U3crvwFN86N3k41JU+0kzSpsLxy+Y63VfPxYTbRUET/wIOf0NTfG96QCPtVKkdVh9aTqnhaN0TpGoaZQEjUqpURxbj+8mT3g6MI4vOC9Qr6vw2wIkvzehr/yKxh6vsnfUUw8UJ1YtlZI7exPsHyFLUr2J3VvpDsaemRcn1KY/uHJZ3MMHK9GNciLA6Skw8SL38mw1tJJZ1h1bMMk1IYgBZLe2ZSXXyv7JaDPq/WJmHWDaIalmXE1cKMA56qMgYGdHxs3Pai5nwEtzIGVS1OcuMnBF6sOMMBTKSFAJmlcYo8NGwjKqe/yy1cVJZe7Ucv/jy/IS3AdE1my8cDya92jDsxifsaONQMm6YIs54SIA=$f7TU33wTTxYFtbVnwJeGGw==$c64l75qukq+980EkB1tQZKEicryt6HBB0IOjeMSRQmLOYCQpTcaXlPRJ3QdS+DfRocor7OL36wLAooRqz1IaN4QQ5+6Wx5reEXFVgKtaDX777I046DM5QU3Jf+iibSbm3mXYMw0z+OknbXyGLnc7ceSaPTJdP1LkCGQ75fLMXHCLWSpqwOKHOBhqyQGwrYlj2WxPzmOfAMaIkjKZIQWzoGjrYRzwkNCQbxykJcwD5TVuVDwAHyp84zcvW0WWoUZ+rrrooJwuJJQEdiTwLZsseqklXRNso4e5eFQwH49T9IDPHkKfVusu6rLiMNgFyc18rFR1d/BYLXBu7uzMAuvQ8wZdcOtJYx+JLJmOaPI65ymGNFTpwTHYnDBTXmpW0qX2dtEglAERw1nrdl94fwsa+I/iw3H7VIivqRF3pAchfvdNF75MIEm6XH2UXY/sZD7zjMdxxwKQiaEg2bpLd2mEtqsLc0mTq/zd03ZLEUnzXsXlp8brCfDgjoVsZAJH+ElQ6wr3dxzRMlDIxWWBYFcv8LnGVQk/Vciqtee780Yh/lgttv06kyXz81dIWjumz8TqV2eV9ZpP/YxTnNzAa91fleGNah/IKhMOV8PsGy2uOnHRiUL218p1T6+Cm9B2kmL1C/2MM3080NjJVWIfqYsyTGPbR+hURK0RB/oteG5QWPs="
 private_key_path = "test_private_key.pem"
@@ -11,9 +12,7 @@ def testOktaDecrypt():
 
 
 def testReadKey():
-    private_key = open(private_key_path).read()
-    assert private_key.__sizeof__() == 3292
-
+    assert os.path.exists(private_key_path)
 
 def testCipherSplit():
     encrypted_fields = okta_ciphertoken.split("$")