c3lf/c3lf-system-3
8
0
Fork 0
Code Issues 26 Pull requests Releases Wiki Activity

handle empty Subject and empty body in incoming mails

Browse source
This commit is contained in:
jedi 2024-01-12 21:19:14 +01:00
parent c79b3185e5
commit 4664d6255d
2 changed files with 93 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
core/mail/protocol.py
View file

@ -117,6 +117,8 @@ def parse_email_body(raw, log=None):
# skip any text/plain (txt) attachments
if ctype == 'text/plain' and 'attachment' not in cdispo:
segment = part.get_payload(decode=True).decode('utf-8')
if not segment:
continue
segment = unescape_and_decode_quoted_printable(segment)
segment = unescape_and_decode_base64(segment)
log.debug(segment)
@ -138,6 +140,9 @@ def parse_email_body(raw, log=None):
log.info("Attachment", ctype, cdispo)
else:
body = parsed.get_payload(decode=True).decode('utf-8')
body = unescape_and_decode_quoted_printable(body)
body = unescape_and_decode_base64(body)
log.debug(body)
return parsed, body, attachments
@ -161,17 +166,17 @@ def receive_email(envelope, log=None):
recipient = envelope.rcpt_tos[0].lower() if envelope.rcpt_tos else header_to.lower()
sender = envelope.mail_from if envelope.mail_from else header_from
subject = parsed.get('Subject')
if not subject:
subject = "No subject"
subject = unescape_and_decode_quoted_printable(subject)
subject = unescape_and_decode_base64(subject)
target_event = find_target_event(recipient)
active_issue_thread, new = find_active_issue_thread(header_in_reply_to, subject)
body_decoded = body
body_decoded = unescape_and_decode_quoted_printable(body_decoded)
body_decoded = unescape_and_decode_base64(body_decoded)
email = Email.objects.create(
sender=sender, recipient=recipient, body=body_decoded, subject=subject, reference=header_message_id,
sender=sender, recipient=recipient, body=body, subject=subject, reference=header_message_id,
in_reply_to=header_in_reply_to, raw=envelope.content.decode('utf-8'), event=target_event,
issue_thread=active_issue_thread)
for attachment in attachments:

84
core/mail/tests/v2/test_mails.py
View file

@ -521,3 +521,87 @@ dGVzdGltYWdl
states = StateChange.objects.filter(issue_thread=IssueThread.objects.all()[0])
self.assertEqual(1, len(states))
self.assertEqual('pending_new', states[0].state)
def test_mail_empty_subject(self):
from aiosmtpd.smtp import Envelope
from asgiref.sync import async_to_sync
import aiosmtplib
import logging
logging.disable(logging.CRITICAL)
aiosmtplib.send = make_mocked_coro()
handler = LMTPHandler()
server = mock.Mock()
session = mock.Mock()
envelope = Envelope()
envelope.mail_from = 'test1@test'
envelope.rcpt_tos = ['test2@test']
envelope.content = b'From: noreply@test\nTo: test2@test\nMessage-ID: <1@test>\n\ntest'
result = async_to_sync(handler.handle_DATA)(server, session, envelope)
logging.disable(logging.NOTSET)
self.assertEqual('250 Message accepted for delivery', result)
self.assertEqual(2, len(Email.objects.all()))
self.assertEqual(1, len(IssueThread.objects.all()))
aiosmtplib.send.assert_called_once()
self.assertEqual('No subject', Email.objects.all()[0].subject)
self.assertEqual('test1@test', Email.objects.all()[0].sender)
self.assertEqual('test2@test', Email.objects.all()[0].recipient)
self.assertEqual('test', Email.objects.all()[0].body)
self.assertEqual(IssueThread.objects.all()[0], Email.objects.all()[0].issue_thread)
self.assertEqual('<1@test>', Email.objects.all()[0].reference)
self.assertEqual(None, Email.objects.all()[0].in_reply_to)
self.assertEqual('Message received', Email.objects.all()[1].subject)
self.assertEqual('test2@test', Email.objects.all()[1].sender)
self.assertEqual('test1@test', Email.objects.all()[1].recipient)
self.assertEqual('Thank you for your message.', Email.objects.all()[1].body)
self.assertEqual(IssueThread.objects.all()[0], Email.objects.all()[1].issue_thread)
self.assertTrue(Email.objects.all()[1].reference.startswith("<"))
self.assertTrue(Email.objects.all()[1].reference.endswith("@localhost>"))
self.assertEqual("<1@test>", Email.objects.all()[1].in_reply_to)
self.assertEqual('No subject', IssueThread.objects.all()[0].name)
self.assertEqual('pending_new', IssueThread.objects.all()[0].state)
self.assertEqual(None, IssueThread.objects.all()[0].assigned_to)
states = StateChange.objects.filter(issue_thread=IssueThread.objects.all()[0])
self.assertEqual(1, len(states))
self.assertEqual('pending_new', states[0].state)
def test_mail_empty_body(self):
from aiosmtpd.smtp import Envelope
from asgiref.sync import async_to_sync
import aiosmtplib
import logging
logging.disable(logging.CRITICAL)
aiosmtplib.send = make_mocked_coro()
handler = LMTPHandler()
server = mock.Mock()
session = mock.Mock()
envelope = Envelope()
envelope.mail_from = '<test1@test>'
envelope.rcpt_tos = ['test2@test']
envelope.content = b'Subject: test\nFrom: <test1@test>\nTo: test2@test\nMessage-ID: <1@test>\n\n'
result = async_to_sync(handler.handle_DATA)(server, session, envelope)
logging.disable(logging.NOTSET)
self.assertEqual('250 Message accepted for delivery', result)
self.assertEqual(2, len(Email.objects.all()))
self.assertEqual(1, len(IssueThread.objects.all()))
aiosmtplib.send.assert_called_once()
self.assertEqual('test', Email.objects.all()[0].subject)
self.assertEqual('<test1@test>', Email.objects.all()[0].sender)
self.assertEqual('test2@test', Email.objects.all()[0].recipient)
self.assertEqual('', Email.objects.all()[0].body)
self.assertEqual(IssueThread.objects.all()[0], Email.objects.all()[0].issue_thread)
self.assertEqual('<1@test>', Email.objects.all()[0].reference)
self.assertEqual(None, Email.objects.all()[0].in_reply_to)
self.assertEqual('Message received', Email.objects.all()[1].subject)
self.assertEqual('test2@test', Email.objects.all()[1].sender)
self.assertEqual('<test1@test>', Email.objects.all()[1].recipient)
self.assertEqual('Thank you for your message.', Email.objects.all()[1].body)
self.assertEqual(IssueThread.objects.all()[0], Email.objects.all()[1].issue_thread)
self.assertTrue(Email.objects.all()[1].reference.startswith("<"))
self.assertTrue(Email.objects.all()[1].reference.endswith("@localhost>"))
self.assertEqual("<1@test>", Email.objects.all()[1].in_reply_to)
self.assertEqual('test', IssueThread.objects.all()[0].name)
self.assertEqual('pending_new', IssueThread.objects.all()[0].state)
self.assertEqual(None, IssueThread.objects.all()[0].assigned_to)
states = StateChange.objects.filter(issue_thread=IssueThread.objects.all()[0])
self.assertEqual(1, len(states))
self.assertEqual('pending_new', states[0].state)