[Python] Cannot decrypt RSA key and complete WhatsApp flows healthcheck

The issue


I am trying to setup a flow with WhatsApp Business API but am unable to complete the decryption of the key as part of the endpoint healthcheck.

I get this error: Encryption/decryption failed


in decrypt_request
aes_key = private_key.decrypt(encrypted_aes_key, OAEP(mgf=MGF1(algorithm=hashes.SHA256()), algorithm=hashes.SHA256(), label=None))
ValueError: Encryption/decryption failed.

How to reproduce the issue


Below are the steps I followed, as per the docs here and here:

1. 
   Set up the flows endpoint.
   
   
2. 
   Generate 2048-bit RSA Key Pair.
   

openssl genrsa -des3 -out private.pem 2048


The key looks like this:

-----BEGIN ENCRYPTED PRIVATE KEY-----
AAA........
BBB....
....
-----END ENCRYPTED PRIVATE KEY-----

1. Export the RSA Public Key to a file:

openssl rsa -in private.pem -outform PEM -pubout -out public.pem

1. Set business public key.

curl -X POST \
'https://graph.facebook.com/v21.0/PHONE_NUMBER_ID/whatsapp_business_encryption' \
-H 'Authorization: Bearer ACCESS_TOKEN' \
-H 'Content-Type: application/x-www-form-urlencoded' \
--data-urlencode 'business_public_key=-----BEGIN PUBLIC KEY-----
AAA
BBB
CCC
-----END PUBLIC KEY-----'

1. Create encryption/decryption code:

import json
import os
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
from fastapi import HTTPException, Request
import logging

# Load the private key string
# PRIVATE_KEY = os.environ.get('PRIVATE_KEY')
PRIVATE_KEY_PATH = "./private.pem"
with open(PRIVATE_KEY_PATH, "rb") as pem_file:

PRIVATE_KEY = pem_file.read().decode("utf-8")

async def data(request: Request):
try:
# Parse the request body
body = json.loads(await request.body())
logging.info(f"body = {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 = await 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 = {
"screen": "SCREEN_NAME",
"data": {
"some_key": "some_value"
}
}

# Return the response as plaintext
return await encrypt_response(response, aes_key, iv)
except Exception as e:
print(e)
raise HTTPException(status_code=500, detail="Internal Server Error")


async def decrypt_request(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 = load_pem_private_key(
PRIVATE_KEY.encode("utf-8"), password=os.getenv("PASSPHRASE").encode('utf-8'))
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


async 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")

1. Trigger the health check from https://business.facebook.com/latest/whatsapp_manager/flows

Can someone tell me what I'm doing wrong?

Continue reading...