Python中验证Amazon SNS签名失败问题排查求助

使用complaint@simulator.amazonses.com发送测试消息时，始终无法完成Amazon SNS签名验证，执行signing_cert.public_key().verify时持续触发InvalidSignature错误。以下是调试后的Python代码，请求协助排查问题：

from cryptography import x509
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import padding
import json
import requests
from datetime import datetime, timedelta
import base64
from dateutil import parser
from django.conf import settings
from django.views.generic import View
import logging

logger = logging.getLogger(__name__)

def create_string_to_sign(payload):
    string_to_sign = (
        f"""Message:
{payload['Message']}
MessageId:
{payload['MessageId']}
"""
    )

    # 'Subject' is optional, add it only if present in the payload
    if "Subject" in payload:
        string_to_sign += f"Subject:
{payload['Subject']}
"

    string_to_sign += f"""Timestamp:
{payload['Timestamp']}
TopicArn:
{payload['TopicArn']}
Type:
{payload['Type']}
"""

    # Add 'UnsubscribeURL' only if present
    if "UnsubscribeURL" in payload:
        string_to_sign += f"UnsubscribeURL:
{payload['UnsubscribeURL']}
"

    return string_to_sign


class AmazonSNSSESWebhookView(View):
    """
    Validate and process webhook events from Amazon SNS for Amazon SES spam complaints.
    """

    def validate(self):
        """
        Sample payload from Amazon SNS
        {
        "Type" : "Notification",
        "MessageId" : "1c2a7465-1f6b-43a2-b92f-0f24b9c7f3c5",
        "TopicArn" : "arn:aws:sns:us-east-1:123456789012:SES_SpamComplaints",
        "Message" : "{\"notificationType\":\"Complaint\",\"complaint\":{\"complainedRecipients\":[{\"emailAddress\":\"example@example.com\"}],\"complaintFeedbackType\":\"abuse\",\"arrivalDate\":\"2024-09-25T14:00:00.000Z\"},\"mail\":{\"timestamp\":\"2024-09-25T13:59:48.000Z\",\"source\":\"sender@example.com\",\"messageId\":\"1234567890\"}}",
        "Timestamp" : "2024-09-25T14:00:00.000Z",
        "SignatureVersion" : "1",
        "Signature" : "...",
        "SigningCertURL" : "https://sns.us-east-1.amazonaws.com/SimpleNotificationService.pem",
        "UnsubscribeURL" : "https://sns.us-east-1.amazonaws.com/?Action=Unsubscribe"
        }
        """
        payload = json.loads(self.request.body)

        # Ensure that the sender is actually Amazon SNS
        signing_cert_url = payload["SigningCertURL"]
        if not signing_cert_url.startswith(
            f"https://sns.{settings.AWS_SES_REGION_NAME}.amazonaws.com/"
        ):
            return False

        # Handle subscription confirmation
        if payload["Type"] == "SubscriptionConfirmation":
            response = requests.get(payload["SubscribeURL"])
            if response.status_code != 200:
                logger.error(f"Failed to confirm Amazon SNS subscription. Payload: {payload}")
                return False
            return True

        # Only process Notification type messages
        if payload["Type"] != "Notification":
            return False

        # Replay attack protection: ignore old messages
        sns_datetime = parser.parse(payload["Timestamp"])
        current_datetime = datetime.now(timezone.utc)
        time_window = timedelta(minutes=5)
        if sns_datetime < current_datetime - time_window:
            return False

        # Retrieve and load signing certificate
        signing_cert = x509.load_pem_x509_certificate(
            requests.get(signing_cert_url).content
        )
        decoded_signature = base64.b64decode(payload["Signature"])
        signature_hash = hashes.SHA1() if payload["SignatureVersion"] == "1" else hashes.SHA256()

        # Generate string to sign
        string_to_sign = create_string_to_sign(payload)

        try:
            signing_cert.public_key().verify(
                decoded_signature,
                string_to_sign.encode("UTF-8"),
                padding=padding.PKCS1v15(),
                algorithm=signature_hash,
            )
            return True
        except Exception as e:
            logger.error(f"Signature verification failed: {str(e)}")
            return False

可能的问题及修复方案 #

1. 待签名字符串格式不符合SNS规范 #

Amazon SNS对生成待签名字符串的格式要求极其严格，包括字段顺序、换行符数量、末尾是否有多余换行等：

• 原代码在拼接字符串时，每个字段后都添加了额外换行（比如UnsubscribeURL字段最后多了一个换行），而SNS要求最后一个字段后不能有多余换行。
• 改用列表拼接的方式确保格式精准：

def create_string_to_sign(payload):
    components = [
        f"Message:\n{payload['Message']}",
        f"MessageId:\n{payload['MessageId']}"
    ]
    if "Subject" in payload:
        components.append(f"Subject:\n{payload['Subject']}")
    components.extend([
        f"Timestamp:\n{payload['Timestamp']}",
        f"TopicArn:\n{payload['TopicArn']}",
        f"Type:\n{payload['Type']}"
    ])
    if "UnsubscribeURL" in payload:
        components.append(f"UnsubscribeURL:\n{payload['UnsubscribeURL']}")
    # 用\n连接所有组件，确保无多余末尾换行
    return "\n".join(components)

2. Message字段的原始值被修改 #

SNS签名时使用的是Message字段的原始JSON字符串值（包括转义字符），但原代码通过json.loads(self.request.body)解析后，payload['Message']会自动去掉转义符，导致待签名字符串与SNS签名时的内容不一致。

• 解决方法：从原始请求体中提取Message的原始字符串，避免解析时丢失转义：

raw_payload = self.request.body.decode('utf-8')
payload = json.loads(raw_payload)
# 用正则提取原始Message字段值
import re
match = re.search(r'"Message"\s*:\s*"([^"]*(?:\\.[^"]*)*)"', raw_payload)
if match:
    original_message = match.group(1).encode('utf-8').decode('unicode_escape')
else:
    original_message = payload['Message']
# 后续拼接时使用original_message代替payload['Message']

3. 证书验证不完整 #

原代码仅检查了证书URL前缀，未验证证书的有效性和颁发者，可能导致使用篡改后的证书：

• 添加证书有效性校验：

from cryptography.x509.oid import NameOID
# 验证证书颁发者
issuer = signing_cert.issuer.get_attributes_for_oid(NameOID.COMMON_NAME)[0].value
if issuer != "Amazon Simple Notification Service":
    logger.error("Invalid certificate issuer")
    return False
# 验证证书有效期
if signing_cert.not_valid_after < datetime.now(timezone.utc):
    logger.error("Certificate has expired")
    return False

4. 异常处理缺失 #

原代码中verify方法失败会直接抛出异常，添加try-except块可以捕获具体错误信息，方便排查问题（已在修改后的代码中添加）。

内容的提问来源于stack exchange，提问作者Botond Bérès