Add AES GCM Python utility

utilities/aes-gcm-python-utility/.gitignore

@@ -0,0 +1,3 @@
+__pycache__
+.DS_Store
+.coverage

utilities/aes-gcm-python-utility/app.py

@@ -0,0 +1,26 @@
+import encryption_util
+import key_util
+import builtins
+
+if __name__=="__main__":
+    println=builtins.print
+    keypair1=key_util.generate_key_pair()
+    keypair2=key_util.generate_key_pair()
+
+    shared_key1=key_util.generate_shared_key(keypair1.private_key,keypair2.public_key)
+    shared_key2=key_util.generate_shared_key(keypair2.private_key,keypair1.public_key)
+    println("shared_key1:",shared_key1)
+    println("shared_key2:",shared_key2)
+    println("shared_key1==shared_key2 ==>",shared_key1==shared_key2)
+
+    raw_data = "Hello This is ONDC Test Data"
+
+    println("-----------------------------------------------")
+    encrypted_data=encryption_util.encrypt_data(shared_key1,raw_data)
+    println("Encrypted Data ===> ", encrypted_data)
+    println("-----------------------------------------------")
+
+    decrypted_data=encryption_util.decrypt_data(shared_key2,encrypted_data)
+    println("decrypted Data ===> ",decrypted_data)
+    println("-----------------------------------------------")
+

utilities/aes-gcm-python-utility/encryption_util.py

@@ -0,0 +1,78 @@
+import base64
+import json
+from Cryptodome.Cipher import AES
+from Crypto.Random import get_random_bytes
+import logging
+
+def encrypt_data(key, data):
+    '''
+    Encrypt the specified plain text using AES/GCM/NoPadding.
+
+    Parameters:
+
+    ``key`` (string): The Shared Key.
+
+    ``data`` (string): The Raw Data to be Encrypted.
+
+    Returns:
+    string: The Encrypted data in base64 encoded string format
+    '''
+    # The standard Initialization Vector (IV) length (96 bits) (12 byte).
+    IV_BYTE_LENGTH=12
+    encrypted_data=None
+    try:
+        shared_key = base64.b64decode(key)  
+        nonce = get_random_bytes(IV_BYTE_LENGTH) # Randomly generate the IV/nonce 
+
+        #  Initialize AES/GCM cipher for encryption
+        cipher = AES.new(shared_key, AES.MODE_GCM, nonce=nonce)
+        # Encrypt the raw data and get the cipher text and authentication tag.
+        ciphertext, auth_tag = cipher.encrypt_and_digest(data.encode())
+
+        #  Set the values for the EncryptedData 
+        encrypted_payload = {
+            'nonce': base64.b64encode(cipher.nonce).decode("utf-8"),
+            'encrypted_data': base64.b64encode(ciphertext).decode("utf-8"),
+            'hmac': base64.b64encode(auth_tag).decode("utf-8")
+        }
+        encrypted_data=base64.b64encode(json.dumps(encrypted_payload).encode()).decode("utf-8")
+    except Exception as e:
+        logging.exception(e)
+
+    # Return the Encrypted Data.    
+    return encrypted_data
+
+def decrypt_data(key, e_data):
+    '''
+        Decrypts the Encrypted Data using Shared Key.
+
+        Parameters:
+
+        ``key`` (string): The Shared Key
+
+        ``data`` (string): The Encrypted Data.
+
+        Returns:
+        string: The Raw Decrypted data
+    '''
+    decrypted_data=None
+    try:
+
+        shared_key = base64.b64decode(key)
+
+        # Decode the base64 string and De-serialize it as
+        decoded_payload = json.loads(base64.b64decode(e_data))
+
+        # Decode the fields of encryptedData from base64 to bytes.
+        nonce = base64.b64decode(decoded_payload["nonce"])
+        encrypted_data = base64.b64decode( decoded_payload["encrypted_data"])
+        auth_tag =base64.b64decode( decoded_payload["hmac"])
+
+        cipher = AES.new(shared_key, AES.MODE_GCM, nonce=nonce)
+        # Decrypt the data
+        plaintext = cipher.decrypt_and_verify(encrypted_data, auth_tag)
+        decrypted_data=plaintext.decode('utf-8')
+    except Exception as e:
+        logging.exception(e)
+    return decrypted_data
+

utilities/aes-gcm-python-utility/key_util.py

@@ -0,0 +1,79 @@
+import base64
+from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
+from cryptography.hazmat.primitives import serialization
+import logging
+
+class DHKeyPair:
+    '''
+    DHKeyPair class stores a pair of public_key and private_key.
+
+    Attributes:
+        ``private_key`` (string): The Private Key.
+
+        ``public_key`` (string): The Public Key.
+    '''
+
+    def __init__(self, private_key, public_key):
+        self.private_key = private_key
+        self.public_key = public_key
+
+def generate_key_pair():
+    '''
+        Generate a Keypair
+
+        Returns:
+        DHKeyPair: public_key and private_key 
+    '''
+
+    keypair= DHKeyPair(None,None)
+    try:
+        inst_private_key = X25519PrivateKey.generate()
+        inst_public_key = inst_private_key.public_key()
+
+        bytes_private_key = inst_private_key.private_bytes(
+            encoding=serialization.Encoding.DER,
+            format=serialization.PrivateFormat.PKCS8,
+            encryption_algorithm=serialization.NoEncryption()
+        )
+
+        bytes_public_key = inst_public_key.public_bytes(
+            encoding=serialization.Encoding.DER,
+            format=serialization.PublicFormat.SubjectPublicKeyInfo
+        )
+        private_key = base64.b64encode(bytes_private_key).decode('utf-8')
+        public_key = base64.b64encode(bytes_public_key).decode('utf-8')
+        keypair.private_key=private_key
+        keypair.public_key=public_key
+    except Exception as e:
+        logging.exception(e)
+
+    return keypair
+
+def generate_shared_key(private_key_str, public_key_str):
+    '''
+        Generate a SharedKey.
+
+        Parameters:
+
+        ``private_key_str`` (string): Private Key of one party.
+
+        ``public_key_str`` (string): Public Key of the other party.
+
+        Returns:
+        string: shared_key in base64 encoded string format
+    '''
+    shared_key=None
+    try:
+        private_key = serialization.load_der_private_key(
+                base64.b64decode(private_key_str),
+                password=None
+        )
+        public_key = serialization.load_der_public_key(
+            base64.b64decode(public_key_str)
+        )
+        shared_key = private_key.exchange(public_key)
+        shared_key = base64.b64encode(shared_key).decode('utf-8')
+    except Exception as e:
+        logging.exception(e)
+
+    return shared_key

utilities/aes-gcm-python-utility/readme.md

@@ -0,0 +1,116 @@
+# AES GCM Encryption and Decryption
+
+Instructions for using the utility
+
+## Add the following packages to your environment:
+
+- [pycryptodomex](https://pypi.org/project/pycryptodomex/)
+- [cryptography](https://pypi.org/project/cryptography/)
+- [pycryptodome](https://pypi.org/project/pycryptodome/)
+
+Add the following files to your project:
+
+```
+key_util.py
+encryption_util.py
+```
+
+Make sure to import both of these files to your project.  
+1\. Make sure to import key_util to generate key_pair and to generate shared_key.  
+2\. Import encryption_util for encryption of raw_data and for decryption of encrypted data.
+
+## Generating a Key Pair
+
+1\. Generate Encryption Key Pair:
+
+```python
+key_pair = key_util.generate_key_pair()
+```
+
+2\. Save your Private key:
+
+```python
+private_key = key_pair.private_key
+```
+
+3\. Share your Public key with the other party:
+
+```python
+public_key = key_pair.public_key
+```
+
+## Generating a Shared key between two parties:
+
+1\. Get the public key from the other party:
+
+```python
+public_key_of_other_party = "PUBLIC_KEY_GOES_HERE"
+```
+
+2\. Generate the sharedKey using your private key and public key of other party:
+
+```python
+my_private_key = "YOUR_PRIVATE_KEY_GOES_HERE"
+
+shared_key = key_util.generate_shared_key(my_private_key, public_key_of_other_party)
+```
+
+3\. This shared key will be used for both encryption and decryption. The same sharedKey can be generated by the other party if your publicKey is shared with them.
+
+## Encryption
+
+1\. Get the public key of the BPP:
+
+```python
+bpp_public_key = "BPP_PUBLIC_KEY_GOES_HERE"
+```
+
+2\. Generate the sharedKey using the BPP's public key and your private key:
+
+```python
+my_private_key = "YOUR_PRIVATE_KEY_GOES_HERE"
+
+shared_key = key_util.generate_shared_key(my_private_key, bpp_public_key)
+```
+
+3\. Encrypt the data:
+
+```python
+raw_data = "Hello This is ONDC Test Data"
+
+encrypted_data = encryption_util.encrypt_data(shared_key, raw_data)
+```
+
+4\. Send the Encrypted data and your public key to the BPP.
+
+## Decryption
+
+1\. Get the Public key of the BAP:
+
+```python
+bap_public_key = "BAP_PUBLIC_KEY_GOES_HERE"
+```
+
+2\. Generate the shared key using your private key and BAP's public key:
+
+```python
+my_private_key = "YOUR_PRIVATE_KEY_GOES_HERE"
+
+shared_key = key_util.generate_shared_key(my_private_key, bap_public_key)
+```
+
+3\. Decrypt the data:
+
+```python
+encrypted_data = "ENCRYPTED_DATA_RECEIVED_FROM_BAP"
+
+decrypted_data = encryption_util.decrypt_data(shared_key, encrypted_data)
+
+print(decrypted_data) # Hello This is ONDC Test Data
+```
+
+## Abbrevations
+
+BAP - Beckn Application Platform (Sender)
+
+BPP - Beckn Provider Platform (Receiver)

utilities/aes-gcm-python-utility/test_encryption_util.py

@@ -0,0 +1,39 @@
+import unittest
+import encryption_util
+import key_util
+
+
+class TestEnc(unittest.TestCase):
+
+    def test_generate_shared_key(self):
+        keypair1=key_util.generate_key_pair()
+        keypair2=key_util.generate_key_pair()
+
+        # positive testcases
+        shared_key1=key_util.generate_shared_key(keypair1.private_key,keypair2.public_key)
+        shared_key2=key_util.generate_shared_key(keypair2.private_key,keypair1.public_key)
+        self.assertEqual(shared_key1,shared_key2,"must be same")
+
+        #negative testcase
+        shared_key2=key_util.generate_shared_key(keypair1.private_key,keypair1.public_key)
+        self.assertNotEqual(shared_key1,shared_key2,"should be different")
+
+    def test_encrypt_decrypt(self):
+        keypair1=key_util.generate_key_pair()
+        keypair2=key_util.generate_key_pair()
+
+        shared_key=key_util.generate_shared_key(keypair1.private_key,keypair2.public_key)
+
+        # positive testcases
+        raw_data = "Hello This is ONDC Test data"
+        encrypted_data=encryption_util.encrypt_data(shared_key,raw_data)
+        decrypted_data=encryption_util.decrypt_data(shared_key,encrypted_data)
+        self.assertEqual(decrypted_data,raw_data,"decrypted text and raw tax must be same")
+
+        #negative testcase
+        invalid_shared_key='YR7XT2fAQR2WSQGf/G/JSLZ+LDuULMdYJI7ZDLGa3H4='
+        decrypted_data=encryption_util.decrypt_data(invalid_shared_key,encrypted_data)
+        self.assertNotEqual(decrypted_data,raw_data,'should be different')
+
+if __name__=='__main__':
+    unittest.main()