Невозможно расшифровать ключ RSA и выполнить проверку работоспособности потоков WhatsApp. ⇐ Python

[Anonymous]

Я пытаюсь настроить поток с помощью WhatsApp Business API, но не могу завершить расшифровку ключа в рамках проверки работоспособности конечной точки.
Я получаю следующую ошибку:

Код: Выделить всё

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.
  
  < /ol>

  Код: Выделить всё

  openssl genrsa -des3 -out private.pem 2048

  Ключ выглядит следующим образом:
  -----НАЧАЛО ЗАШИФРОВАННОГО ЧАСТНОГО КЛЮЧА -----
  AAA...... ..
  BBB....
  ....
  -----КОНЕЦ ЗАШИФРОВАННОГО ЧАСТНОГО КЛЮЧА -----
• Экспорт открытого ключа RSA в файл:

Код: Выделить всё

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

• Установить открытый ключ компании.

Код: Выделить всё

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

• Запустите проверку работоспособности на странице https://business.facebook.com/latest/wh ... ager/flows

Может кто-нибудь сказать мне, что я делаю неправильно?

Подробнее здесь: https://stackoverflow.com/questions/793 ... ealthcheck