From c23c8e5be6d5be498cd7877b0e0e29072461c49d Mon Sep 17 00:00:00 2001 From: "bruce.gibbins" Date: Wed, 19 Apr 2023 20:07:43 +1000 Subject: [PATCH] Add user defined IMAP Debug Level and improved exception message --- helpdesk/email.py | 12 +- helpdesk/settings.py | 3 + helpdesk/tests/test_get_email.py | 1013 +++++++++++++++++------------- quicktest.py | 4 +- 4 files changed, 594 insertions(+), 438 deletions(-) diff --git a/helpdesk/email.py b/helpdesk/email.py index f9faf4ce..68aa06b6 100644 --- a/helpdesk/email.py +++ b/helpdesk/email.py @@ -251,8 +251,7 @@ def imap_oauth_sync(q, logger, server): include_client_id=True, ) - # TODO: Somehow link this to the debug level set within Django settings logging - server.debug = 4 + server.debug = settings.HELPDESK_IMAP_DEBUG_LEVEL # TODO: Perhaps store the authentication string template externally? Settings? Queue Table? server.authenticate( @@ -263,15 +262,14 @@ def imap_oauth_sync(q, logger, server): # Select the Inbound Mailbox folder server.select(q.email_box_imap_folder) - except imaplib.IMAP4.abort: - logger.error("IMAP authentication failed.") + except imaplib.IMAP4.abort as e1: + logger.error(f"IMAP authentication failed in OAUTH: {e1}", exc_info=True) server.logout() sys.exit() - except ssl.SSLError: + except ssl.SSLError as e2: logger.error( - "IMAP login failed due to SSL error. This is often due to a timeout. " - "Please check your connection and try again." + f"IMAP login failed due to SSL error. (This is often due to a timeout): {e2}", exc_info=True ) server.logout() sys.exit() diff --git a/helpdesk/settings.py b/helpdesk/settings.py index 15c6022e..7dc7ac5c 100644 --- a/helpdesk/settings.py +++ b/helpdesk/settings.py @@ -262,3 +262,6 @@ HELPDESK_OAUTH = getattr( "scope": [""] } ) + +# Set Debug Logging Level for IMAP Services. Default to '0' for No Debugging +HELPDESK_IMAP_DEBUG_LEVEL = getattr(settings, 'HELPDESK_IMAP_DEBUG_LEVEL', 0) diff --git a/helpdesk/tests/test_get_email.py b/helpdesk/tests/test_get_email.py index 217b6a82..f7d37800 100644 --- a/helpdesk/tests/test_get_email.py +++ b/helpdesk/tests/test_get_email.py @@ -420,436 +420,589 @@ class GetEmailParametricTemplate(object): self.assertEqual(ticket2.ticket_for_url, "QQ-%s" % ticket2.id) self.assertEqual(ticket2.title, test_email_subject) self.assertEqual(ticket2.description, test_email_body) -# def test_commas_in_mail_headers(self): -# """Tests correctly decoding mail headers when a comma is encoded into -# UTF-8. See bug report #832.""" -# -# # Create the from using standard RFC required formats -# # Override the last_name to ensure we get a non-ascii character in it -# test_email_from_meta = utils.generate_email_address("fr_FR", last_name_override="Bouissières") -# test_email_subject = "Commas in From lines" -# test_email_body = "Testing commas in from email UTF-8." -# test_email = "To: helpdesk@example.com\nFrom: " + test_email_from_meta[0] + \ -# "\nSubject: " + test_email_subject + "\n\n" + test_email_body -# test_mail_len = len(test_email) -# -# if self.socks: -# from socks import ProxyConnectionError -# with self.assertRaisesRegex(ProxyConnectionError, '%s:%s' % (unrouted_socks_server, unused_port)): -# call_command('get_email') -# -# else: -# # Test local email reading -# if self.method == 'local': -# with mock.patch('os.listdir') as mocked_listdir, \ -# mock.patch('helpdesk.email.isfile') as mocked_isfile, \ -# mock.patch('builtins.open', mock.mock_open(read_data=test_email)), \ -# mock.patch('os.unlink'): -# mocked_isfile.return_value = True -# mocked_listdir.return_value = ['filename1', 'filename2'] -# -# call_command('get_email') -# -# mocked_listdir.assert_called_with( -# '/var/lib/mail/helpdesk/') -# mocked_isfile.assert_any_call( -# '/var/lib/mail/helpdesk/filename1') -# mocked_isfile.assert_any_call( -# '/var/lib/mail/helpdesk/filename2') -# -# elif self.method == 'pop3': -# # mock poplib.POP3's list and retr methods to provide responses -# # as per RFC 1939 -# pop3_emails = { -# '1': ("+OK", test_email.split('\n')), -# '2': ("+OK", test_email.split('\n')), -# } -# pop3_mail_list = ("+OK 2 messages", ("1 %d" % -# test_mail_len, "2 %d" % test_mail_len)) -# mocked_poplib_server = mock.Mock() -# mocked_poplib_server.list = mock.Mock( -# return_value=pop3_mail_list) -# mocked_poplib_server.retr = mock.Mock( -# side_effect=lambda x: pop3_emails[x]) -# with mock.patch('helpdesk.email.poplib', autospec=True) as mocked_poplib: -# mocked_poplib.POP3 = mock.Mock( -# return_value=mocked_poplib_server) -# call_command('get_email') -# -# elif self.method in ['imap', 'oauth']: -# # mock imaplib.IMAP4's search and fetch methods with responses -# # from RFC 3501 -# imap_emails = { -# "1": ("OK", (("1", test_email),)), -# "2": ("OK", (("2", test_email),)), -# } -# imap_mail_list = ("OK", ("1 2",)) -# mocked_imaplib_server = mock.Mock() -# mocked_imaplib_server.search = mock.Mock( -# return_value=imap_mail_list) -# -# # we ignore the second arg as the data item/mime-part is -# # constant (RFC822) -# mocked_imaplib_server.fetch = mock.Mock( -# side_effect=lambda x, _: imap_emails[x]) -# with mock.patch('helpdesk.email.imaplib', autospec=True) as mocked_imaplib: -# mocked_imaplib.IMAP4 = mock.Mock( -# return_value=mocked_imaplib_server) -# call_command('get_email') -# -# ticket1 = get_object_or_404(Ticket, pk=1) -# self.assertEqual(ticket1.ticket_for_url, "QQ-%s" % ticket1.id) -# self.assertEqual(ticket1.submitter_email, test_email_from_meta[1]) -# self.assertEqual(ticket1.title, test_email_subject) -# self.assertEqual(ticket1.description, test_email_body) -# -# ticket2 = get_object_or_404(Ticket, pk=2) -# self.assertEqual(ticket2.ticket_for_url, "QQ-%s" % ticket2.id) -# self.assertEqual(ticket2.submitter_email, test_email_from_meta[1]) -# self.assertEqual(ticket2.title, test_email_subject) -# self.assertEqual(ticket2.description, test_email_body) -# -# def test_read_email_with_template_tag(self): -# """Tests reading plain text emails from a queue and creating tickets, -# except this time the email body contains a Django template tag. -# For each email source supported, we mock the backend to provide -# authentically formatted responses containing our test data.""" -# -# # example email text from Django docs: -# # https://docs.djangoproject.com/en/1.10/ref/unicode/ -# test_email_from = "Arnbjörg Ráðormsdóttir " -# test_email_subject = "My visit to Sør-Trøndelag" -# test_email_body = "Reporting some issue with the template tag: {% if helpdesk %}." -# test_email = "To: helpdesk@example.com\nFrom: " + test_email_from + \ -# "\nSubject: " + test_email_subject + "\n\n" + test_email_body -# test_mail_len = len(test_email) -# -# if self.socks: -# from socks import ProxyConnectionError -# with self.assertRaisesRegex(ProxyConnectionError, '%s:%s' % (unrouted_socks_server, unused_port)): -# call_command('get_email') -# -# else: -# # Test local email reading -# if self.method == 'local': -# with mock.patch('os.listdir') as mocked_listdir, \ -# mock.patch('helpdesk.email.isfile') as mocked_isfile, \ -# mock.patch('builtins.open', mock.mock_open(read_data=test_email)), \ -# mock.patch('os.unlink'): -# mocked_isfile.return_value = True -# mocked_listdir.return_value = ['filename1', 'filename2'] -# -# call_command('get_email') -# -# mocked_listdir.assert_called_with( -# '/var/lib/mail/helpdesk/') -# mocked_isfile.assert_any_call( -# '/var/lib/mail/helpdesk/filename1') -# mocked_isfile.assert_any_call( -# '/var/lib/mail/helpdesk/filename2') -# -# elif self.method == 'pop3': -# # mock poplib.POP3's list and retr methods to provide responses -# # as per RFC 1939 -# pop3_emails = { -# '1': ("+OK", test_email.split('\n')), -# '2': ("+OK", test_email.split('\n')), -# } -# pop3_mail_list = ("+OK 2 messages", ("1 %d" % -# test_mail_len, "2 %d" % test_mail_len)) -# mocked_poplib_server = mock.Mock() -# mocked_poplib_server.list = mock.Mock( -# return_value=pop3_mail_list) -# mocked_poplib_server.retr = mock.Mock( -# side_effect=lambda x: pop3_emails[x]) -# with mock.patch('helpdesk.email.poplib', autospec=True) as mocked_poplib: -# mocked_poplib.POP3 = mock.Mock( -# return_value=mocked_poplib_server) -# call_command('get_email') -# -# elif self.method in ['imap', 'oauth']: -# # mock imaplib.IMAP4's search and fetch methods with responses -# # from RFC 3501 -# imap_emails = { -# "1": ("OK", (("1", test_email),)), -# "2": ("OK", (("2", test_email),)), -# } -# imap_mail_list = ("OK", ("1 2",)) -# mocked_imaplib_server = mock.Mock() -# mocked_imaplib_server.search = mock.Mock( -# return_value=imap_mail_list) -# -# # we ignore the second arg as the data item/mime-part is -# # constant (RFC822) -# mocked_imaplib_server.fetch = mock.Mock( -# side_effect=lambda x, _: imap_emails[x]) -# with mock.patch('helpdesk.email.imaplib', autospec=True) as mocked_imaplib: -# mocked_imaplib.IMAP4 = mock.Mock( -# return_value=mocked_imaplib_server) -# call_command('get_email') -# -# ticket1 = get_object_or_404(Ticket, pk=1) -# self.assertEqual(ticket1.ticket_for_url, "QQ-%s" % ticket1.id) -# self.assertEqual(ticket1.title, test_email_subject) -# self.assertEqual(ticket1.description, test_email_body) -# -# ticket2 = get_object_or_404(Ticket, pk=2) -# self.assertEqual(ticket2.ticket_for_url, "QQ-%s" % ticket2.id) -# self.assertEqual(ticket2.title, test_email_subject) -# self.assertEqual(ticket2.description, test_email_body) -# -# def test_read_html_multipart_email(self): -# """Tests reading multipart MIME (HTML body and plain text alternative) -# emails from a queue and creating tickets. -# For each email source supported, we mock the backend to provide -# authentically formatted responses containing our test data.""" -# -# # example email text from Python docs: -# # https://docs.python.org/3/library/email-examples.html -# from email.mime.multipart import MIMEMultipart -# from email.mime.text import MIMEText -# -# me = "my@example.com" -# you = "your@example.com" -# # NOTE: CC'd emails need to be alphabetical and tested as such! -# # implementation uses sets, so only way to ensure tickets created -# # in right order is to change set to list and sort it -# cc_one = "nobody@example.com" -# cc_two = "other@example.com" -# cc = cc_one + ", " + cc_two -# subject = "Link" -# -# # Create message container - the correct MIME type is -# # multipart/alternative. -# msg = MIMEMultipart('alternative') -# msg['Subject'] = subject -# msg['From'] = me -# msg['To'] = you -# msg['Cc'] = cc -# -# # Create the body of the message (a plain-text and an HTML version). -# text = "Hi!\nHow are you?\nHere is the link you wanted:\nhttps://www.python.org" -# html = """\ -# -# -# -#

Hi!
-# How are you?
-# Here is the link you wanted. -#

-# -# -# """ -# -# # Record the MIME types of both parts - text/plain and text/html. -# part1 = MIMEText(text, 'plain') -# part2 = MIMEText(html, 'html') -# -# # Attach parts into message container. -# # According to RFC 2046, the last part of a multipart message, in this case -# # the HTML message, is best and preferred. -# msg.attach(part1) -# msg.attach(part2) -# -# test_mail_len = len(msg) -# -# if self.socks: -# from socks import ProxyConnectionError -# with self.assertRaisesRegex(ProxyConnectionError, '%s:%s' % (unrouted_socks_server, unused_port)): -# call_command('get_email') -# -# else: -# # Test local email reading -# if self.method == 'local': -# with mock.patch('os.listdir') as mocked_listdir, \ -# mock.patch('helpdesk.email.isfile') as mocked_isfile, \ -# mock.patch('builtins.open', mock.mock_open(read_data=msg.as_string())), \ -# mock.patch('os.unlink'): -# mocked_isfile.return_value = True -# mocked_listdir.return_value = ['filename1', 'filename2'] -# -# call_command('get_email') -# -# mocked_listdir.assert_called_with( -# '/var/lib/mail/helpdesk/') -# mocked_isfile.assert_any_call( -# '/var/lib/mail/helpdesk/filename1') -# mocked_isfile.assert_any_call( -# '/var/lib/mail/helpdesk/filename2') -# -# elif self.method == 'pop3': -# # mock poplib.POP3's list and retr methods to provide responses -# # as per RFC 1939 -# pop3_emails = { -# '1': ("+OK", msg.as_string().split('\n')), -# '2': ("+OK", msg.as_string().split('\n')), -# } -# pop3_mail_list = ("+OK 2 messages", ("1 %d" % -# test_mail_len, "2 %d" % test_mail_len)) -# mocked_poplib_server = mock.Mock() -# mocked_poplib_server.list = mock.Mock( -# return_value=pop3_mail_list) -# mocked_poplib_server.retr = mock.Mock( -# side_effect=lambda x: pop3_emails[x]) -# with mock.patch('helpdesk.email.poplib', autospec=True) as mocked_poplib: -# mocked_poplib.POP3 = mock.Mock( -# return_value=mocked_poplib_server) -# call_command('get_email') -# -# -# elif self.method in ['imap', 'oauth']: -# # mock imaplib.IMAP4's search and fetch methods with responses -# # from RFC 3501 -# imap_emails = { -# "1": ("OK", (("1", msg.as_string()),)), -# "2": ("OK", (("2", msg.as_string()),)), -# } -# imap_mail_list = ("OK", ("1 2",)) -# mocked_imaplib_server = mock.Mock() -# mocked_imaplib_server.search = mock.Mock( -# return_value=imap_mail_list) -# -# # we ignore the second arg as the data item/mime-part is -# # constant (RFC822) -# mocked_imaplib_server.fetch = mock.Mock( -# side_effect=lambda x, _: imap_emails[x]) -# with mock.patch('helpdesk.email.imaplib', autospec=True) as mocked_imaplib: -# mocked_imaplib.IMAP4 = mock.Mock( -# return_value=mocked_imaplib_server) -# call_command('get_email') -# -# ticket1 = get_object_or_404(Ticket, pk=1) -# self.assertEqual(ticket1.ticket_for_url, "QQ-%s" % ticket1.id) -# self.assertEqual(ticket1.title, subject) -# # plain text should become description -# self.assertEqual(ticket1.description, text) -# # HTML MIME part should be attached to follow up -# followup1 = get_object_or_404(FollowUp, pk=1) -# self.assertEqual(followup1.ticket.id, 1) -# attach1 = get_object_or_404(FollowUpAttachment, pk=1) -# self.assertEqual(attach1.followup.id, 1) -# self.assertEqual(attach1.filename, 'email_html_body.html') -# cc0 = get_object_or_404(TicketCC, pk=1) -# self.assertEqual(cc0.email, you) -# cc1 = get_object_or_404(TicketCC, pk=2) -# self.assertEqual(cc1.email, cc_one) -# cc2 = get_object_or_404(TicketCC, pk=3) -# self.assertEqual(cc2.email, cc_two) -# self.assertEqual(len(TicketCC.objects.filter(ticket=1)), 3) -# -# ticket2 = get_object_or_404(Ticket, pk=2) -# self.assertEqual(ticket2.ticket_for_url, "QQ-%s" % ticket2.id) -# self.assertEqual(ticket2.title, subject) -# # plain text should become description -# self.assertEqual(ticket2.description, text) -# # HTML MIME part should be attached to follow up -# followup2 = get_object_or_404(FollowUp, pk=2) -# self.assertEqual(followup2.ticket.id, 2) -# attach2 = get_object_or_404(FollowUpAttachment, pk=2) -# self.assertEqual(attach2.followup.id, 2) -# self.assertEqual(attach2.filename, 'email_html_body.html') -# -# def test_read_pgp_signed_email(self): -# """Tests reading a PGP signed email to ensure we handle base64 -# and PGP signatures appropriately.""" -# -# # example email text from #567 on GitHub -# with open(os.path.join(THIS_DIR, "test_files/pgp.eml"), encoding="utf-8") as fd: -# test_email = fd.read() -# test_mail_len = len(test_email) -# -# if self.socks: -# from socks import ProxyConnectionError -# with self.assertRaisesRegex(ProxyConnectionError, '%s:%s' % (unrouted_socks_server, unused_port)): -# call_command('get_email') -# -# else: -# # Test local email reading -# if self.method == 'local': -# with mock.patch('os.listdir') as mocked_listdir, \ -# mock.patch('helpdesk.email.isfile') as mocked_isfile, \ -# mock.patch('builtins.open', mock.mock_open(read_data=test_email)), \ -# mock.patch('os.unlink'): -# mocked_isfile.return_value = True -# mocked_listdir.return_value = ['filename1'] -# -# call_command('get_email') -# -# mocked_listdir.assert_called_with( -# '/var/lib/mail/helpdesk/') -# mocked_isfile.assert_any_call( -# '/var/lib/mail/helpdesk/filename1') -# -# elif self.method == 'pop3': -# # mock poplib.POP3's list and retr methods to provide responses -# # as per RFC 1939 -# pop3_emails = { -# '1': ("+OK", test_email.split('\n')), -# } -# pop3_mail_list = ("+OK 1 message", ("1 %d" % test_mail_len)) -# mocked_poplib_server = mock.Mock() -# mocked_poplib_server.list = mock.Mock( -# return_value=pop3_mail_list) -# mocked_poplib_server.retr = mock.Mock( -# side_effect=lambda x: pop3_emails['1']) -# with mock.patch('helpdesk.email.poplib', autospec=True) as mocked_poplib: -# mocked_poplib.POP3 = mock.Mock( -# return_value=mocked_poplib_server) -# call_command('get_email') -# -# -# elif self.method in ['imap', 'oauth']: -# # mock imaplib.IMAP4's search and fetch methods with responses -# # from RFC 3501 -# imap_emails = { -# "1": ("OK", (("1", test_email),)), -# } -# imap_mail_list = ("OK", ("1",)) -# mocked_imaplib_server = mock.Mock() -# mocked_imaplib_server.search = mock.Mock( -# return_value=imap_mail_list) -# -# # we ignore the second arg as the data item/mime-part is -# # constant (RFC822) -# mocked_imaplib_server.fetch = mock.Mock( -# side_effect=lambda x, _: imap_emails[x]) -# with mock.patch('helpdesk.email.imaplib', autospec=True) as mocked_imaplib: -# mocked_imaplib.IMAP4 = mock.Mock( -# return_value=mocked_imaplib_server) -# call_command('get_email') -# -# ticket1 = get_object_or_404(Ticket, pk=1) -# self.assertEqual(ticket1.ticket_for_url, "QQ-%s" % ticket1.id) -# self.assertEqual( -# ticket1.title, "example email that crashes django-helpdesk get_email") -# self.assertEqual( -# ticket1.description, """hi, thanks for looking into this :)\n\nhttps://github.com/django-helpdesk/django-helpdesk/issues/567#issuecomment-342954233""") -# # MIME part should be attached to follow up -# followup1 = get_object_or_404(FollowUp, pk=1) -# self.assertEqual(followup1.ticket.id, 1) -# attach1 = get_object_or_404(FollowUpAttachment, pk=1) -# self.assertEqual(attach1.followup.id, 1) -# self.assertEqual(attach1.filename, 'part-1_signature.asc') -# self.assertEqual(attach1.file.read(), b"""-----BEGIN PGP SIGNATURE----- -# -# iQIcBAEBCAAGBQJaA3dnAAoJELBLc7QPITnLN54P/3Zsu7+AIQWDFTvziJfCqswG -# u99fG+iWa6ER+iuZG0YU1BdIxIjSKt1pvqB0yXITlT9FCdf1zc0pmeJ08I0a5pVa -# iaym5prVUro5BNQ6Vqoo0jvOCKNrACtFNv85zDzXbPNP8TrUss41U+ackPHkOHov -# cmJ5YZFQebYXXpibFSIDimVGfwI57vyTWvolttZFLSI1mgGX7MvHaKh253QLdXIo -# EUih40rOw3f/nYPEKyW8QA72ImBsZdcZI5buiiCC1bgMkKSFSNAFiIanYEpGNMnO -# 3zYKBpbpBhnWSi5orwx47/v4/Yb/qVr5ppuV23+YoMfEGT8cHPTAdYpnpE27ByAv -# jvpxKEwmkUzD1WxOmQdCcPJPyWz1OBUVvjj0nn0Espnz8V8esl9+IFs739lpFBHu -# fWWA315LTmIJMGH5Ujf4myiQeXDo6Gsy6WhE13q7MKTq3tnyi5dJG9GJCBf646dL -# RwcDf9O7MvKSV2kSPmryLnUF7D+2fva+Cy+CvJDVJCo5zr4ucXPXZ4htpI6Pjpd5 -# oPHvbqxSCMJrQ7eAFTYmBNGauSyr0XvGM1qmHBZD/laQEJHYgLT2ILrymZhVDHtK -# W7tXhGjMoUvqAxiKkmG3UHFqN4k3EYo13PwoOWyJHD1M9ArbX/Sk9l8DDguCh3DW -# a9eiiQ+3V1v+7wWHXCzq -# =6JeP -# -----END PGP SIGNATURE----- -# """) -# # should this be 'application/pgp-signature'? -# # self.assertEqual(attach1.mime_type, 'text/plain') -# + + def test_commas_in_mail_headers(self): + """Tests correctly decoding mail headers when a comma is encoded into + UTF-8. See bug report #832.""" + + # Create the from using standard RFC required formats + # Override the last_name to ensure we get a non-ascii character in it + test_email_from_meta = utils.generate_email_address("fr_FR", last_name_override="Bouissières") + test_email_subject = "Commas in From lines" + test_email_body = "Testing commas in from email UTF-8." + test_email = "To: helpdesk@example.com\nFrom: " + test_email_from_meta[0] + \ + "\nSubject: " + test_email_subject + "\n\n" + test_email_body + test_mail_len = len(test_email) + + if self.socks: + from socks import ProxyConnectionError + with self.assertRaisesRegex(ProxyConnectionError, '%s:%s' % (unrouted_socks_server, unused_port)): + call_command('get_email') + + else: + # Test local email reading + if self.method == 'local': + with mock.patch('os.listdir') as mocked_listdir, \ + mock.patch('helpdesk.email.isfile') as mocked_isfile, \ + mock.patch('builtins.open', mock.mock_open(read_data=test_email)), \ + mock.patch('os.unlink'): + mocked_isfile.return_value = True + mocked_listdir.return_value = ['filename1', 'filename2'] + + call_command('get_email') + + mocked_listdir.assert_called_with( + '/var/lib/mail/helpdesk/') + mocked_isfile.assert_any_call( + '/var/lib/mail/helpdesk/filename1') + mocked_isfile.assert_any_call( + '/var/lib/mail/helpdesk/filename2') + + elif self.method == 'pop3': + # mock poplib.POP3's list and retr methods to provide responses + # as per RFC 1939 + pop3_emails = { + '1': ("+OK", test_email.split('\n')), + '2': ("+OK", test_email.split('\n')), + } + pop3_mail_list = ("+OK 2 messages", ("1 %d" % + test_mail_len, "2 %d" % test_mail_len)) + mocked_poplib_server = mock.Mock() + mocked_poplib_server.list = mock.Mock( + return_value=pop3_mail_list) + mocked_poplib_server.retr = mock.Mock( + side_effect=lambda x: pop3_emails[x]) + with mock.patch('helpdesk.email.poplib', autospec=True) as mocked_poplib: + mocked_poplib.POP3 = mock.Mock( + return_value=mocked_poplib_server) + call_command('get_email') + + elif self.method == 'imap': + # mock imaplib.IMAP4's search and fetch methods with responses + # from RFC 3501 + imap_emails = { + "1": ("OK", (("1", test_email),)), + "2": ("OK", (("2", test_email),)), + } + imap_mail_list = ("OK", ("1 2",)) + mocked_imaplib_server = mock.Mock() + mocked_imaplib_server.search = mock.Mock( + return_value=imap_mail_list) + + # we ignore the second arg as the data item/mime-part is + # constant (RFC822) + mocked_imaplib_server.fetch = mock.Mock( + side_effect=lambda x, _: imap_emails[x]) + with mock.patch('helpdesk.email.imaplib', autospec=True) as mocked_imaplib: + mocked_imaplib.IMAP4 = mock.Mock( + return_value=mocked_imaplib_server) + call_command('get_email') + + elif self.method == 'oauth': + # mock the oauthlib session and requests oauth backendclient + # then mock imaplib.IMAP4's search and fetch methods with responses + # from RFC 3501 + imap_emails = { + "1": ("OK", (("1", test_email),)), + "2": ("OK", (("2", test_email),)), + } + imap_mail_list = ("OK", ("1 2",)) + mocked_imaplib_server = mock.Mock() + mocked_imaplib_server.search = mock.Mock( + return_value=imap_mail_list) + + # we ignore the second arg as the data item/mime-part is + # constant (RFC822) + mocked_imaplib_server.fetch = mock.Mock( + side_effect=lambda x, _: imap_emails[x]) + + mocked_oauth_backend_client = mock.Mock() + with mock.patch('helpdesk.email.oauth2lib', autospec=True) as mocked_oauth2lib: + mocked_oauth2lib.BackendApplicationClient = mock.Mock( + return_value=mocked_oauth_backend_client) + + mocked_oauth_session = mock.Mock() + mocked_oauth_session.fetch_token = mock.Mock( + return_value={} + ) + + with mock.patch('helpdesk.email.requests_oauthlib', autospec=True) as mocked_requests_oauthlib: + mocked_requests_oauthlib.OAuth2Session = mock.Mock( + return_value=mocked_oauth_session) + + with mock.patch('helpdesk.email.imaplib', autospec=True) as mocked_imaplib: + mocked_imaplib.IMAP4 = mock.Mock( + return_value=mocked_imaplib_server) + + call_command('get_email') + + ticket1 = get_object_or_404(Ticket, pk=1) + self.assertEqual(ticket1.ticket_for_url, "QQ-%s" % ticket1.id) + self.assertEqual(ticket1.submitter_email, test_email_from_meta[1]) + self.assertEqual(ticket1.title, test_email_subject) + self.assertEqual(ticket1.description, test_email_body) + + ticket2 = get_object_or_404(Ticket, pk=2) + self.assertEqual(ticket2.ticket_for_url, "QQ-%s" % ticket2.id) + self.assertEqual(ticket2.submitter_email, test_email_from_meta[1]) + self.assertEqual(ticket2.title, test_email_subject) + self.assertEqual(ticket2.description, test_email_body) + + def test_read_email_with_template_tag(self): + """Tests reading plain text emails from a queue and creating tickets, + except this time the email body contains a Django template tag. + For each email source supported, we mock the backend to provide + authentically formatted responses containing our test data.""" + + # example email text from Django docs: + # https://docs.djangoproject.com/en/1.10/ref/unicode/ + test_email_from = "Arnbjörg Ráðormsdóttir " + test_email_subject = "My visit to Sør-Trøndelag" + test_email_body = "Reporting some issue with the template tag: {% if helpdesk %}." + test_email = "To: helpdesk@example.com\nFrom: " + test_email_from + \ + "\nSubject: " + test_email_subject + "\n\n" + test_email_body + test_mail_len = len(test_email) + + if self.socks: + from socks import ProxyConnectionError + with self.assertRaisesRegex(ProxyConnectionError, '%s:%s' % (unrouted_socks_server, unused_port)): + call_command('get_email') + + else: + # Test local email reading + if self.method == 'local': + with mock.patch('os.listdir') as mocked_listdir, \ + mock.patch('helpdesk.email.isfile') as mocked_isfile, \ + mock.patch('builtins.open', mock.mock_open(read_data=test_email)), \ + mock.patch('os.unlink'): + mocked_isfile.return_value = True + mocked_listdir.return_value = ['filename1', 'filename2'] + + call_command('get_email') + + mocked_listdir.assert_called_with( + '/var/lib/mail/helpdesk/') + mocked_isfile.assert_any_call( + '/var/lib/mail/helpdesk/filename1') + mocked_isfile.assert_any_call( + '/var/lib/mail/helpdesk/filename2') + + elif self.method == 'pop3': + # mock poplib.POP3's list and retr methods to provide responses + # as per RFC 1939 + pop3_emails = { + '1': ("+OK", test_email.split('\n')), + '2': ("+OK", test_email.split('\n')), + } + pop3_mail_list = ("+OK 2 messages", ("1 %d" % + test_mail_len, "2 %d" % test_mail_len)) + mocked_poplib_server = mock.Mock() + mocked_poplib_server.list = mock.Mock( + return_value=pop3_mail_list) + mocked_poplib_server.retr = mock.Mock( + side_effect=lambda x: pop3_emails[x]) + with mock.patch('helpdesk.email.poplib', autospec=True) as mocked_poplib: + mocked_poplib.POP3 = mock.Mock( + return_value=mocked_poplib_server) + call_command('get_email') + + elif self.method == 'imap': + # mock imaplib.IMAP4's search and fetch methods with responses + # from RFC 3501 + imap_emails = { + "1": ("OK", (("1", test_email),)), + "2": ("OK", (("2", test_email),)), + } + imap_mail_list = ("OK", ("1 2",)) + mocked_imaplib_server = mock.Mock() + mocked_imaplib_server.search = mock.Mock( + return_value=imap_mail_list) + + # we ignore the second arg as the data item/mime-part is + # constant (RFC822) + mocked_imaplib_server.fetch = mock.Mock( + side_effect=lambda x, _: imap_emails[x]) + with mock.patch('helpdesk.email.imaplib', autospec=True) as mocked_imaplib: + mocked_imaplib.IMAP4 = mock.Mock( + return_value=mocked_imaplib_server) + call_command('get_email') + + elif self.method == 'oauth': + # mock the oauthlib session and requests oauth backendclient + # then mock imaplib.IMAP4's search and fetch methods with responses + # from RFC 3501 + imap_emails = { + "1": ("OK", (("1", test_email),)), + "2": ("OK", (("2", test_email),)), + } + imap_mail_list = ("OK", ("1 2",)) + mocked_imaplib_server = mock.Mock() + mocked_imaplib_server.search = mock.Mock( + return_value=imap_mail_list) + + # we ignore the second arg as the data item/mime-part is + # constant (RFC822) + mocked_imaplib_server.fetch = mock.Mock( + side_effect=lambda x, _: imap_emails[x]) + + mocked_oauth_backend_client = mock.Mock() + with mock.patch('helpdesk.email.oauth2lib', autospec=True) as mocked_oauth2lib: + mocked_oauth2lib.BackendApplicationClient = mock.Mock( + return_value=mocked_oauth_backend_client) + + mocked_oauth_session = mock.Mock() + mocked_oauth_session.fetch_token = mock.Mock( + return_value={} + ) + + with mock.patch('helpdesk.email.requests_oauthlib', autospec=True) as mocked_requests_oauthlib: + mocked_requests_oauthlib.OAuth2Session = mock.Mock( + return_value=mocked_oauth_session) + + with mock.patch('helpdesk.email.imaplib', autospec=True) as mocked_imaplib: + mocked_imaplib.IMAP4 = mock.Mock( + return_value=mocked_imaplib_server) + + call_command('get_email') + + ticket1 = get_object_or_404(Ticket, pk=1) + self.assertEqual(ticket1.ticket_for_url, "QQ-%s" % ticket1.id) + self.assertEqual(ticket1.title, test_email_subject) + self.assertEqual(ticket1.description, test_email_body) + + ticket2 = get_object_or_404(Ticket, pk=2) + self.assertEqual(ticket2.ticket_for_url, "QQ-%s" % ticket2.id) + self.assertEqual(ticket2.title, test_email_subject) + self.assertEqual(ticket2.description, test_email_body) + + def test_read_html_multipart_email(self): + """Tests reading multipart MIME (HTML body and plain text alternative) + emails from a queue and creating tickets. + For each email source supported, we mock the backend to provide + authentically formatted responses containing our test data.""" + + # example email text from Python docs: + # https://docs.python.org/3/library/email-examples.html + from email.mime.multipart import MIMEMultipart + from email.mime.text import MIMEText + + me = "my@example.com" + you = "your@example.com" + # NOTE: CC'd emails need to be alphabetical and tested as such! + # implementation uses sets, so only way to ensure tickets created + # in right order is to change set to list and sort it + cc_one = "nobody@example.com" + cc_two = "other@example.com" + cc = cc_one + ", " + cc_two + subject = "Link" + + # Create message container - the correct MIME type is + # multipart/alternative. + msg = MIMEMultipart('alternative') + msg['Subject'] = subject + msg['From'] = me + msg['To'] = you + msg['Cc'] = cc + + # Create the body of the message (a plain-text and an HTML version). + text = "Hi!\nHow are you?\nHere is the link you wanted:\nhttps://www.python.org" + html = """\ + + + +

Hi!
+ How are you?
+ Here is the link you wanted. +

+ + + """ + + # Record the MIME types of both parts - text/plain and text/html. + part1 = MIMEText(text, 'plain') + part2 = MIMEText(html, 'html') + + # Attach parts into message container. + # According to RFC 2046, the last part of a multipart message, in this case + # the HTML message, is best and preferred. + msg.attach(part1) + msg.attach(part2) + + test_mail_len = len(msg) + + if self.socks: + from socks import ProxyConnectionError + with self.assertRaisesRegex(ProxyConnectionError, '%s:%s' % (unrouted_socks_server, unused_port)): + call_command('get_email') + + else: + # Test local email reading + if self.method == 'local': + with mock.patch('os.listdir') as mocked_listdir, \ + mock.patch('helpdesk.email.isfile') as mocked_isfile, \ + mock.patch('builtins.open', mock.mock_open(read_data=msg.as_string())), \ + mock.patch('os.unlink'): + mocked_isfile.return_value = True + mocked_listdir.return_value = ['filename1', 'filename2'] + + call_command('get_email') + + mocked_listdir.assert_called_with( + '/var/lib/mail/helpdesk/') + mocked_isfile.assert_any_call( + '/var/lib/mail/helpdesk/filename1') + mocked_isfile.assert_any_call( + '/var/lib/mail/helpdesk/filename2') + + elif self.method == 'pop3': + # mock poplib.POP3's list and retr methods to provide responses + # as per RFC 1939 + pop3_emails = { + '1': ("+OK", msg.as_string().split('\n')), + '2': ("+OK", msg.as_string().split('\n')), + } + pop3_mail_list = ("+OK 2 messages", ("1 %d" % + test_mail_len, "2 %d" % test_mail_len)) + mocked_poplib_server = mock.Mock() + mocked_poplib_server.list = mock.Mock( + return_value=pop3_mail_list) + mocked_poplib_server.retr = mock.Mock( + side_effect=lambda x: pop3_emails[x]) + with mock.patch('helpdesk.email.poplib', autospec=True) as mocked_poplib: + mocked_poplib.POP3 = mock.Mock( + return_value=mocked_poplib_server) + call_command('get_email') + + + elif self.method == 'imap': + # mock imaplib.IMAP4's search and fetch methods with responses + # from RFC 3501 + imap_emails = { + "1": ("OK", (("1", msg.as_string()),)), + "2": ("OK", (("2", msg.as_string()),)), + } + imap_mail_list = ("OK", ("1 2",)) + mocked_imaplib_server = mock.Mock() + mocked_imaplib_server.search = mock.Mock( + return_value=imap_mail_list) + + # we ignore the second arg as the data item/mime-part is + # constant (RFC822) + mocked_imaplib_server.fetch = mock.Mock( + side_effect=lambda x, _: imap_emails[x]) + with mock.patch('helpdesk.email.imaplib', autospec=True) as mocked_imaplib: + mocked_imaplib.IMAP4 = mock.Mock( + return_value=mocked_imaplib_server) + call_command('get_email') + + elif self.method == 'oauth': + # mock the oauthlib session and requests oauth backendclient + # then mock imaplib.IMAP4's search and fetch methods with responses + # from RFC 3501 + imap_emails = { + "1": ("OK", (("1", msg.as_string()),)), + "2": ("OK", (("2", msg.as_string()),)), + } + imap_mail_list = ("OK", ("1 2",)) + mocked_imaplib_server = mock.Mock() + mocked_imaplib_server.search = mock.Mock( + return_value=imap_mail_list) + + # we ignore the second arg as the data item/mime-part is + # constant (RFC822) + mocked_imaplib_server.fetch = mock.Mock( + side_effect=lambda x, _: imap_emails[x]) + + mocked_oauth_backend_client = mock.Mock() + with mock.patch('helpdesk.email.oauth2lib', autospec=True) as mocked_oauth2lib: + mocked_oauth2lib.BackendApplicationClient = mock.Mock( + return_value=mocked_oauth_backend_client) + + mocked_oauth_session = mock.Mock() + mocked_oauth_session.fetch_token = mock.Mock( + return_value={} + ) + + with mock.patch('helpdesk.email.requests_oauthlib', autospec=True) as mocked_requests_oauthlib: + mocked_requests_oauthlib.OAuth2Session = mock.Mock( + return_value=mocked_oauth_session) + + with mock.patch('helpdesk.email.imaplib', autospec=True) as mocked_imaplib: + mocked_imaplib.IMAP4 = mock.Mock( + return_value=mocked_imaplib_server) + + call_command('get_email') + + ticket1 = get_object_or_404(Ticket, pk=1) + self.assertEqual(ticket1.ticket_for_url, "QQ-%s" % ticket1.id) + self.assertEqual(ticket1.title, subject) + # plain text should become description + self.assertEqual(ticket1.description, text) + # HTML MIME part should be attached to follow up + followup1 = get_object_or_404(FollowUp, pk=1) + self.assertEqual(followup1.ticket.id, 1) + attach1 = get_object_or_404(FollowUpAttachment, pk=1) + self.assertEqual(attach1.followup.id, 1) + self.assertEqual(attach1.filename, 'email_html_body.html') + cc0 = get_object_or_404(TicketCC, pk=1) + self.assertEqual(cc0.email, you) + cc1 = get_object_or_404(TicketCC, pk=2) + self.assertEqual(cc1.email, cc_one) + cc2 = get_object_or_404(TicketCC, pk=3) + self.assertEqual(cc2.email, cc_two) + self.assertEqual(len(TicketCC.objects.filter(ticket=1)), 3) + + ticket2 = get_object_or_404(Ticket, pk=2) + self.assertEqual(ticket2.ticket_for_url, "QQ-%s" % ticket2.id) + self.assertEqual(ticket2.title, subject) + # plain text should become description + self.assertEqual(ticket2.description, text) + # HTML MIME part should be attached to follow up + followup2 = get_object_or_404(FollowUp, pk=2) + self.assertEqual(followup2.ticket.id, 2) + attach2 = get_object_or_404(FollowUpAttachment, pk=2) + self.assertEqual(attach2.followup.id, 2) + self.assertEqual(attach2.filename, 'email_html_body.html') + + def test_read_pgp_signed_email(self): + """Tests reading a PGP signed email to ensure we handle base64 + and PGP signatures appropriately.""" + + # example email text from #567 on GitHub + with open(os.path.join(THIS_DIR, "test_files/pgp.eml"), encoding="utf-8") as fd: + test_email = fd.read() + test_mail_len = len(test_email) + + if self.socks: + from socks import ProxyConnectionError + with self.assertRaisesRegex(ProxyConnectionError, '%s:%s' % (unrouted_socks_server, unused_port)): + call_command('get_email') + + else: + # Test local email reading + if self.method == 'local': + with mock.patch('os.listdir') as mocked_listdir, \ + mock.patch('helpdesk.email.isfile') as mocked_isfile, \ + mock.patch('builtins.open', mock.mock_open(read_data=test_email)), \ + mock.patch('os.unlink'): + mocked_isfile.return_value = True + mocked_listdir.return_value = ['filename1'] + + call_command('get_email') + + mocked_listdir.assert_called_with( + '/var/lib/mail/helpdesk/') + mocked_isfile.assert_any_call( + '/var/lib/mail/helpdesk/filename1') + + elif self.method == 'pop3': + # mock poplib.POP3's list and retr methods to provide responses + # as per RFC 1939 + pop3_emails = { + '1': ("+OK", test_email.split('\n')), + } + pop3_mail_list = ("+OK 1 message", ("1 %d" % test_mail_len)) + mocked_poplib_server = mock.Mock() + mocked_poplib_server.list = mock.Mock( + return_value=pop3_mail_list) + mocked_poplib_server.retr = mock.Mock( + side_effect=lambda x: pop3_emails['1']) + with mock.patch('helpdesk.email.poplib', autospec=True) as mocked_poplib: + mocked_poplib.POP3 = mock.Mock( + return_value=mocked_poplib_server) + call_command('get_email') + + + elif self.method == 'imap': + # mock imaplib.IMAP4's search and fetch methods with responses + # from RFC 3501 + imap_emails = { + "1": ("OK", (("1", test_email),)), + } + imap_mail_list = ("OK", ("1",)) + mocked_imaplib_server = mock.Mock() + mocked_imaplib_server.search = mock.Mock( + return_value=imap_mail_list) + + # we ignore the second arg as the data item/mime-part is + # constant (RFC822) + mocked_imaplib_server.fetch = mock.Mock( + side_effect=lambda x, _: imap_emails[x]) + with mock.patch('helpdesk.email.imaplib', autospec=True) as mocked_imaplib: + mocked_imaplib.IMAP4 = mock.Mock( + return_value=mocked_imaplib_server) + call_command('get_email') + + elif self.method == 'oauth': + # mock the oauthlib session and requests oauth backendclient + # then mock imaplib.IMAP4's search and fetch methods with responses + # from RFC 3501 + imap_emails = { + "1": ("OK", (("1", test_email),)), + } + imap_mail_list = ("OK", ("1",)) + mocked_imaplib_server = mock.Mock() + mocked_imaplib_server.search = mock.Mock( + return_value=imap_mail_list) + + # we ignore the second arg as the data item/mime-part is + # constant (RFC822) + mocked_imaplib_server.fetch = mock.Mock( + side_effect=lambda x, _: imap_emails[x]) + + mocked_oauth_backend_client = mock.Mock() + with mock.patch('helpdesk.email.oauth2lib', autospec=True) as mocked_oauth2lib: + mocked_oauth2lib.BackendApplicationClient = mock.Mock( + return_value=mocked_oauth_backend_client) + + mocked_oauth_session = mock.Mock() + mocked_oauth_session.fetch_token = mock.Mock( + return_value={} + ) + + with mock.patch('helpdesk.email.requests_oauthlib', autospec=True) as mocked_requests_oauthlib: + mocked_requests_oauthlib.OAuth2Session = mock.Mock( + return_value=mocked_oauth_session) + + with mock.patch('helpdesk.email.imaplib', autospec=True) as mocked_imaplib: + mocked_imaplib.IMAP4 = mock.Mock( + return_value=mocked_imaplib_server) + + call_command('get_email') + + ticket1 = get_object_or_404(Ticket, pk=1) + self.assertEqual(ticket1.ticket_for_url, "QQ-%s" % ticket1.id) + self.assertEqual( + ticket1.title, "example email that crashes django-helpdesk get_email") + self.assertEqual( + ticket1.description, """hi, thanks for looking into this :)\n\nhttps://github.com/django-helpdesk/django-helpdesk/issues/567#issuecomment-342954233""") + # MIME part should be attached to follow up + followup1 = get_object_or_404(FollowUp, pk=1) + self.assertEqual(followup1.ticket.id, 1) + attach1 = get_object_or_404(FollowUpAttachment, pk=1) + self.assertEqual(attach1.followup.id, 1) + self.assertEqual(attach1.filename, 'part-1_signature.asc') + self.assertEqual(attach1.file.read(), b"""-----BEGIN PGP SIGNATURE----- + +iQIcBAEBCAAGBQJaA3dnAAoJELBLc7QPITnLN54P/3Zsu7+AIQWDFTvziJfCqswG +u99fG+iWa6ER+iuZG0YU1BdIxIjSKt1pvqB0yXITlT9FCdf1zc0pmeJ08I0a5pVa +iaym5prVUro5BNQ6Vqoo0jvOCKNrACtFNv85zDzXbPNP8TrUss41U+ackPHkOHov +cmJ5YZFQebYXXpibFSIDimVGfwI57vyTWvolttZFLSI1mgGX7MvHaKh253QLdXIo +EUih40rOw3f/nYPEKyW8QA72ImBsZdcZI5buiiCC1bgMkKSFSNAFiIanYEpGNMnO +3zYKBpbpBhnWSi5orwx47/v4/Yb/qVr5ppuV23+YoMfEGT8cHPTAdYpnpE27ByAv +jvpxKEwmkUzD1WxOmQdCcPJPyWz1OBUVvjj0nn0Espnz8V8esl9+IFs739lpFBHu +fWWA315LTmIJMGH5Ujf4myiQeXDo6Gsy6WhE13q7MKTq3tnyi5dJG9GJCBf646dL +RwcDf9O7MvKSV2kSPmryLnUF7D+2fva+Cy+CvJDVJCo5zr4ucXPXZ4htpI6Pjpd5 +oPHvbqxSCMJrQ7eAFTYmBNGauSyr0XvGM1qmHBZD/laQEJHYgLT2ILrymZhVDHtK +W7tXhGjMoUvqAxiKkmG3UHFqN4k3EYo13PwoOWyJHD1M9ArbX/Sk9l8DDguCh3DW +a9eiiQ+3V1v+7wWHXCzq +=6JeP +-----END PGP SIGNATURE----- +""") + # should this be 'application/pgp-signature'? + # self.assertEqual(attach1.mime_type, 'text/plain') + + class GetEmailCCHandling(TestCase): """TestCase that checks CC handling in email. Needs its own test harness.""" diff --git a/quicktest.py b/quicktest.py index e387be6e..a710a41f 100755 --- a/quicktest.py +++ b/quicktest.py @@ -109,7 +109,9 @@ class QuickDjangoTest: HELPDESK_TEAMS_MIGRATION_DEPENDENCIES=[], HELPDESK_KBITEM_TEAM_GETTER=lambda _: None, # test the API - HELPDESK_ACTIVATE_API_ENDPOINT=True + HELPDESK_ACTIVATE_API_ENDPOINT=True, + # Set IMAP Server Debug Verbosity + HELPDESK_IMAP_DEBUG_LEVEL=int(os.environ.get("HELPDESK_IMAP_DEBUG_LEVEL", "0")), ) from django.test.runner import DiscoverRunner