feat: SES add minimal address validation (#4769)

moto/ses/models.py

@@ -24,7 +24,7 @@ from .exceptions import (
     RuleAlreadyExists,
     MissingRenderingAttributeException,
 )
-from .utils import get_random_message_id
+from .utils import get_random_message_id, is_valid_address
 from .feedback import COMMON_MAIL, BOUNCE, COMPLAINT, DELIVERY
 
 RECIPIENT_LIMIT = 50
@@ -160,6 +160,13 @@ class SESBackend(BaseBackend):
         if not self._is_verified_address(source):
             self.rejected_messages_count += 1
             raise MessageRejectedError("Email address not verified %s" % source)
+        destination_addresses = [
+            address for addresses in destinations.values() for address in addresses
+        ]
+        for address in [source, *destination_addresses]:
+            valid, msg = is_valid_address(address)
+            if not valid:
+                raise InvalidParameterValue(msg)
 
         self.__process_sns_feedback__(source, destinations, region)
 
@@ -178,6 +185,13 @@ class SESBackend(BaseBackend):
         if not self._is_verified_address(source):
             self.rejected_messages_count += 1
             raise MessageRejectedError("Email address not verified %s" % source)
+        destination_addresses = [
+            address for addresses in destinations.values() for address in addresses
+        ]
+        for address in [source, *destination_addresses]:
+            valid, msg = is_valid_address(address)
+            if not valid:
+                raise InvalidParameterValue(msg)
 
         if not self.templates.get(template[0]):
             raise TemplateDoesNotExist("Template (%s) does not exist" % template[0])
@@ -259,6 +273,10 @@ class SESBackend(BaseBackend):
             )
         if recipient_count > RECIPIENT_LIMIT:
             raise MessageRejectedError("Too many recipients.")
+        for address in [addr for addr in [source, *destinations] if addr is not None]:
+            valid, msg = is_valid_address(address)
+            if not valid:
+                raise InvalidParameterValue(msg)
 
         self.__process_sns_feedback__(source, destinations, region)
 

moto/ses/utils.py

@@ -1,5 +1,6 @@
 import random
 import string
+from email.utils import parseaddr
 
 
 def random_hex(length):
@@ -16,3 +17,11 @@ def get_random_message_id():
         random_hex(12),
         random_hex(6),
     )
+
+
+def is_valid_address(addr):
+    _, address = parseaddr(addr)
+    address = address.split("@")
+    if len(address) != 2 or not address[1]:
+        return False, "Missing domain"
+    return True, None

tests/test_ses/test_ses_boto3.py

@@ -139,6 +139,29 @@ def test_send_unverified_email_with_chevrons():
     )
 
 
+@mock_ses
+def test_send_email_invalid_address():
+    conn = boto3.client("ses", region_name="us-east-1")
+    conn.verify_domain_identity(Domain="example.com")
+
+    with pytest.raises(ClientError) as ex:
+        conn.send_email(
+            Source="test@example.com",
+            Destination={
+                "ToAddresses": ["test_to@example.com", "invalid_address"],
+                "CcAddresses": [],
+                "BccAddresses": [],
+            },
+            Message={
+                "Subject": {"Data": "test subject"},
+                "Body": {"Text": {"Data": "test body"}},
+            },
+        )
+    err = ex.value.response["Error"]
+    err["Code"].should.equal("InvalidParameterValue")
+    err["Message"].should.equal("Missing domain")
+
+
 @mock_ses
 def test_send_templated_email():
     conn = boto3.client("ses", region_name="us-east-1")
@@ -184,6 +207,35 @@ def test_send_templated_email():
     sent_count.should.equal(3)
 
 
+@mock_ses
+def test_send_templated_email_invalid_address():
+    conn = boto3.client("ses", region_name="us-east-1")
+    conn.verify_domain_identity(Domain="example.com")
+    conn.create_template(
+        Template={
+            "TemplateName": "test_template",
+            "SubjectPart": "lalala",
+            "HtmlPart": "",
+            "TextPart": "",
+        }
+    )
+
+    with pytest.raises(ClientError) as ex:
+        conn.send_templated_email(
+            Source="test@example.com",
+            Destination={
+                "ToAddresses": ["test_to@example.com", "invalid_address"],
+                "CcAddresses": [],
+                "BccAddresses": [],
+            },
+            Template="test_template",
+            TemplateData='{"name": "test"}',
+        )
+    err = ex.value.response["Error"]
+    err["Code"].should.equal("InvalidParameterValue")
+    err["Message"].should.equal("Missing domain")
+
+
 @mock_ses
 def test_send_html_email():
     conn = boto3.client("ses", region_name="us-east-1")
@@ -243,6 +295,25 @@ def test_send_raw_email_validate_domain():
     sent_count.should.equal(2)
 
 
+@mock_ses
+def test_send_raw_email_invalid_address():
+    conn = boto3.client("ses", region_name="us-east-1")
+    conn.verify_domain_identity(Domain="example.com")
+
+    message = get_raw_email()
+    del message["To"]
+
+    with pytest.raises(ClientError) as ex:
+        conn.send_raw_email(
+            Source=message["From"],
+            Destinations=["test_to@example.com", "invalid_address"],
+            RawMessage={"Data": message.as_string()},
+        )
+    err = ex.value.response["Error"]
+    err["Code"].should.equal("InvalidParameterValue")
+    err["Message"].should.equal("Missing domain")
+
+
 def get_raw_email():
     message = MIMEMultipart()
     message["Subject"] = "Test"

tests/test_ses/test_ses_utils.py

@@ -0,0 +1,17 @@
+import sure  # noqa # pylint: disable=unused-import
+
+from moto.ses.utils import is_valid_address
+
+
+def test_is_valid_address():
+    valid, msg = is_valid_address("test@example.com")
+    valid.should.be.ok
+    msg.should.be.none
+
+    valid, msg = is_valid_address("test@")
+    valid.should_not.be.ok
+    msg.should.be.a(str)
+
+    valid, msg = is_valid_address("test")
+    valid.should_not.be.ok
+    msg.should.be.a(str)