c3lf/c3lf-system-3
7
0
Fork 0
Code Issues 51 Pull requests Projects 2 Releases Wiki Activity Actions

convert Quoted Printable email subject and body to utf-8

Browse source
This commit is contained in:
j3d1 2023-12-29 21:50:58 +01:00
parent 07213bd421
commit b4687a8821
3 changed files with 139 additions and 16 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

32
core/mail/migrations/0002_printed_quotable.py Normal file
View file

@ -0,0 +1,32 @@
# Generated by Django 4.2.7 on 2023-12-09 02:13
import quopri
from django.db import migrations
from mail.protocol import unescape_and_decode_quoted_printable, unescape_and_decode_base64
class Migration(migrations.Migration):
initial = True
dependencies = [
('mail', '0001_initial'),
]
def convert_printed_quotable(apps, schema_editor):
Email = apps.get_model('mail', 'Email')
for mail in Email.objects.all():
mail.body = unescape_and_decode_quoted_printable(mail.body)
mail.body = unescape_and_decode_base64(mail.body)
mail.subject = unescape_and_decode_quoted_printable(mail.subject)
mail.subject = unescape_and_decode_base64(mail.subject)
mail.save()
IssueThread = apps.get_model('tickets', 'IssueThread')
for issue in IssueThread.objects.all():
issue.name = unescape_and_decode_quoted_printable(issue.name)
issue.name = unescape_and_decode_base64(issue.name)
issue.save()
operations = [
migrations.RunPython(convert_printed_quotable),
]

60
core/mail/protocol.py
View file

@ -1,4 +1,5 @@
import logging
import aiosmtplib
from asgiref.sync import sync_to_async
from channels.layers import get_channel_layer
@ -8,6 +9,39 @@ from notify_sessions.models import SystemEvent
from tickets.models import IssueThread, StateChange
def find_quoted_printable(s, marker):
positions = [i for i in range(len(s)) if s.lower().startswith('=?utf-8?' + marker + '?', i)]
for pos in positions:
end = s.find('?=', pos + 10)
if end == -1:
continue
yield pos, end + 3
def unescape_and_decode_quoted_printable(s):
import quopri
decoded = ''
last_end = 0
for start, end in find_quoted_printable(s, 'q'):
decoded += s[last_end:start]
decoded += quopri.decodestring(s[start + 10:end - 3]).decode('utf-8')
last_end = end
decoded += s[last_end:]
return decoded
def unescape_and_decode_base64(s):
import base64
decoded = ''
last_end = 0
for start, end in find_quoted_printable(s, 'b'):
decoded += s[last_end:start]
decoded += base64.b64decode(s[start + 10:end - 3]).decode('utf-8')
last_end = end
decoded += s[last_end:]
return decoded
def collect_references(issue_thread):
mails = issue_thread.emails.order_by('timestamp')
references = []
@ -105,6 +139,8 @@ class LMTPHandler:
recipient = envelope.rcpt_tos[0]
sender = envelope.mail_from
subject = parsed.get('Subject')
subject = unescape_and_decode_quoted_printable(subject)
subject = unescape_and_decode_base64(subject)
target_event = None
try:
address_map = await sync_to_async(EventAddress.objects.get)(address=recipient)
@ -114,15 +150,13 @@ class LMTPHandler:
pass
active_issue_thread, new = await sync_to_async(find_active_issue_thread)(header_in_reply_to, subject)
body_decoded = body.decode('utf-8')
body_decoded = unescape_and_decode_quoted_printable(body_decoded)
body_decoded = unescape_and_decode_base64(body_decoded)
email = await sync_to_async(Email.objects.create)(sender=sender,
recipient=recipient,
body=body.decode('utf-8'),
subject=subject,
reference=header_message_id,
in_reply_to=header_in_reply_to,
raw=envelope.content.decode('utf-8'),
event=target_event,
email = await sync_to_async(Email.objects.create)(
sender=sender, recipient=recipient, body=body_decoded, 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)
log.info(f"Created email {email.id}")
systemevent = await sync_to_async(SystemEvent.objects.create)(type='email received', reference=email.id)
@ -138,13 +172,9 @@ class LMTPHandler:
references = await sync_to_async(collect_references)(active_issue_thread)
reply_email = await sync_to_async(Email.objects.create)(sender=recipient, # "noreply@" + MAIL_DOMAIN,
recipient=sender,
body="Thank you for your message.",
subject="Message received",
in_reply_to=header_message_id,
event=target_event,
issue_thread=active_issue_thread)
reply_email = await sync_to_async(Email.objects.create)(
sender=recipient, recipient=sender, body="Thank you for your message.", subject="Message received",
in_reply_to=header_message_id, event=target_event, issue_thread=active_issue_thread)
await send_smtp(make_reply(reply_email, references), log)
log.info("Sent auto reply")

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

@ -105,6 +105,67 @@ class LMTPHandlerTestCase(TestCase): # TODO replace with less hacky test
self.assertEqual('pending_new', IssueThread.objects.all()[0].state)
self.assertEqual(None, IssueThread.objects.all()[0].assigned_to)
def test_handle_quoted_printable(self):
from aiosmtpd.smtp import Envelope
from asgiref.sync import async_to_sync
import aiosmtplib
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 =?utf-8?Q?=C3=A4?=\nFrom: test3@test\nTo: test4@test\nMessage-ID: <1@test>\n\nText mit Quoted-Printable-Kodierung: =?utf-8?Q?=C3=A4=C3=B6=C3=BC=C3=9F?='
result = async_to_sync(handler.handle_DATA)(server, session, envelope)
self.assertEqual(result, '250 Message accepted for delivery')
self.assertEqual(len(Email.objects.all()), 2)
self.assertEqual(len(IssueThread.objects.all()), 1)
aiosmtplib.send.assert_called_once()
self.assertEqual('test ä', Email.objects.all()[0].subject)
self.assertEqual('Text mit Quoted-Printable-Kodierung: äöüß', Email.objects.all()[0].body)
def test_handle_quoted_printable_2(self):
#=?UTF-8?Q?suche_M=C3=BCtze?=
from aiosmtpd.smtp import Envelope
from asgiref.sync import async_to_sync
import aiosmtplib
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: =?UTF-8?Q?suche_M=C3=BCtze?=\nFrom: test3@test\nTo: test4@test\nMessage-ID: <1@test>\n\nText mit Quoted-Printable-Kodierung: =?utf-8?Q?=C3=A4=C3=B6=C3=BC=C3=9F?='
result = async_to_sync(handler.handle_DATA)(server, session, envelope)
self.assertEqual(result, '250 Message accepted for delivery')
self.assertEqual(len(Email.objects.all()), 2)
self.assertEqual(len(IssueThread.objects.all()), 1)
aiosmtplib.send.assert_called_once()
self.assertEqual('suche_Mütze', Email.objects.all()[0].subject)
self.assertEqual('Text mit Quoted-Printable-Kodierung: äöüß', Email.objects.all()[0].body)
def test_handle_base64(self):
from aiosmtpd.smtp import Envelope
from asgiref.sync import async_to_sync
import aiosmtplib
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: =?utf-8?B?dGVzdA==?=\nFrom: test3@test\nTo: test4@test\nMessage-ID: <1@test>\n\nText mit Base64-Kodierung: =?utf-8?B?w6TDtsO8w58=?='
result = async_to_sync(handler.handle_DATA)(server, session, envelope)
self.assertEqual(result, '250 Message accepted for delivery')
self.assertEqual(len(Email.objects.all()), 2)
self.assertEqual(len(IssueThread.objects.all()), 1)
aiosmtplib.send.assert_called_once()
self.assertEqual('test', Email.objects.all()[0].subject)
self.assertEqual('Text mit Base64-Kodierung: äöüß', Email.objects.all()[0].body)
def test_handle_client_reply(self):
issue_thread = IssueThread.objects.create(
name="test",