Gettig error when we do data_exchange in whatsapp flow

Question

I face one error when I was doing data_exchange. I used facebook's example of data_exchange.

I am getting this error

We failed to decrypt your payload. Please check your encryption/decryption logic.

Here it is my decryption code and it is working fine.

def decrypt_request(self,encrypted_flow_data_b64, encrypted_aes_key_b64, initial_vector_b64):
    flow_data = b64decode(encrypted_flow_data_b64)
    iv = b64decode(initial_vector_b64)

    # Decrypt the AES encryption key
    encrypted_aes_key = b64decode(encrypted_aes_key_b64)
    
    PRIVATE_KEY = request.env['res.company'].search([],limit=1)
    encoded_data = b64decode(PRIVATE_KEY.private_key_attachment).decode('utf-8').encode('utf-8')
    private_key = load_pem_private_key(
        encoded_data, password=None)
    aes_key = private_key.decrypt(encrypted_aes_key, OAEP(
        mgf=MGF1(algorithm=hashes.SHA256()), algorithm=hashes.SHA256(), label=None))

    # Decrypt the Flow data
    encrypted_flow_data_body = flow_data[:-16]
    encrypted_flow_data_tag = flow_data[-16:]
    decryptor = Cipher(algorithms.AES(aes_key),
                       modes.GCM(iv, encrypted_flow_data_tag)).decryptor()
    decrypted_data_bytes = decryptor.update(
        encrypted_flow_data_body) + decryptor.finalize()
    decrypted_data = json.loads(decrypted_data_bytes.decode("utf-8"))
    return decrypted_data, aes_key, iv`

Here it is my encryption code:

def encrypt_response(self,response, aes_key, iv):
    # Flip the initialization vector
    flipped_iv = bytearray(b ^ 0xFF for b in iv)
    
    # Encrypt the response data
    encryptor = Cipher(
        algorithms.AES(aes_key),
        modes.GCM(flipped_iv[:12]),
        backend=default_backend()
    ).encryptor()
    
    # Encrypt and finalize
    response_bytes = json.dumps(response).encode("utf-8")
    ciphertext = encryptor.update(response_bytes) + encryptor.finalize()
    
    # Combine ciphertext and tag
    encrypted_payload = ciphertext + encryptor.tag
    
    # Encode as base64
    encoded_payload = b64encode(encrypted_payload).decode("utf-8")
    
    return encoded_payload

This is main method:

def whatsapp_data_exchange_flow(self,**kw):
    data = json.loads(request.httprequest.data)

    # Read the request fields
    encrypted_flow_data_b64 = data['encrypted_flow_data']
    encrypted_aes_key_b64 = data['encrypted_aes_key']
    initial_vector_b64 = data['initial_vector']

    decrypted_data, aes_key, iv = self.decrypt_request(
        encrypted_flow_data_b64, encrypted_aes_key_b64, initial_vector_b64)
    print(decrypted_data)

    # Return the next screen & data to the client
    response = {
        "version": "3.0",
        "screen": "BOOKING_INFO"
    }
    encrpted_response = self.encrypt_response(response, aes_key, iv)

    headers = werkzeug.datastructures.Headers(None)
    headers['Content-Type'] = 'text/plain'
    response = request.make_response(final_response,headers=headers.to_wsgi_list(),status=200)
    return response

Answer 1

Just set your endpoint and WhatsApp flow to automatically call your endpoint. Here, you can see the code at your end is set up to decrypt the payload and encrypt the response. Also, you can see the basic ping call and INIT call response in django:

from base64 import b64decode, b64encode
from cryptography.hazmat.primitives.asymmetric.padding import OAEP, MGF1, hashes
from cryptography.hazmat.primitives.ciphers import algorithms, Cipher, modes
from cryptography.hazmat.primitives.serialization import load_pem_private_key

@csrf_exempt
def getWhatsappflowcall(request, pk):
    try:
        # Parse the request body
        body = json.loads(request.body)

        # Read the request fields
        encrypted_flow_data_b64 = body['encrypted_flow_data']
        encrypted_aes_key_b64 = body['encrypted_aes_key']
        initial_vector_b64 = body['initial_vector']

        decrypted_data, aes_key, iv = decrypt_request(
            encrypted_flow_data_b64, encrypted_aes_key_b64, initial_vector_b64)
        logger.info(f'after decrypt body got for whatsapp flow {decrypted_data} {aes_key} {iv}')

        response = generate_response(decrypted_data, pk)

        return HttpResponse(encrypt_response(response, aes_key, iv), content_type='text/plain')
    except Exception as e:
        return JsonResponse({}, status=500)


def decrypt_request(encrypted_flow_data_b64, encrypted_aes_key_b64, initial_vector_b64):
    try:
        flow_data = b64decode(encrypted_flow_data_b64)
        iv = b64decode(initial_vector_b64)

        # Decrypt the AES encryption key
        encrypted_aes_key = b64decode(encrypted_aes_key_b64)
        private_key = load_pem_private_key(
            PRIVATE_KEY.encode('utf-8'), password=None)
        aes_key = private_key.decrypt(encrypted_aes_key, OAEP(
            mgf=MGF1(algorithm=hashes.SHA256()), algorithm=hashes.SHA256(), label=None))

        # Decrypt the Flow data
        encrypted_flow_data_body = flow_data[:-16]
        encrypted_flow_data_tag = flow_data[-16:]
        decryptor = Cipher(algorithms.AES(aes_key),
                        modes.GCM(iv, encrypted_flow_data_tag)).decryptor()
        
        decrypted_data_bytes = decryptor.update(
            encrypted_flow_data_body) + decryptor.finalize()
        decrypted_data = json.loads(decrypted_data_bytes.decode("utf-8"))
        return decrypted_data, aes_key, iv
    except:
        logger.error(f'error in decrypt_request body of whatsapp flow {traceback.print_exc()}')


def encrypt_response(response, aes_key, iv):
    # Flip the initialization vector
    flipped_iv = bytearray()
    for byte in iv:
        flipped_iv.append(byte ^ 0xFF)

    # Encrypt the response data
    encryptor = Cipher(algorithms.AES(aes_key),
                       modes.GCM(flipped_iv)).encryptor()
    return b64encode(
        encryptor.update(json.dumps(response).encode("utf-8")) +
        encryptor.finalize() +
        encryptor.tag
    ).decode("utf-8")

def generate_response(input_data, pk):
    response = None
    # first check is ping call only , return status ative call

    if input_data.get("action","") == "ping":
        response = {
                        "data": {
                            "status": "active"
                        }
                    }
    return response