Mailgun/SendGrid inbound: workaround Django filename issue

Workaround for Django multipart/form-data limitation
where certain attachment filenames cause fields to be dropped
or to end up in request.POST rather than request.FILES.

Handle the MultiValueDictKeyError in inbound webhooks when
this has occurred. Also update docs to recommend avoiding
the problem by using Mailgun and SendGrid's "raw MIME" options.

Also handle reported cases of empty, duplicate keys in Mailgun's
content-id-map.

Fixes #272
This commit is contained in:
medmunds
2022-05-10 11:47:57 -07:00
committed by Mike Edmunds
parent 6a2e30ba8f
commit 09f21a5c2d
9 changed files with 244 additions and 52 deletions

View File

@@ -15,7 +15,7 @@ from anymail.webhooks.mailgun import MailgunInboundWebhookView
from .test_mailgun_webhooks import (
TEST_WEBHOOK_SIGNING_KEY, mailgun_sign_payload,
mailgun_sign_legacy_payload, querydict_to_postdict)
from .utils import sample_image_content, sample_email_content
from .utils import sample_image_content, sample_email_content, encode_multipart, make_fileobj
from .webhook_cases import WebhookTestCase
@@ -104,6 +104,7 @@ class MailgunInboundTestCase(WebhookTestCase):
att2.name = 'image.png'
email_content = sample_email_content()
att3 = BytesIO(email_content)
att3.name = '\\share\\mail\\forwarded.msg'
att3.content_type = 'message/rfc822; charset="us-ascii"'
raw_event = mailgun_sign_legacy_payload({
'message-headers': '[]',
@@ -125,6 +126,7 @@ class MailgunInboundTestCase(WebhookTestCase):
self.assertEqual(attachments[0].get_filename(), 'test.txt')
self.assertEqual(attachments[0].get_content_type(), 'text/plain')
self.assertEqual(attachments[0].get_content_text(), 'test attachment')
self.assertEqual(attachments[1].get_filename(), 'forwarded.msg') # Django strips paths
self.assertEqual(attachments[1].get_content_type(), 'message/rfc822')
self.assertEqualIgnoringHeaderFolding(attachments[1].get_content_bytes(), email_content)
@@ -135,6 +137,68 @@ class MailgunInboundTestCase(WebhookTestCase):
self.assertEqual(inline.get_content_type(), 'image/png')
self.assertEqual(inline.get_content_bytes(), image_content)
def test_filtered_attachment_filenames(self):
# Make sure the inbound webhook can deal with missing fields caused by
# Django's multipart/form-data filename filtering. (The attachments are lost,
# but shouldn't cause errors in the inbound webhook.)
filenames = [
"", "path\\", "path/"
".", "path\\.", "path/.",
"..", "path\\..", "path/..",
]
num_attachments = len(filenames)
payload = {
"attachment-%d" % (i+1): make_fileobj("content", filename=filenames[i], content_type="text/pdf")
for i in range(num_attachments)
}
payload.update({
'message-headers': '[]',
'attachment-count': str(num_attachments),
})
# Must do our own multipart/form-data encoding for empty filenames:
response = self.client.post('/anymail/mailgun/inbound/',
data=encode_multipart("BoUnDaRy", mailgun_sign_legacy_payload(payload)),
content_type="multipart/form-data; boundary=BoUnDaRy")
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(self.inbound_handler, sender=MailgunInboundWebhookView,
event=ANY, esp_name='Mailgun')
# Different Django releases strip different filename patterns.
# Just verify that at least some attachments got dropped (so the test is valid)
# without causing an error in the inbound webhook:
attachments = kwargs['event'].message.attachments
self.assertLess(len(attachments), num_attachments)
def test_unusual_content_id_map(self):
# Under unknown conditions, Mailgun appears to generate a content-id-map with multiple
# empty keys (and possibly other duplicate keys). We still want to correctly identify
# inline attachments from it.
raw_event = mailgun_sign_legacy_payload({
'message-headers': '[]',
'attachment-count': '4',
'content-id-map': '{"": "attachment-1", "": "attachment-2",'
' "<abc>": "attachment-3", "<abc>": "attachment-4"}',
'attachment-1': make_fileobj("att1"),
'attachment-2': make_fileobj("att2"),
'attachment-3': make_fileobj("att3"),
'attachment-4': make_fileobj("att4"),
})
response = self.client.post('/anymail/mailgun/inbound/', data=raw_event)
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(self.inbound_handler, sender=MailgunInboundWebhookView,
event=ANY, esp_name='Mailgun')
event = kwargs['event']
message = event.message
self.assertEqual(len(message.attachments), 0) # all inlines
inlines = [part for part in message.walk() if part.is_inline_attachment()]
self.assertEqual(len(inlines), 4)
self.assertEqual(inlines[0]["Content-ID"], "")
self.assertEqual(inlines[1]["Content-ID"], "")
self.assertEqual(inlines[2]["Content-ID"], "<abc>")
self.assertEqual(inlines[3]["Content-ID"], "<abc>")
def test_inbound_mime(self):
# Mailgun provides the full, raw MIME message if the webhook url ends in 'mime'
raw_event = mailgun_sign_legacy_payload({

View File

@@ -9,7 +9,7 @@ from anymail.inbound import AnymailInboundMessage
from anymail.signals import AnymailInboundEvent
from anymail.webhooks.sendgrid import SendGridInboundWebhookView
from .utils import dedent_bytes, sample_image_content, sample_email_content
from .utils import dedent_bytes, sample_image_content, sample_email_content, encode_multipart, make_fileobj
from .webhook_cases import WebhookTestCase
@@ -96,15 +96,16 @@ class SendgridInboundTestCase(WebhookTestCase):
att2.name = 'image.png'
email_content = sample_email_content()
att3 = BytesIO(email_content)
att3.name = '\\share\\mail\\forwarded.msg'
att3.content_type = 'message/rfc822; charset="us-ascii"'
raw_event = {
'headers': '',
'attachments': '3',
'attachment-info': json.dumps({
"attachment3": {"filename": "", "name": "", "charset": "US-ASCII", "type": "message/rfc822"},
"attachment2": {"filename": "image.png", "name": "image.png", "type": "image/png",
"content-id": "abc123"},
"attachment1": {"filename": "test.txt", "name": "test.txt", "type": "text/plain"},
"attachment3": {"filename": "\\share\\mail\\forwarded.msg",
"charset": "US-ASCII", "type": "message/rfc822"},
"attachment2": {"filename": "image.png", "type": "image/png", "content-id": "abc123"},
"attachment1": {"filename": "test.txt", "charset": "UTF-8", "type": "text/plain"},
}),
'content-ids': '{"abc123": "attachment2"}',
'attachment1': att1,
@@ -123,6 +124,7 @@ class SendgridInboundTestCase(WebhookTestCase):
self.assertEqual(attachments[0].get_filename(), 'test.txt')
self.assertEqual(attachments[0].get_content_type(), 'text/plain')
self.assertEqual(attachments[0].get_content_text(), 'test attachment')
self.assertEqual(attachments[1].get_filename(), 'forwarded.msg') # Django strips path
self.assertEqual(attachments[1].get_content_type(), 'message/rfc822')
self.assertEqualIgnoringHeaderFolding(attachments[1].get_content_bytes(), email_content)
@@ -133,6 +135,45 @@ class SendgridInboundTestCase(WebhookTestCase):
self.assertEqual(inline.get_content_type(), 'image/png')
self.assertEqual(inline.get_content_bytes(), image_content)
def test_filtered_attachment_filenames(self):
# Make sure the inbound webhook can deal with missing fields caused by
# Django's multipart/form-data filename filtering. (The attachments are lost,
# but shouldn't cause errors in the inbound webhook.)
filenames = [
"", "path\\", "path/"
".", "path\\.", "path/.",
"..", "path\\..", "path/..",
]
num_attachments = len(filenames)
payload = {
"attachment%d" % (i+1): make_fileobj("content", filename=filenames[i], content_type="text/pdf")
for i in range(num_attachments)
}
attachment_info = {
key: {"filename": value.name, "type": "text/pdf"}
for key, value in payload.items()
}
payload.update({
'headers': '',
'attachments': str(num_attachments),
'attachment-info': json.dumps(attachment_info),
})
# Must do our own form-data encoding to properly test empty attachment filenames.
# Must do our own multipart/form-data encoding for empty filenames:
response = self.client.post('/anymail/sendgrid/inbound/',
data=encode_multipart("BoUnDaRy", payload),
content_type="multipart/form-data; boundary=BoUnDaRy")
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(self.inbound_handler, sender=SendGridInboundWebhookView,
event=ANY, esp_name='SendGrid')
# Different Django releases strip different filename patterns.
# Just verify that at least some attachments got dropped (so the test is valid)
# without causing an error in the inbound webhook:
attachments = kwargs['event'].message.attachments
self.assertLess(len(attachments), num_attachments)
def test_inbound_mime(self):
# SendGrid has an option to send the full, raw MIME message
raw_event = {

View File

@@ -5,12 +5,12 @@ import uuid
import warnings
from base64 import b64decode
from contextlib import contextmanager
from io import StringIO
from io import BytesIO, StringIO
from pathlib import Path
from unittest import TestCase
from unittest.util import safe_repr
from django.test import Client
import django.test.client
def decode_att(att):
@@ -177,7 +177,7 @@ class AnymailTestMixin(TestCase):
sys.stdout = old_stdout
class ClientWithCsrfChecks(Client):
class ClientWithCsrfChecks(django.test.Client):
"""Django test Client that enforces CSRF checks
https://docs.djangoproject.com/en/stable/ref/csrf/#testing
@@ -227,3 +227,35 @@ def dedent_bytes(text):
if margin:
text = re.sub(b'(?m)^' + margin, b'', text)
return text
def make_fileobj(content, filename=None, content_type=None, encoding=None):
"""
Returns a file-like object that can be used in Django test Client
post data to simulate an uploaded file.
"""
# The logic that unpacks this is in django.test.client.encode_file.
if isinstance(content, str):
content = content.encode(encoding or 'utf-8')
fileobj = BytesIO(content)
if filename is not None:
fileobj.name = filename
if content_type is not None:
fileobj.content_type = content_type
return fileobj
def encode_multipart(boundary, data):
"""
Version of :func:`django.test.client.encode_multipart` that allows
empty filenames. (The original function substitutes the field's
name if a file has an empty name.)
"""
# For simplicity, encode with the original function, and then
# replace any 'filename="<key>"' with 'filename=""'. This isn't
# entirely robust, but is sufficient for testing use.
encoded = django.test.client.encode_multipart(boundary, data)
re_keys = r"|".join(re.escape(key) for key in data.keys())
return re.sub(
rb'filename="(%s)"' % re_keys.encode("ascii"),
b'filename=""', encoded)