CVE-2023-33185: Improper Verification of Cryptographic Signature (CWE-347)
value: Django-SES is a drop-in mail backend for Django. The django_ses library implements a mail backend for Django using AWS Simple Email Service. The library exports the `SESEventWebhookView class` intended to receive signed requests from AWS to handle email bounces, subscriptions, etc. These requests are signed by AWS and are verified by django_ses, however the verification of this signature was found to be flawed as it allowed users to specify arbitrary public certificates. This issue was patched in version 3.5.0.
CWE description: The product does not verify, or incorrectly verifies, the cryptographic signature for data.
CVSS v3 base score: 5.4
Published: 2023-05-26T21:15Z
Repository: https://github.com/django-ses/django-ses
Language: Python
Files changed: django_ses/utils.py
Commit date: 2023-05-21 09:35:36 -0600
Commit message: Restrict amazonaws allowed certificate URLs.

Diff: diff --git a/django_ses/utils.py b/django_ses/utils.py
index 6cf7559..b97a1f9 100644
--- a/django_ses/utils.py
+++ b/django_ses/utils.py
@@ -1,23 +1,27 @@
 import base64
 import logging
+import re
 import warnings
 from builtins import bytes
-
-from django_ses.deprecation import RemovedInDjangoSES20Warning
-
+from urllib.error import URLError
 from urllib.parse import urlparse
 from urllib.request import urlopen
-from urllib.error import URLError
 
 from django.core.exceptions import ImproperlyConfigured
+
 from django_ses import settings
+from django_ses.deprecation import RemovedInDjangoSES20Warning
 
 logger = logging.getLogger(__name__)
 
 _CERT_CACHE = {}
 
+SES_REGEX_CERT_URL = re.compile(
+    "(?i)^https://sns\.[a-z0-9\-]+\.amazonaws\.com(\.cn)?/SimpleNotificationService\-[a-z0-9]+\.pem$"
+)
+
 
 def clear_cert_cache():
     """Clear the certificate cache.
 
     This one-liner exists to discourage imports and direct usage of
@@ -181,10 +185,21 @@ class EventMessageVerifier(object):
             return None
 
         url_obj = urlparse(cert_url)
         for trusted_domain in settings.EVENT_CERT_DOMAINS:
             parts = trusted_domain.split(".")
+            if "amazonaws.com" in trusted_domain:
+                if not SES_REGEX_CERT_URL.match(cert_url):
+                    if len(parts) < 4:
+                        return None
+                    else:
+                        logger.warning('Possible security risk for: "%s"', cert_url)
+                        logger.warning(
+                            "It is strongly recommended to configure the full domain in EVENT_CERT_DOMAINS. "
+                            "See v3.5.0 release notes for more details."
+                        )
+
             if url_obj.netloc.split(".")[-len(parts) :] == parts:
                 return cert_url
 
         return None
 
@@ -194,30 +209,32 @@ class EventMessageVerifier(object):
         This is used to verify the bounce message when it is received.
         """
 
         # Depending on the message type the fields to add to the message
         # differ so we handle that here.
-        msg_type = self._data.get('Type')
-        if msg_type == 'Notification':
+        msg_type = self._data.get("Type")
+        if msg_type == "Notification":
             fields_to_sign = [
-                'Message',
-                'MessageId',
-                'Subject',
-                'Timestamp',
-                'TopicArn',
-                'Type',
+                "Message",
+                "MessageId",
+                "Subject",
+                "Timestamp",
+                "TopicArn",
+                "Type",
             ]
-        elif (msg_type == 'SubscriptionConfirmation' or
-              msg_type == 'UnsubscribeConfirmation'):
+        elif (
+            msg_type == "SubscriptionConfirmation"
+            or msg_type == "UnsubscribeConfirmation"
+        ):
             fields_to_sign = [
-                'Message',
-                'MessageId',
-                'SubscribeURL',
-                'Timestamp',
-                'Token',
-                'TopicArn',
-                'Type',
+                "Message",
+                "MessageId",
+                "SubscribeURL",
+                "Timestamp",
+                "Token",
+                "TopicArn",
+                "Type",
             ]
         else:
             # Unrecognized type
             logger.warning('Unrecognized SNS message Type: "%s"', msg_type)
             return None
@@ -235,18 +252,18 @@ class EventMessageVerifier(object):
         return "".join(bytes_to_sign).encode()
 
 
 def BounceMessageVerifier(*args, **kwargs):
     warnings.warn(
-        'utils.BounceMessageVerifier is deprecated. It is renamed to EventMessageVerifier.',
+        "utils.BounceMessageVerifier is deprecated. It is renamed to EventMessageVerifier.",
         RemovedInDjangoSES20Warning,
     )
 
     # parameter name is renamed from bounce_dict to notification.
-    if 'bounce_dict' in kwargs:
-        kwargs['notification'] = kwargs['bounce_dict']
-        del kwargs['bounce_dict']
+    if "bounce_dict" in kwargs:
+        kwargs["notification"] = kwargs["bounce_dict"]
+        del kwargs["bounce_dict"]
 
     return EventMessageVerifier(*args, **kwargs)
 
 
 def verify_event_message(notification):
@@ -260,33 +277,34 @@ def verify_event_message(notification):
 def verify_bounce_message(msg):
     """
     Verify an SES/SNS bounce(event) notification message.
     """
     warnings.warn(
-        'utils.verify_bounce_message is deprecated. It is renamed to verify_event_message.',
+        "utils.verify_bounce_message is deprecated. It is renamed to verify_event_message.",
         RemovedInDjangoSES20Warning,
     )
     return verify_event_message(msg)
 
 
 def confirm_sns_subscription(notification):
     logger.info(
-        'Received subscription confirmation: TopicArn: %s',
-        notification.get('TopicArn'),
+        "Received subscription confirmation: TopicArn: %s",
+        notification.get("TopicArn"),
         extra={
-            'notification': notification,
+            "notification": notification,
         },
     )
 
     # Get the subscribe url and hit the url to confirm the subscription.
-    subscribe_url = notification.get('SubscribeURL')
+    subscribe_url = notification.get("SubscribeURL")
     try:
         urlopen(subscribe_url).read()
     except URLError as e:
         # Some kind of error occurred when confirming the request.
         logger.error(
-            'Could not confirm subscription: "%s"', e,
+            'Could not confirm subscription: "%s"',
+            e,
             extra={
-                'notification': notification,
+                "notification": notification,
             },
             exc_info=True,
         )

Vulnerable code: 
from django_ses.deprecation import RemovedInDjangoSES20Warning

from urllib.error import URLError
        msg_type = self._data.get('Type')
        if msg_type == 'Notification':
                'Message',
                'MessageId',
                'Subject',
                'Timestamp',
                'TopicArn',
                'Type',
        elif (msg_type == 'SubscriptionConfirmation' or
              msg_type == 'UnsubscribeConfirmation'):
                'Message',
                'MessageId',
                'SubscribeURL',
                'Timestamp',
                'Token',
                'TopicArn',
                'Type',
        'utils.BounceMessageVerifier is deprecated. It is renamed to EventMessageVerifier.',
    if 'bounce_dict' in kwargs:
        kwargs['notification'] = kwargs['bounce_dict']
        del kwargs['bounce_dict']
        'utils.verify_bounce_message is deprecated. It is renamed to verify_event_message.',
        'Received subscription confirmation: TopicArn: %s',
        notification.get('TopicArn'),
            'notification': notification,
    subscribe_url = notification.get('SubscribeURL')
            'Could not confirm subscription: "%s"', e,
                'notification': notification,
Fixed code: import re
from urllib.error import URLError

from django_ses.deprecation import RemovedInDjangoSES20Warning
SES_REGEX_CERT_URL = re.compile(
    "(?i)^https://sns\.[a-z0-9\-]+\.amazonaws\.com(\.cn)?/SimpleNotificationService\-[a-z0-9]+\.pem$"
)

            if "amazonaws.com" in trusted_domain:
                if not SES_REGEX_CERT_URL.match(cert_url):
                    if len(parts) < 4:
                        return None
                    else:
                        logger.warning('Possible security risk for: "%s"', cert_url)
                        logger.warning(
                            "It is strongly recommended to configure the full domain in EVENT_CERT_DOMAINS. "
                            "See v3.5.0 release notes for more details."
                        )

        msg_type = self._data.get("Type")
        if msg_type == "Notification":
                "Message",
                "MessageId",
                "Subject",
                "Timestamp",
                "TopicArn",
                "Type",
        elif (
            msg_type == "SubscriptionConfirmation"
            or msg_type == "UnsubscribeConfirmation"
        ):
                "Message",
                "MessageId",
                "SubscribeURL",
                "Timestamp",
                "Token",
                "TopicArn",
                "Type",
        "utils.BounceMessageVerifier is deprecated. It is renamed to EventMessageVerifier.",
    if "bounce_dict" in kwargs:
        kwargs["notification"] = kwargs["bounce_dict"]
        del kwargs["bounce_dict"]
        "utils.verify_bounce_message is deprecated. It is renamed to verify_event_message.",
        "Received subscription confirmation: TopicArn: %s",
        notification.get("TopicArn"),
            "notification": notification,
    subscribe_url = notification.get("SubscribeURL")
            'Could not confirm subscription: "%s"',
            e,
                "notification": notification,