使用Python3.9+Pipedream开发WhatsApp Flow时加密响应报错求助

我正在构建用于获取订单的WhatsApp Flow，能成功接收请求并解密消息，但加密响应发往WhatsApp时失败，报错：Invalid response from endpoint。使用Python 3.9 + Pipedream，求解决思路。

我的加密代码 #

from base64 import b64decode, b64encode
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
import json

def handler(pd: "pipedream"):
    # 获取解密后的AES密钥和IV
    aes_key_b64 = pd.steps["Decrypt_WhatsApp_Key"]["$return_value"]["decrypted_aes_key"]
    iv_b64 = pd.steps["trigger"]["event"]["body"]["initial_vector"]

    # 将AES密钥和IV从base64解码为字节
    aes_key = b64decode(aes_key_b64)
    iv = b64decode(iv_b64)

    # 生成翻转后的IV
    iv_flipped = flip_iv(iv)

    # 构造响应内容
    response = {
        "version": "3.0",
        "screen": "SUCCESS",
        "data": {
            "extension_message_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)

    # 加密响应
    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 {
        "status": 200,
        "body": encrypted_response,
        "headers": {
            "Content-Type": "application/json"
        }
    }

def flip_iv(iv):
    flipped_bytes = []
    for byte in iv:
        flipped_byte = byte ^ 0xFF
        flipped_bytes.append(flipped_byte)
    return bytes(flipped_bytes)

排查与修复方案 #

• 修正IV翻转逻辑：WhatsApp Flow要求响应使用的IV是原始IV前8字节保留，最后4字节每个字节按位取反（异或0xFF），而非全字节翻转。你的当前逻辑错误地翻转了所有12字节，导致加密IV不匹配。修正后的flip_iv函数：

  def flip_iv(iv):
      if len(iv) != 12:
          raise ValueError("IV must be 12 bytes for AES-GCM")
      # 前8字节不变，最后4字节逐字节异或0xFF
      return iv[:8] + bytes(byte ^ 0xFF for byte in iv[8:])
  

• 修正响应格式：WhatsApp期望接收的是包含encrypted_data字段的JSON对象，而非直接返回base64加密字符串。修改返回部分：

  return {
      "status": 200,
      "body": json.dumps({"encrypted_data": encrypted_response}),
      "headers": {
          "Content-Type": "application/json"
      }
  }
  

• 验证flow_token有效性：确认flow_token确实来自解密后的请求数据，步骤引用pd.steps["Decrypt_WhatsApp_Message"]["$return_value"]["flow_token"]无拼写或路径错误。
• 检查JSON序列化完整性：确保响应内容response序列化无异常，可添加异常捕获：

  try:
      response = json.dumps(response)
  except Exception as e:
      print(f"JSON序列化失败: {e}")
      return {"status": 500, "body": str(e)}
  

• 确认加密流程正确性：AES-GCM加密的结果必须是密文 + 16字节标签，你的当前代码逻辑正确，但可打印加密前后的长度、base64结果进行本地验证。

内容的提问来源于stack exchange，提问作者Giovanni Còco