无法解密RSA密钥，完成WhatsApp Flows健康检查失败求助

问题描述 #

我正在配置WhatsApp Business API的流程，但卡在了端点健康检查的密钥解密环节，始终无法完成这一步。

报错信息 #

遇到的核心错误提示是：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.

复现步骤 #

我是按照官方指引一步步操作的，具体流程如下：

• 搭建好流程端点
• 生成2048位RSA密钥对，执行的命令：

openssl genrsa -des3 -out private.pem 2048

生成的私钥格式如下：

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

• 导出RSA公钥到文件：

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

• 通过API设置商家公钥：

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-----'

• 编写加解密逻辑代码：

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

• 从WhatsApp管理器界面触发健康检查

有没有大佬能帮我排查下哪里出问题了？

备注：内容来源于stack exchange，提问作者Séamus CAREY