Last active
December 28, 2025 00:16
-
-
Save giovannicocco/23505cf648fce1e92a178cae567cc78d to your computer and use it in GitHub Desktop.
WhatsApp Flow (Encrypt response)
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| from base64 import b64decode, b64encode | |
| from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes | |
| import json | |
| def flip_iv(iv): | |
| flipped_bytes = [] | |
| for byte in iv: | |
| flipped_byte = byte ^ 0xFF | |
| flipped_bytes.append(flipped_byte) | |
| return bytes(flipped_bytes) | |
| def handler(pd): | |
| # GET AES Key | |
| aes_key_b64 = pd.steps["Decrypt_WhatsApp_Key"]["$return_value"]["decrypted_aes_key"] | |
| print("Decrypt AES Key:", aes_key_b64) | |
| # Variable iv_base64 like initial IV | |
| iv_base64 = pd.steps["trigger"]["event"]["body"]["initial_vector"] | |
| # Decoding AES key and base64 IV to bytes | |
| aes_key = b64decode(aes_key_b64) | |
| iv = b64decode(iv_base64) | |
| # Inverting the IV | |
| iv_flipped = flip_iv(iv) | |
| print("Inverted IV:", b64encode(iv_flipped).decode("utf-8")) | |
| # Adding prints to check the values of the "flow_token" and "status" fields | |
| #params = { | |
| #"flow_token": pd.steps["Decrypt_WhatsApp_Message"]["$return_value"]["flow_token"], | |
| #"status": pd.steps["shopify_developer_app"]["$return_value"]["orders"][0]["id"] | |
| #} | |
| #print("Flow_token value:", params["flow_token"]) | |
| #print("Status value:", params["status"]) | |
| response = { | |
| "version": "3.0", | |
| "screen": "SUCCESS", | |
| "data": { | |
| "extension_messge_response": { | |
| "params": { | |
| "flow_token": pd.steps["Decrypt_WhatsApp_Message"]["$return_value"]["flow_token"], | |
| "status": pd.steps["shopify_developer_app"]["$return_value"]["orders"][0]["id"] | |
| } | |
| } | |
| } | |
| } | |
| response = json.dumps(response) | |
| print (response) | |
| # Encrypting the response | |
| cipher = Cipher(algorithms.AES(aes_key), modes.GCM(iv_flipped)) | |
| encryptor = cipher.encryptor() | |
| encrypted = encryptor.update(response.encode("utf-8")) + encryptor.finalize() + encryptor.tag | |
| encrypted_response = b64encode(encrypted).decode("utf-8") | |
| return encrypted_response, 200, {"Content-Type": "application/json"} |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment