[Python] Verifying Amazon SNS signatures in Python

I am struggling to get a successful signature validation when I send a test message through complaint@simulator.amazonses.com. Below is my current relevant Python code after some trial and error. But I always get an InvalidSignature error on signing_cert.public_key().verify. Any ideas on what I am doing wrong?

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

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

string_to_sign += f"""Timestamp:\n{payload['Timestamp']}\nTopicArn:\n{payload['TopicArn']}\nType:\n{payload['Type']}\n"""

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

print(string_to_sign)

return string_to_sign


class AmazonSNSSESWebhookView(WebhookView):
"""
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

# We need to handle the 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
else:
return True

# Message Type: Ensure that you only process SNS messages of type Notification.
# There are other message types (e.g., SubscriptionConfirmation and UnsubscribeConfirmation),
# which you may want to handle separately. For SubscriptionConfirmation,
# you should respond to confirm the subscription.
if payload["Type"] != "Notification":
return False

# Check that the message is recent, protect against replay attacks
# Ignore if it is old
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 the 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()
)

# Sign the string.
string_to_sign = create_string_to_sign(payload)

return signing_cert.public_key().verify(
decoded_signature,
string_to_sign.encode("UTF-8"),
padding=padding.PKCS1v15(),
algorithm=signature_hash,
)

Continue reading...