Cognito邮件MFA实现报错：aws-encryption-sdk-python提示65无效SerializationVersion

我正在用Python Lambda实现AWS Cognito的邮件MFA功能，遵循Cognito自定义短信/邮件发送者的官方文档。当用户输入用户名密码触发Cognito调用Lambda时，使用aws-encryption-sdk-python解密验证码时遇到以下错误：

Traceback (most recent call last):
  File "/var/task/aws_encryption_sdk/internal/formatting/deserialize.py", line 87, in _verified_version_from_id
    return SerializationVersion(version_id)
  File "/var/lang/lib/python3.9/enum.py", line 384, in __call__
    return cls.__new__(cls, value)
  File "/var/lang/lib/python3.9/enum.py", line 702, in __new__
    raise ve_exc
ValueError: 65 is not a valid SerializationVersion

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/var/task/aws_encryption_sdk/__init__.py", line 186, in decrypt
    plaintext = decryptor.read()
  File "/var/task/aws_encryption_sdk/streaming_client.py", line 250, in read
    self._prep_message()
  File "/var/task/aws_encryption_sdk/streaming_client.py", line 782, in _prep_message
    self._header, self.header_auth = self._read_header()
  File "/var/task/aws_encryption_sdk/streaming_client.py", line 797, in _read_header
    header, raw_header = deserialize_header(self.source_stream, self.config.max_encrypted_data_keys)
  File "/var/task/aws_encryption_sdk/internal/formatting/deserialize.py", line 336, in deserialize_header
    version = _verified_version_from_id(version_id)
  File "/var/task/aws_encryption_sdk/internal/formatting/deserialize.py", line 89, in _verified_version_from_id
    raise NotSupportedError("Unsupported version {}".format(version_id), error)
aws_encryption_sdk.exceptions.NotSupportedError: ('Unsupported version 65', ValueError('65 is not a valid SerializationVersion'))

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/var/task/aws_encryption_sdk/streaming_client.py", line 218, in __exit__
    self.close()
  File "/var/task/aws_encryption_sdk/streaming_client.py", line 985, in close
    raise SerializationError("Footer not read")
aws_encryption_sdk.exceptions.SerializationError: Footer not read

我并未手动设置过版本65，以下是我的Lambda代码（已确认环境变量传递正确）：

import os
import json
import boto3
from botocore.exceptions import ClientError
from common.utils import *
from common.sqlUtils import *
from common.authentication import *
from common.authorization import *
from common.exceptions import GeneralException
from sendgrid import SendGridAPIClient
from sendgrid.helpers.mail import Mail
import aws_encryption_sdk
from aws_encryption_sdk.identifiers import CommitmentPolicy

# Configure the encryption SDK client with the KMS key from the environment variables.
awsEncryptionSdkClient = aws_encryption_sdk.EncryptionSDKClient(
    commitment_policy=CommitmentPolicy.REQUIRE_ENCRYPT_REQUIRE_DECRYPT
)
decryptionKeyArn = os.environ["cognitoCodeEncryptionKeyArn"]
kms_key_provider = aws_encryption_sdk.StrictAwsKmsMasterKeyProvider(
    key_ids=[decryptionKeyArn]
)

def sendEmail(event, context):
    try:
        #TODO check if email is verified, user is confirmed, etc

        plaintextCode = None
        if "request" in event and "code" in event["request"]:
            print("Line 35: ", event)
            encryptedCode = event["request"]["code"]
            print("encryptedCode: ", encryptedCode)
            print("decryptionKeyArn ", decryptionKeyArn)
            plaintextCode, plaintextHeader = awsEncryptionSdkClient.decrypt(
                source=encryptedCode,
                key_provider=kms_key_provider
            )
            print("plaintextCode:", plaintextCode)

        subject = None
        html_content = None
        if plaintextCode is not None:
            subject = 'Code'
            html_content = f'<strong>Your code is: {plaintextCode}</strong>'

        message = Mail(
            from_email='no-reply@mydomain.com',
            to_emails=event["request"]["userAttributes"]["email"],
            subject=subject,
            html_content=html_content
        )

        sendgridSecret = getSecret(os.environ['cognitoSendgridSecretArn'])
        if isJson(sendgridSecret):
            sendgridSecret = json.loads(sendgridSecret)['SENDGRID_API_KEY']

        sg = SendGridAPIClient(sendgridSecret)
        response = sg.send(message)
        print(response.status_code)
        print(response.body)
        print(response.headers)

        #TODO check if email was sent successfully

        return json_response({"sendgrid message": response})

    except Exception as e:
        httpCode = 500
        if isinstance(e, GeneralException):
            httpCode = e.httpCode
        print(str(e))
        return json_response({"message": str(e)}, httpCode)

解决提示 #

• 修正密文格式处理：Cognito传递的code是Base64编码的字符串，必须先解码为字节流再传入解密方法。直接传入字符串会导致SDK将字符的ASCII码当作加密格式版本解析（65是字符'A'的ASCII码），引发错误。修正代码如下：

  import base64
  # ...
  encryptedCode = event["request"]["code"]
  # 解码Base64字符串为字节
  encrypted_bytes = base64.b64decode(encryptedCode)
  plaintextCode, plaintextHeader = awsEncryptionSdkClient.decrypt(
      source=encrypted_bytes,
      key_provider=kms_key_provider
  )
  # 将解密后的字节转为字符串
  plaintextCode = plaintextCode.decode('utf-8')
  

• 验证KMS密钥权限：确保Lambda执行角色拥有该KMS密钥的kms:Decrypt权限，否则即使格式正确也会解密失败。

• 检查SDK版本：使用最新稳定版的aws-encryption-sdk-python，避免版本不匹配导致的序列化兼容问题。

内容的提问来源于stack exchange，提问作者JustANoob